-module(ensq). -export([start/0, init/1, producer/2, producer/3, list/0, send/2, touch/1]). -export_type([ host/0, channel/0, topic_name/0, channel_name/0 ]). -type host() :: {Host :: inet:ip_address() | inet:hostname(), Port :: inet:port_number()}. -type single_target() :: host(). -type multi_target() :: [host()]. -type target() :: single_target() | multi_target(). -type channel_name() :: binary(). -type channel() :: {Channel :: channel_name(), Callback :: atom()}. -type topic_name() :: atom() | binary(). -type topic() :: {Topic :: topic_name(), [channel()], [target()]} | {Topic :: topic_name(), [channel()]}. -type discovery_server() :: host(). -type spec() :: {[discovery_server()], [topic()]}. start() -> application:start(inets), application:start(syntax_tools), application:start(compiler), application:start(goldrush), application:start(lager), application:start(ensq). %%-------------------------------------------------------------------- %% @doc %% This function is used to initialize one or more topics on a given %% set of discovery servers. This call can be done multiple times %% in the case different discovery servers are used for different %% topic sets. %% %% @end %%-------------------------------------------------------------------- -spec init(spec()) -> ok. init({DiscoveryServers, Topics}) -> [topic_from_sepc(DiscoveryServers, Topic) || Topic <- Topics], ok. %%-------------------------------------------------------------------- %% @doc %% Returns a list of all currently known discoveryserver/topic %% combinations. %% %% @end %%-------------------------------------------------------------------- list() -> ensq_topic:list(). %%-------------------------------------------------------------------- %% @doc %% Sends a message to a topics target servers. %% %% @end %%-------------------------------------------------------------------- send(Topic, Msg) when is_binary(Msg), is_pid(Topic) orelse is_atom(Topic) -> ensq_topic:send(Topic, Msg). %%-------------------------------------------------------------------- %% @doc %% Creates a producer connection to a single host. %% %% @end %%-------------------------------------------------------------------- -spec producer(Channel::atom()|binary(), Host::inet:ip_address() | inet:hostname(), Port::inet:port_number()) -> {ok, Pid::pid()} | {error, Reason::term()}. producer(Channel, Host, Port) -> producer(Channel, [{Host, Port}]). %%-------------------------------------------------------------------- %% @doc %% Creates a producer connection to multiple hosts. %% %% @end %%-------------------------------------------------------------------- -spec producer(Channel::atom()|binary(), Targets :: [host()]) -> {ok, Pid::pid()} | {error, Reason::term()}. producer(Channel, Targets) -> ensq_topic:discover(Channel, [], [], Targets). topic_from_sepc(DiscoveryServers, {Topic, Channels}) -> ensq_topic:discover(Topic, DiscoveryServers, Channels); topic_from_sepc(DiscoveryServers, {Topic, Channels, []}) -> ensq_topic:discover(Topic, DiscoveryServers, Channels); topic_from_sepc(DiscoveryServers, {Topic, Channels, Targets}) -> ensq_topic:discover(Topic, DiscoveryServers, Channels, Targets). touch({S, MsgID}) -> gen_tcp:send(S, ensq_proto:encode({touch, MsgID})).