From 3c15ff32e89e401b4dde2b8acc9699be2614b996 Mon Sep 17 00:00:00 2001 From: Anders Svensson Date: Wed, 18 May 2011 18:29:12 +0200 Subject: Initial commit of the diameter application. The application provides an implementation of the Diameter protocol as defined in RFC 3588. --- lib/diameter/src/transport/diameter_etcp.erl | 311 +++++++++++++++++++++++++++ 1 file changed, 311 insertions(+) create mode 100644 lib/diameter/src/transport/diameter_etcp.erl (limited to 'lib/diameter/src/transport/diameter_etcp.erl') diff --git a/lib/diameter/src/transport/diameter_etcp.erl b/lib/diameter/src/transport/diameter_etcp.erl new file mode 100644 index 0000000000..d925d62545 --- /dev/null +++ b/lib/diameter/src/transport/diameter_etcp.erl @@ -0,0 +1,311 @@ +%% +%% %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). -- cgit v1.2.3