%%
%% %CopyrightBegin%
%%
%% Copyright Ericsson AB 2010-2011. All Rights Reserved.
%%
%% The contents of this file are subject to the Erlang Public License,
%% Version 1.1, (the "License"); you may not use this file except in
%% compliance with the License. You should have received a copy of the
%% Erlang Public License along with this software. If not, it can be
%% retrieved online at http://www.erlang.org/.
%%
%% Software distributed under the License is distributed on an "AS IS"
%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
%% the License for the specific language governing rights and limitations
%% under the License.
%%
%% %CopyrightEnd%
%%
%%
%% This module implements a transport_module that uses Erlang message
%% passing for transport.
%%
-module(diameter_etcp).
-behaviour(gen_server).
%% transport_module interface.
-export([start/3]).
%% gen_tcp-ish interface used by diameter_tcp.
-export([listen/2,
accept/1,
connect/3,
send/2,
close/1,
setopts/2,
port/1]).
%% child start
-export([start_link/1]).
%% gen_server callbacks
-export([init/1,
handle_call/3,
handle_cast/2,
handle_info/2,
code_change/3,
terminate/2]).
%% Server states.
-record(listener,
{acceptors = [] :: [pid()]}).
-record(connection,
{parent :: pid(),
peer :: {connect, reference()} %% {connect, MRef}
| accept
| pid()}).
%% start/3
%% 'module' option makes diameter_tcp call here instead of gen_tcp/inet.
start(T, Svc, Opts)
when is_list(Opts) ->
diameter_etcp_sup:start(),
diameter_tcp:start(T, Svc, [{module, ?MODULE} | Opts]).
%% listen/2
%%
%% Spawn a process that represents the listening socket. The local
%% port number can be any term, not just an integer. The listener
%% process registers its host/port with diameter_reg and this is the
%% handle with which connect/3 finds the appropriate listening
%% process.
listen(LPort, Opts) ->
Parent = self(),
diameter_etcp_sup:start_child({listen, Parent, LPort, Opts}).
%% accept/1
%%
%% Output: pid()
accept(LPid) ->
start(fun(Ref, Parent) -> acceptor(LPid, Ref, Parent) end).
%% connect/3
%%
%% Output: pid()
%% RAddr here can either be a 4/8-tuple address or {Node, Addr}.
connect(RAddr, RPort, _Opts) ->
start(fun(Ref, Parent) -> connector(RAddr, RPort, Ref, Parent) end).
%% send/2
send(Pid, Bin) ->
Pid ! {send, Bin},
ok.
%% close/1
close(Pid) ->
Pid ! close,
monitor(Pid),
receive {'DOWN', _, process, Pid, _} -> ok end.
%% setopts/2
setopts(_, _) ->
ok.
%% port/1
port(_) ->
3868. %% We have no local port: fake it.
%% start_link/1
start_link(T) ->
gen_server:start_link(?MODULE, T, []).
%% ---------------------------------------------------------------------------
%% # init/1
%% ---------------------------------------------------------------------------
%% Maintain a list of acceptor pids as the process state. Each accept
%% adds a pid to the list, each connect removes one.
init({listen, Parent, LPort, Opts}) ->
monitor(Parent),
{ip, LAddr} = lists:keyfind(ip, 1, Opts),
true = diameter_reg:add_new({?MODULE, listener, LAddr, LPort}),
{ok, #listener{}};
init({connect, Fun, Ref, Parent}) ->
{ok, #connection{parent = Parent,
peer = Fun(Ref, Parent)}}.
%% ---------------------------------------------------------------------------
%% # handle_call/3
%% ---------------------------------------------------------------------------
handle_call(_, _, State) ->
{reply, nok, State}.
%% ---------------------------------------------------------------------------
%% # handle_cast/2
%% ---------------------------------------------------------------------------
handle_cast(_, State) ->
{noreply, State}.
%% ---------------------------------------------------------------------------
%% # handle_info/2
%% ---------------------------------------------------------------------------
handle_info(T, #listener{acceptors = L} = S) ->
{noreply, S#listener{acceptors = l(T,L)}};
handle_info(T, State) ->
{noreply, transition(T, State)}.
%% ---------------------------------------------------------------------------
%% # code_change/3
%% ---------------------------------------------------------------------------
code_change(_, State, _) ->
{ok, State}.
%% ---------------------------------------------------------------------------
%% # terminate/2
%% ---------------------------------------------------------------------------
terminate(_, _) ->
ok.
%% ---------------------------------------------------------------------------
monitor(Pid) ->
erlang:monitor(process, Pid).
putr(Key, Val) ->
put({?MODULE, Key}, Val).
eraser(Key) ->
erase({?MODULE, Key}).
%% l/2
l({'DOWN', _, process, _, _} = T, _) ->
x(T);
%% New accepting process.
l({accept, APid}, As) ->
As ++ [APid];
%% Peer wants to connect but we have no acceptor ...
l({connect, Peer}, [] = As) ->
Peer ! {refused, self()},
As;
%% ... or we do.
l({connect, Peer}, [APid | Rest]) ->
Peer ! {accepted, APid},
Rest.
x(T) ->
exit({shutdown, T}).
%% start/1
start(Fun) ->
Ref = make_ref(),
{ok, Pid}
= T
= diameter_etcp_sup:start_child({connect, Fun, Ref, self()}),
MRef = monitor(Pid),
receive
{ok, Ref} ->
T;
{'DOWN', MRef, process, _, Reason} ->
{error, Reason}
end.
%% acceptor/3
acceptor(LPid, Ref, Parent) ->
LPid ! {accept, self()}, %% announce that we're accepting
putr(ref, {ok, Ref}),
monitor(Parent),
monitor(LPid),
accept.
%% connector/4
connector(RAddr, RPort, Ref, Parent) ->
c(match(RAddr, RPort), Ref, Parent).
c([], _, _) ->
x(refused);
c([{_,LPid}], Ref, Parent) ->
LPid ! {connect, self()},
putr(ref, {ok, Ref}),
monitor(Parent),
{connect, monitor(LPid)}.
match({Node, RAddr}, RPort) ->
rpc:call(Node, diameter_reg, match, [{?MODULE, listener, RAddr, RPort}]);
match(RAddr, RPort) ->
match({node(), RAddr}, RPort).
%% transition/2
%% Unexpected parent or peer death.
transition({'DOWN', _, process, _, _} = T, S) ->
element(2,S) ! {tcp_error, self(), T},
x(T);
%% Connector is receiving acceptor pid from listener.
transition({accepted, Peer}, #connection{parent = Parent,
peer = {connect, MRef}}) ->
monitor(Peer),
erlang:demonitor(MRef, [flush]),
Peer ! {connect, self()},
Parent ! {ok, _} = eraser(ref),
#connection{parent = Parent,
peer = Peer};
%% Connector is receiving connection refusal from listener.
transition({refused, _} = T, #connection{peer = {connect, _}}) ->
x(T);
%% Acceptor is receiving peer connect.
transition({connect, Peer}, #connection{parent = Parent,
peer = accept}) ->
monitor(Peer),
Parent ! {ok, _} = eraser(ref),
#connection{parent = Parent,
peer = Peer};
%% Incoming message.
transition({recv, Bin}, #connection{parent = Parent} = S) ->
Parent ! {tcp, self(), Bin},
S;
%% Outgoing message.
transition({send, Bin}, #connection{peer = Peer} = S) ->
Peer ! {recv, Bin},
S;
%% diameter_etcp:close/1 call when a peer is connected ...
transition(close = T, #connection{peer = Peer})
when is_pid(Peer) ->
Peer ! {close, self()},
x(T);
%% ... or not.
transition(close = T, #connection{}) ->
x(T);
%% Peer is closing the connection.
transition({close, Peer} = T, #connection{parent = Parent,
peer = Peer})
when is_pid(Peer) ->
Parent ! {tcp_closed, self()},
x(T).
