%% @copyright 2016 Choven Corp. %% %% This file is part of erlangzmq. %% %% erlangzmq is free software: you can redistribute it and/or modify %% it under the terms of the GNU Affero General Public License as published by %% the Free Software Foundation, either version 3 of the License, or %% (at your option) any later version. %% %% erlangzmq is distributed in the hope that it will be useful, %% but WITHOUT ANY WARRANTY; without even the implied warranty of %% MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the %% GNU Affero General Public License for more details. %% %% You should have received a copy of the GNU Affero General Public License %% along with erlangzmq. If not, see %% @doc ZeroMQ Socket Type implementation for Erlang %% @hidden -module(erlangzmq_socket). -behaviour(gen_server). %% api behaviour -export([start_link/2]). %% gen_server behaviors -export([code_change/3, handle_call/3, handle_cast/2, handle_info/2, init/1, terminate/2]). %% public API implementation -spec start_link(Type::term(), Identity::string()) -> {ok, Pid::pid()} | {error, Reason::term()}. start_link(Type, Identity) when is_atom(Type), is_list(Identity) -> gen_server:start_link(?MODULE, {Type, Identity}, []). %% gen_server implementation init({Type, Identity}) -> process_flag(trap_exit, true), case erlangzmq_pattern:module(Type) of {error, Reason} -> {stop, Reason}; ModuleName -> ModuleName:init(Identity) end. code_change(_OldVsn, State, _Extra) -> {ok, State}. handle_call({connect, Protocol, Host, Port, Resource}, _From, State) -> Module = module(State), {SocketType, PeerOpts} = peer_flags(State), case erlangzmq_peer:connect(SocketType, Protocol, Host, Port, Resource, PeerOpts) of {ok, Pid} -> Module:accept_peer(State, Pid); {error, Reason} -> {reply, {error, Reason}, State} end; handle_call({accept, SocketPid}, _From, State) -> Module = module(State), {SocketType, PeerOpts} = peer_flags(State), case erlangzmq_peer:accept(SocketType, SocketPid, PeerOpts) of {ok, Pid} -> Module:accept_peer(State, Pid); {error, Reason} -> {reply, {error, Reason}, State} end; handle_call({send, Data}, From, State) -> Module = module(State), Module:send(State, Data, From); handle_call(recv, From, State) -> Module = module(State), Module:recv(State, From); handle_call({send_multipart, Multipart}, From, State) -> Module = module(State), Module:send_multipart(State, Multipart, From); handle_call(recv_multipart, From, State) -> Module = module(State), Module:recv_multipart(State, From); handle_call({bind, tcp, Host, Port}, _From, State) -> Reply = erlangzmq_bind:start_link(Host, Port), {reply, Reply, State}; handle_call(get_flags, _From, State) -> {reply, peer_flags(State), State}; handle_call({bind, Protocol, _Host, _Port}, _From, State) -> {reply, {error, {unsupported_protocol, Protocol}}, State}. handle_cast({peer_ready, From, Identity}, State) -> Module = module(State), Module:peer_ready(State, From, Identity); handle_cast({subscribe, Topic}, State) -> pattern_support(State, subscribe, [Topic]); handle_cast({cancel, Topic}, State) -> pattern_support(State, cancel, [Topic]); handle_cast({peer_subscribe, From, Subscription}, State) -> pattern_support(State, peer_subscribe, [From, Subscription]); handle_cast({peer_cancel_subscribe, From, Subscription}, State) -> pattern_support(State, peer_cancel_subscribe, [From, Subscription]); handle_cast({peer_reconnected, From}, State) -> pattern_support(State, peer_reconnected, [From], nowarn); handle_cast(CastMsg, State) -> error_logger:info_report([ unhandled_handle_cast, {module, ?MODULE}, {msg, CastMsg} ]), {noreply, State}. handle_info({peer_recv_message, Message, From}, State) -> Module = module(State), Module:peer_recv_message(State, Message, From); handle_info({queue_ready, Identity, From}, State) -> Module = module(State), Module:queue_ready(State, Identity, From); handle_info({'EXIT', PeerPid, {shutdown, _Reason}}, State) -> Module = module(State), Module:peer_disconected(State, PeerPid); handle_info(InfoMsg, State) -> error_logger:info_report([ unhandled_handle_info, {module, ?MODULE}, {msg, InfoMsg} ]), {noreply, State}. terminate(_Reason, _State) -> %% TODO: close all sockets ok. %% private api module(State) -> %% module pattern that handling this request MUST to be first value %% of the state element(1, State). identity(State) -> %% identity of pattern that handling this request MUST to be second value %% of the state element(2, State). peer_flags(State) -> Module = module(State), {SocketType, PeerOpts} = Module:peer_flags(State), Identity = identity(State), {SocketType, [{identity, Identity} | PeerOpts]}. pattern_support(State, Function, Args) -> pattern_support(State, Function, Args, warn). pattern_support(State, Function, Args, Alert) -> Module = module(State), IsExported = erlang:function_exported(Module, Function, length(Args) + 1), case {IsExported, Alert} of {true, _} -> apply(Module, Function, [State] ++ Args); {false, warn} -> error_logger:warning_report([ pattern_not_supported, {module, Module}, {method, Function}, {args, Args} ]), {noreply, State}; {false, _} -> {noreply, State} end.