%-*-Mode:erlang;coding:utf-8;tab-width:4;c-basic-offset:4;indent-tabs-mode:()-*- % ex: set ft=erlang fenc=utf-8 sts=4 ts=4 sw=4 et: %%% %%%------------------------------------------------------------------------ %%% @doc %%% ==CloudI Cassandra Data Module== %%% %%% This ia a _very_ thin shim on top of erlang_cassandra from %%% https://github.com/dieswaytoofast/erlang_cassandra` %%% Note that your results are "double ok"'d. i.e., whatever %%% erlang_cassandra sends back is sent back from send_sync. %%% As a result, all successful responses will be of the form %%% {ok, {ok, Response}} %%% All failures from erlang_cassandra will be of the form %%% {ok, {error, Reason}} %%% send_sync errors will be standard 'cloudi' errors %%% @end %%% %%% BSD LICENSE %%% %%% Copyright (c) 2013, Mahesh Paolini-Subramanya %%% All rights reserved. %%% %%% Redistribution and use in source and binary forms, with or without %%% modification, are permitted provided that the following conditions are met: %%% %%% * Redistributions of source code must retain the above copyright %%% notice, this list of conditions and the following disclaimer. %%% * Redistributions in binary form must reproduce the above copyright %%% notice, this list of conditions and the following disclaimer in %%% the documentation and/or other materials provided with the %%% distribution. %%% * All advertising materials mentioning features or use of this %%% software must display the following acknowledgment: %%% This product includes %%% software developed by Mahesh Paolini-Subramanya %%% * The name of the author may not be used to endorse or promote %%% products derived from this software without specific prior %%% written permission %%% %%% THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND %%% CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, %%% INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES %%% OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE %%% DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR %%% CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, %%% SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, %%% BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR %%% SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS %%% INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, %%% WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING %%% NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE %%% OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH %%% DAMAGE. %%% %%% @author Mahesh Paolini-Subramanya %%% @copyright 2013 Mahesh Paolini-Subramanya %%% @version 1.3.0 {@date} {@time} %%%------------------------------------------------------------------------ -module(cloudi_service_db_cassandra). -author('Mahesh Paolini-Subramanya '). -behaviour(cloudi_service). %% external interface %% cassandra API -export([set_keyspace/3]). -export([describe_keyspace/3]). -export([system_add_keyspace/3]). -export([system_update_keyspace/3]). -export([system_drop_keyspace/3]). -export([insert/7]). -export([get/6]). -export([remove/7]). -export([system_add_column_family/3]). -export([system_describe_column_family/4]). -export([system_drop_column_family/4]). -export([system_update_column_family/3]). -export([truncate/4]). -export([add/7]). -export([remove_counter/6]). -export([get_slice/7]). -export([multiget_slice/7]). -export([get_count/7]). -export([multiget_count/7]). -export([get_range_slices/7]). -export([get_indexed_slices/7]). -export([execute_cql_query/4]). -export([prepare_cql_query/4]). -export([execute_prepared_cql_query/4]). -export([describe_version/2]). -export([describe_snitch/2]). -export([describe_partitioner/2]). -export([describe_schema_versions/2]). -export([describe_keyspaces/2]). -export([describe_cluster_name/2]). -export([describe_ring/3]). %% cloudi_service callbacks -export([cloudi_service_init/3, cloudi_service_handle_request/11, cloudi_service_handle_info/3, cloudi_service_terminate/2]). -include_lib("cloudi_core/include/cloudi_logger.hrl"). -include_lib("erlang_cassandra/include/erlang_cassandra_types.hrl"). -define(DEFAULT_DATABASE, undefined). -define(DEFAULT_THRIFT_HOST, "localhost"). -define(DEFAULT_THRIFT_PORT, 9160). -define(DEFAULT_CONNECTION_NAME, <<"cloudi_default_connection_name">>). -define(DEFAULT_POOL_OPTIONS, [{size, 5}, {max_overflow, 10} ]). -define(DEFAULT_CONNECTION_OPTIONS, [{thrift_host, ?DEFAULT_THRIFT_HOST}, {thrift_port, ?DEFAULT_THRIFT_PORT} ]). -type error() :: {error, Reason :: term()}. -type connection_name() :: {any(), any(), string() | binary()} | string() | binary(). -type dispatcher() :: cloudi_service:dispatcher() | cloudi:context(). -type name() :: cloudi_service:service_name(). -type thrift_host() :: undefined | string(). -type thrift_port() :: undefined | integer(). -type client_name() :: binary(). -type server_ref() :: atom() | pid() | client_name(). -type fq_server_ref() :: {thrift_host(), thrift_port(), server_ref()}. -type destination() :: server_ref() | fq_server_ref(). -type keyspace_definition() :: #ksDef{}. -type row_key() :: binary(). -type key_range() :: #keyRange{}. -type column_family() :: binary(). -type column_timestamp() :: non_neg_integer(). -type consistency_level() :: non_neg_integer(). -type column() :: #column{}. -type column_parent() :: #columnParent{}. -type column_path() :: #columnPath{}. -type column_family_definition() :: #cfDef{}. -type counter_column() :: #counterColumn{}. -type slice_predicate() :: #slicePredicate{}. -type cql_query() :: binary(). -type cql_query_id() :: integer(). -type compression() :: binary(). -record(state, { connection_name :: connection_name() }). -type response() :: [tuple()] | error(). %%%------------------------------------------------------------------------ %%% External interface functions %%%------------------------------------------------------------------------ %% @doc Set the keyspace to be used by the connection -spec set_keyspace(dispatcher(), name(), destination()) -> response(). set_keyspace(Dispatcher, Name, Destination) when is_list(Name) -> cloudi:send_sync(Dispatcher, Name, {set_keyspace, Destination}). %% @doc Describe the keyspace used by the connection -spec describe_keyspace(dispatcher(), name(), destination()) -> response(). describe_keyspace(Dispatcher, Name, Destination) when is_list(Name) -> cloudi:send_sync(Dispatcher, Name, {describe_keyspace, Destination}). %% @doc Add a keyspace -spec system_add_keyspace(dispatcher(), name(), keyspace_definition()) -> response(). system_add_keyspace(Dispatcher, Name, KeyspaceDefinition) when is_list(Name) -> cloudi:send_sync(Dispatcher, Name, {system_add_keyspace, KeyspaceDefinition}). %% @doc Update a keyspace -spec system_update_keyspace(dispatcher(), name(), keyspace_definition()) -> response(). system_update_keyspace(Dispatcher, Name, KeyspaceDefinition) when is_list(Name) -> cloudi:send_sync(Dispatcher, Name, {system_update_keyspace, KeyspaceDefinition}). %% @doc Remove a keyspace -spec system_drop_keyspace(dispatcher(), name(), destination()) -> response(). system_drop_keyspace(Dispatcher, Name, Destination) when is_list(Name) -> cloudi:send_sync(Dispatcher, Name, {system_drop_keyspace, Destination}). %% @doc Insert a column -spec insert(dispatcher(), name(), destination(), row_key(), column_parent(), column(), consistency_level()) -> response(). insert(Dispatcher, Name, Destination, RowKey, ColumnParent, Column, ConsistencyLevel) -> cloudi:send_sync(Dispatcher, Name, {insert, Destination, RowKey, ColumnParent, Column, ConsistencyLevel}). %% @doc Get a column -spec get(dispatcher(), name(), destination(), row_key(), column_path(), consistency_level()) -> response(). get(Dispatcher, Name, Destination, RowKey, ColumnPath, ConsistencyLevel) -> cloudi:send_sync(Dispatcher, Name, {get, Destination, RowKey, ColumnPath, ConsistencyLevel}). %% @doc Remove data from the row specified by key at the granularity %% specified by column_path, and the given timestamp -spec remove(dispatcher(), name(), destination(), row_key(), column_path(), column_timestamp(), consistency_level()) -> response(). remove(Dispatcher, Name, Destination, RowKey, ColumnPath, ColumnTimestamp, ConsistencyLevel) -> cloudi:send_sync(Dispatcher, Name, {remove, Destination, RowKey, ColumnPath, ColumnTimestamp, ConsistencyLevel}). %% @doc Add a column family -spec system_add_column_family(dispatcher(), name(), column_family_definition()) -> response(). system_add_column_family(Dispatcher, Name, ColumnFamilyDefinition) -> cloudi:send_sync(Dispatcher, Name, {system_add_column_family, ColumnFamilyDefinition}). %% @doc Get the column family definition -spec system_describe_column_family(dispatcher(), name(), destination(), column_family()) -> response(). system_describe_column_family(Dispatcher, Name, Destination, ColumnFamily) -> cloudi:send_sync(Dispatcher, Name, {system_describe_column_family, Destination, ColumnFamily}). %% @doc Drop a column family -spec system_drop_column_family(dispatcher(), name(), destination(), column_family()) -> response(). system_drop_column_family(Dispatcher, Name, Destination, ColumnFamily) -> cloudi:send_sync(Dispatcher, Name, {system_drop_column_family, Destination, ColumnFamily}). %% @doc Update a column family -spec system_update_column_family(dispatcher(), name(), column_family_definition()) -> response(). system_update_column_family(Dispatcher, Name, ColumnFamilyDefinition) -> cloudi:send_sync(Dispatcher, Name, {system_update_column_family, ColumnFamilyDefinition}). %% @doc Remove all rows from a column family -spec truncate(dispatcher(), name(), destination(), column_family()) -> response(). truncate(Dispatcher, Name, Destination, ColumnFamily) -> cloudi:send_sync(Dispatcher, Name, {truncate, Destination, ColumnFamily}). %% @doc Increment a counter column -spec add(dispatcher(), name(), destination(), row_key(), column_parent(), counter_column(), consistency_level()) -> response(). add(Dispatcher, Name, Destination, RowKey, ColumnParent, CounterColumn, ConsistencyLevel) -> cloudi:send_sync(Dispatcher, Name, {add, Destination, RowKey, ColumnParent, CounterColumn, ConsistencyLevel}). %% @doc Remove a counter -spec remove_counter(dispatcher(), name(), destination(), row_key(), column_path(), consistency_level()) -> response(). remove_counter(Dispatcher, Name, Destination, RowKey, ColumnPath, ConsistencyLevel) -> cloudi:send_sync(Dispatcher, Name, {remove_counter, Destination, RowKey, ColumnPath, ConsistencyLevel}). %% @doc Get a group of columns based on a slice -spec get_slice(dispatcher(), name(), destination(), row_key(), column_parent(), slice_predicate(), consistency_level()) -> response(). get_slice(Dispatcher, Name, Destination, RowKey, ColumnParent, SlicePredicate, ConsistencyLevel) -> cloudi:send_sync(Dispatcher, Name, {get_slice, Destination, RowKey, ColumnParent, SlicePredicate, ConsistencyLevel}). %% @doc Get a group of columns based on a slice and a list of rows -spec multiget_slice(dispatcher(), name(), destination(), [row_key()], column_parent(), slice_predicate(), consistency_level()) -> response(). multiget_slice(Dispatcher, Name, Destination, RowKeys, ColumnParent, SlicePredicate, ConsistencyLevel) when is_list(RowKeys) -> cloudi:send_sync(Dispatcher, Name, {multiget_slice, Destination, RowKeys, ColumnParent, SlicePredicate, ConsistencyLevel}). %% @doc Count columns based on a slice %% WARNING: NOT O(1) -spec get_count(dispatcher(), name(), destination(), row_key(), column_parent(), slice_predicate(), consistency_level()) -> response(). get_count(Dispatcher, Name, Destination, RowKey, ColumnParent, SlicePredicate, ConsistencyLevel) -> cloudi:send_sync(Dispatcher, Name, {get_count, Destination, RowKey, ColumnParent, SlicePredicate, ConsistencyLevel}). %% @doc Count columns based on a slice and a list of rows %% WARNING: NOT O(1) -spec multiget_count(dispatcher(), name(), destination(), [row_key()], column_parent(), slice_predicate(), consistency_level()) -> response(). multiget_count(Dispatcher, Name, Destination, RowKeys, ColumnParent, SlicePredicate, ConsistencyLevel) when is_list(RowKeys) -> cloudi:send_sync(Dispatcher, Name, {multiget_count, Destination, RowKeys, ColumnParent, SlicePredicate, ConsistencyLevel}). %% @doc Get a list of slices for the keys within the specified KeyRange -spec get_range_slices(dispatcher(), name(), destination(), column_parent(), slice_predicate(), key_range(), consistency_level()) -> response(). get_range_slices(Dispatcher, Name, Destination, ColumnParent, SlicePredicate, KeyRange, ConsistencyLevel) -> cloudi:send_sync(Dispatcher, Name, {get_range_slices, Destination, ColumnParent, SlicePredicate, KeyRange, ConsistencyLevel}). %% @doc Get a list of slices using IndexRange -spec get_indexed_slices(dispatcher(), name(), destination(), column_parent(), slice_predicate(), key_range(), consistency_level()) -> response(). get_indexed_slices(Dispatcher, Name, Destination, ColumnParent, IndexClause, SlicePredicate, ConsistencyLevel) -> cloudi:send_sync(Dispatcher, Name, {get_indexed_slices, Destination, ColumnParent, IndexClause, SlicePredicate, ConsistencyLevel}). %% @doc Execute a CQL query -spec execute_cql_query(dispatcher(), name(), cql_query(), compression()) -> response(). execute_cql_query(Dispatcher, Name, CqlQuery, Compression) when is_binary(CqlQuery) -> cloudi:send_sync(Dispatcher, Name, {execute_cql_query, CqlQuery, Compression}). %% @doc Prepare a CQL query -spec prepare_cql_query(dispatcher(), name(), cql_query(), compression()) -> response(). prepare_cql_query(Dispatcher, Name, CqlQuery, Compression) when is_binary(CqlQuery) -> cloudi:send_sync(Dispatcher, Name, {prepare_cql_query, CqlQuery, Compression}). %% @doc Execute a prepared a CQL query -spec execute_prepared_cql_query(dispatcher(), name(), cql_query_id(), list()) -> response(). execute_prepared_cql_query(Dispatcher, Name, CqlQuery, Values) when is_integer(CqlQuery), is_list(Values) -> cloudi:send_sync(Dispatcher, Name, {execute_prepared_cql_query, CqlQuery, Values}). %% @doc Get the Thrift API version -spec describe_version(dispatcher(), name()) -> response(). describe_version(Dispatcher, Name) -> cloudi:send_sync(Dispatcher, Name, {describe_version}). %% @doc Get the snitch used for the cluster -spec describe_snitch(dispatcher(), name()) -> response(). describe_snitch(Dispatcher, Name) -> cloudi:send_sync(Dispatcher, Name, {describe_snitch}). %% @doc Get the partitioner used for the cluster -spec describe_partitioner(dispatcher(), name()) -> response(). describe_partitioner(Dispatcher, Name) -> cloudi:send_sync(Dispatcher, Name, {describe_partitioner}). %% @doc Get the schema_versions used for the cluster -spec describe_schema_versions(dispatcher(), name()) -> response(). describe_schema_versions(Dispatcher, Name) -> cloudi:send_sync(Dispatcher, Name, {describe_schema_versions}). %% @doc Get the cluster_name -spec describe_cluster_name(dispatcher(), name()) -> response(). describe_cluster_name(Dispatcher, Name) -> cloudi:send_sync(Dispatcher, Name, {describe_cluster_name}). %% @doc Get the list of all the keyspaces -spec describe_keyspaces(dispatcher(), name()) -> response(). describe_keyspaces(Dispatcher, Name) -> cloudi:send_sync(Dispatcher, Name, {describe_keyspaces}). %% @doc Gets the token ring; a map of ranges to host addresses -spec describe_ring(dispatcher(), name(), destination()) -> response(). describe_ring(Dispatcher, Name, Destination) -> cloudi:send_sync(Dispatcher, Name, {describe_ring, Destination}). %%%------------------------------------------------------------------------ %%% Callback functions from cloudi_service %%%------------------------------------------------------------------------ cloudi_service_init(Args, _Prefix, Dispatcher) -> Defaults = [{connection_name, ?DEFAULT_CONNECTION_NAME}, {pool_options, ?DEFAULT_POOL_OPTIONS}, {connection_options, ?DEFAULT_CONNECTION_OPTIONS}], [ConnectionName, PoolOptions, ConnectionOptions] = cloudi_proplists:take_values(Defaults, Args), % Use the connection_name to create a CQL connection pool. Note that % if thisin this case, pool_options should be '1', unless no % prepared statements are going to be used case erlang_cassandra:start_cql_pool(binary_name(ConnectionName), PoolOptions, ConnectionOptions) of {ok, _} -> cloudi_service:subscribe(Dispatcher, subscription_name(ConnectionName)), {ok, #state{connection_name = ConnectionName}}; {error, Reason} -> {stop, Reason} end. cloudi_service_handle_request(_Type, _Name, _Pattern, _RequestInfo, Request, Timeout, _Priority, _TransId, _Pid, #state{connection_name = ConnectionName } = State, _Dispatcher) -> case Request of Command when is_binary(Command) -> {reply, do_query(Command, ConnectionName, Timeout), State}; Command -> {reply, process_query(ConnectionName, Command), State} end. cloudi_service_handle_info(Request, State, _) -> ?LOG_WARN("Unknown info \"~p\"", [Request]), {noreply, State}. cloudi_service_terminate(_, undefined) -> ok; cloudi_service_terminate(_, #state{connection_name = ConnectionName}) -> erlang_cassandra:stop_cql_pool(ConnectionName), ok. %%%------------------------------------------------------------------------ %%% Private functions %%%------------------------------------------------------------------------ %% do a single query and return a boolean to determine if the query succeeded do_query(Query, ConnectionName, _Timeout) -> try Command = cloudi_string:binary_to_term(Query), process_query(ConnectionName, Command) of {error, invalid_call} -> ?LOG_DEBUG("Invalid erlang_cassandra command tuple ~p", [binary_to_list(Query)]), <<>>; Response -> cloudi_string:term_to_binary(Response) catch _:Reason -> ?LOG_DEBUG("exception when processing " "erlang_cassandra command tuple ~p: ~p", [binary_to_list(Query), Reason]), <<>> end. process_query(_ConnectionName, {'set_keyspace', Destination}) -> erlang_cassandra:set_keyspace(Destination); process_query(_ConnectionName, {'describe_keyspace', Destination}) -> erlang_cassandra:describe_keyspace(Destination); process_query(_ConnectionName, {'system_add_keyspace', KeyspaceDefinition}) -> erlang_cassandra:system_add_keyspace(KeyspaceDefinition); process_query(_ConnectionName, {'system_update_keyspace', KeyspaceDefinition}) -> erlang_cassandra:system_update_keyspace(KeyspaceDefinition); process_query(_ConnectionName, {'system_drop_keyspace', Destination}) -> erlang_cassandra:system_drop_keyspace(Destination); process_query(_ConnectionName, {'insert', Destination, RowKey, ColumnParent, Column, ConsistencyLevel}) -> erlang_cassandra:insert(Destination, RowKey, ColumnParent, Column, ConsistencyLevel); process_query(_ConnectionName, {'get', Destination, RowKey, ColumnPath, ConsistencyLevel}) -> erlang_cassandra:get(Destination, RowKey, ColumnPath, ConsistencyLevel); process_query(_ConnectionName, {'remove', Destination, RowKey, ColumnPath, ColumnTimestamp, ConsistencyLevel}) -> erlang_cassandra:remove(Destination, RowKey, ColumnPath, ColumnTimestamp, ConsistencyLevel); process_query(_ConnectionName, {'system_add_column_family', ColumnFamilyDefinition}) -> erlang_cassandra:system_add_column_family(ColumnFamilyDefinition); process_query(_ConnectionName, {'system_describe_column_family', Destination, ColumnFamily}) -> erlang_cassandra:system_describe_column_family(Destination, ColumnFamily); process_query(_ConnectionName, {'system_drop_column_family', Destination, ColumnFamily}) -> erlang_cassandra:system_drop_column_family(Destination, ColumnFamily); process_query(_ConnectionName, {'system_update_column_family', ColumnFamilyDefinition}) -> erlang_cassandra:system_update_column_family(ColumnFamilyDefinition); process_query(_ConnectionName, {'truncate', Destination, ColumnFamily}) -> erlang_cassandra:truncate(Destination, ColumnFamily); process_query(_ConnectionName, {'add', Destination, RowKey, ColumnParent, CounterColumn, ConsistencyLevel}) -> erlang_cassandra:add(Destination, RowKey, ColumnParent, CounterColumn, ConsistencyLevel); process_query(_ConnectionName, {'remove_counter', Destination, RowKey, ColumnPath, ConsistencyLevel}) -> erlang_cassandra:remove_counter(Destination, RowKey, ColumnPath, ConsistencyLevel); process_query(_ConnectionName, {'get_slice', Destination, RowKey, ColumnParent, SlicePredicate, ConsistencyLevel}) -> erlang_cassandra:get_slice(Destination, RowKey, ColumnParent, SlicePredicate, ConsistencyLevel); process_query(_ConnectionName, {'multiget_slice', Destination, RowKeys, ColumnParent, SlicePredicate, ConsistencyLevel}) -> erlang_cassandra:multiget_slice(Destination, RowKeys, ColumnParent, SlicePredicate, ConsistencyLevel); process_query(_ConnectionName, {'get_count', Destination, RowKey, ColumnParent, SlicePredicate, ConsistencyLevel}) -> erlang_cassandra:get_count(Destination, RowKey, ColumnParent, SlicePredicate, ConsistencyLevel); process_query(_ConnectionName, {'multiget_count', Destination, RowKeys, ColumnParent, SlicePredicate, ConsistencyLevel}) -> erlang_cassandra:multiget_count(Destination, RowKeys, ColumnParent, SlicePredicate, ConsistencyLevel); process_query(_ConnectionName, {'get_range_slices', Destination, ColumnParent, SlicePredicate, KeyRange, ConsistencyLevel}) -> erlang_cassandra:get_range_slices(Destination, ColumnParent, SlicePredicate, KeyRange, ConsistencyLevel); process_query(_ConnectionName, {'get_indexed_slices', Destination, ColumnParent, IndexClause, SlicePredicate, ConsistencyLevel}) -> erlang_cassandra:get_indexed_slices(Destination, ColumnParent, IndexClause, SlicePredicate, ConsistencyLevel); process_query(ConnectionName, {'execute_cql_query', CqlQuery, Compression}) -> erlang_cassandra:execute_cql_query(ConnectionName, CqlQuery, Compression); process_query(ConnectionName, {'prepare_cql_query', CqlQuery, Compression}) -> erlang_cassandra:prepare_cql_query(ConnectionName, CqlQuery, Compression); process_query(ConnectionName, {'execute_prepared_cql_query', CqlQuery, Values}) -> erlang_cassandra:execute_prepared_cql_query(ConnectionName, CqlQuery, Values); process_query(_ConnectionName, {'describe_version'}) -> erlang_cassandra:describe_version(); process_query(_ConnectionName, {'describe_snitch'}) -> erlang_cassandra:describe_snitch(); process_query(_ConnectionName, {'describe_partitioner'}) -> erlang_cassandra:describe_partitioner(); process_query(_ConnectionName, {'describe_schema_versions'}) -> erlang_cassandra:describe_schema_versions(); process_query(_ConnectionName, {'describe_cluster_name'}) -> erlang_cassandra:describe_cluster_name(); process_query(_ConnectionName, {'describe_keyspaces'}) -> erlang_cassandra:describe_keyspaces(); process_query(_ConnectionName, {'describe_ring', Destination}) -> erlang_cassandra:describe_keyspaces(Destination); process_query(_ConnectionName, _) -> {error, invalid_call}. binary_name({Host, Port, Name}) when is_list(Name) -> {Host, Port, erlang:list_to_binary(Name)}; binary_name({Host, Port, Name}) when is_binary(Name) -> {Host, Port, Name}; binary_name(Name) when is_list(Name) -> erlang:list_to_binary(Name); binary_name(Name) when is_binary(Name) -> Name. subscription_name({_, _, Name}) when is_list(Name) -> Name; subscription_name({_, _, Name}) when is_binary(Name) -> erlang:binary_to_list(Name); subscription_name(Name) when is_list(Name) -> Name; subscription_name(Name) when is_binary(Name) -> erlang:binary_to_list(Name).