diff options
Diffstat (limited to 'lib/diameter/src')
-rw-r--r-- | lib/diameter/src/base/diameter_lib.erl | 2 | ||||
-rw-r--r-- | lib/diameter/src/base/diameter_peer.erl | 6 | ||||
-rw-r--r-- | lib/diameter/src/base/diameter_peer_fsm.erl | 82 | ||||
-rw-r--r-- | lib/diameter/src/base/diameter_service.erl | 25 | ||||
-rw-r--r-- | lib/diameter/src/base/diameter_traffic.erl | 1 | ||||
-rw-r--r-- | lib/diameter/src/compiler/diameter_exprecs.erl | 4 | ||||
-rw-r--r-- | lib/diameter/src/transport/diameter_sctp.erl | 57 | ||||
-rw-r--r-- | lib/diameter/src/transport/diameter_tcp.erl | 92 |
8 files changed, 161 insertions, 108 deletions
diff --git a/lib/diameter/src/base/diameter_lib.erl b/lib/diameter/src/base/diameter_lib.erl index 8792e97621..1c1ea42cb5 100644 --- a/lib/diameter/src/base/diameter_lib.erl +++ b/lib/diameter/src/base/diameter_lib.erl @@ -283,7 +283,7 @@ ip(T) %% Or not: convert from '.'/':'-separated decimal/hex. ip(Addr) -> - {ok, A} = inet_parse:address(Addr), %% documented in inet(3) + {ok, A} = inet:parse_address(Addr), A. %% --------------------------------------------------------------------------- diff --git a/lib/diameter/src/base/diameter_peer.erl b/lib/diameter/src/base/diameter_peer.erl index 2759f17e64..4cb5a57a54 100644 --- a/lib/diameter/src/base/diameter_peer.erl +++ b/lib/diameter/src/base/diameter_peer.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2015. All Rights Reserved. +%% Copyright Ericsson AB 2010-2017. All Rights Reserved. %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. @@ -202,10 +202,10 @@ match1(Addr, Match) -> match(Addr, {ok, A}, _) -> Addr == A; match(Addr, {error, _}, RE) -> - match == re:run(inet_parse:ntoa(Addr), RE, [{capture, none}]). + match == re:run(inet:ntoa(Addr), RE, [{capture, none}, caseless]). addr([_|_] = A) -> - inet_parse:address(A); + inet:parse_address(A); addr(A) -> {ok, A}. diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl index 77ee3d6057..d99f11a697 100644 --- a/lib/diameter/src/base/diameter_peer_fsm.erl +++ b/lib/diameter/src/base/diameter_peer_fsm.erl @@ -128,9 +128,8 @@ %% outgoing DPR; boolean says whether or not %% the request was sent explicitly with %% diameter:call/4. - codec :: #{decode_format := record, + codec :: #{decode_format := diameter:decode_format(), string_decode := boolean(), - strict_arities => diameter:strict_arities(), strict_mbit := boolean(), rfc := 3588 | 6733, ordered_encode := false}, @@ -260,8 +259,7 @@ i({Ack, WPid, {M, Ref} = T, Opts, {SvcOpts, Nodes, Dict0, Svc}}) -> strict_mbit, rfc, ordered_encode], - SvcOpts#{ordered_encode => false, - decode_format => record})}. + SvcOpts#{ordered_encode => false})}. %% The transport returns its local ip addresses so that different %% transports on the same service can use different local addresses. %% The local addresses are put into Host-IP-Address avps here when @@ -546,11 +544,11 @@ put_route(Pid) -> MRef = monitor(process, Pid), put(Pid, MRef). -%% get_route/2 +%% get_route/3 -%% incoming answer -get_route(_, #diameter_packet{header = #diameter_header{is_request = false}} - = Pkt) -> +%% Incoming answer. +get_route(_, _, #diameter_packet{header = #diameter_header{is_request = false}} + = Pkt) -> Seqs = diameter_codec:sequence_numbers(Pkt), case erase(Seqs) of {Pid, Ref, MRef} -> @@ -561,8 +559,14 @@ get_route(_, #diameter_packet{header = #diameter_header{is_request = false}} false end; -%% incoming request -get_route(Ack, _) -> +%% Requests answered here ... +get_route(_, N, _) + when N == 'CER'; + N == 'DPR' -> + false; + +%% ... or not. +get_route(Ack, _, _) -> Ack. %% erase_route/1 @@ -747,7 +751,7 @@ recv1('DPA' = Name, %% Any other message with a header and no length errors. recv1(Name, H, Msg, #state{parent = Pid, ack = Ack} = S) -> Pkt = pkt(H, Msg), - Pid ! {recv, self(), get_route(Ack, Pkt), Name, Pkt}, + Pid ! {recv, self(), get_route(Ack, Name, Pkt), Name, Pkt}, handle(Name, Pkt, S). %% pkt/2 @@ -812,7 +816,8 @@ handle('DPA' = N, %% service: explicit DPR is counted in the same way %% as other explicitly sent requests. incr(recv, H, Dict0), - incr_rc(recv, diameter_codec:decode(Dict0, Opts, Pkt), Dict0) + {_, RecPkt} = decode(Dict0, Opts, Pkt), + incr_rc(recv, RecPkt, Dict0) end, diameter_peer:close(TPid), {stop, N}; @@ -916,21 +921,30 @@ handle_request(Name, = S) -> ?LOG(recv, Name), incr(recv, H, Dict0), - send_answer(Name, diameter_codec:decode(Dict0, Opts, Pkt), S). + send_answer(Name, decode(Dict0, Opts, Pkt), S). + +%% decode/3 +%% +%% Decode the message as record for diameter_capx, and in the +%% configured format for events. + +decode(Dict0, Opts, Pkt) -> + {diameter_codec:decode(Dict0, Opts, Pkt), + diameter_codec:decode(Dict0, Opts#{decode_format := record}, Pkt)}. %% send_answer/3 -send_answer(Type, ReqPkt, #state{transport = TPid, - dictionary = Dict, - codec = Opts} - = S) -> - incr_error(recv, ReqPkt, Dict), +send_answer(Type, {DecPkt, RecPkt}, #state{transport = TPid, + dictionary = Dict, + codec = Opts} + = S) -> + incr_error(recv, RecPkt, Dict), #diameter_packet{header = H, transport_data = TD} - = ReqPkt, + = RecPkt, - {Msg, PostF} = build_answer(Type, ReqPkt, S), + {Msg, PostF} = build_answer(Type, DecPkt, RecPkt, S), %% An answer message clears the R and T flags and retains the P %% flag. The E flag is set at encode. @@ -958,15 +972,15 @@ eval([F|A], S) -> eval(T, _) -> close(T). -%% build_answer/3 +%% build_answer/4 build_answer('CER', + DecPkt, #diameter_packet{msg = CER, header = #diameter_header{version = ?DIAMETER_VERSION, is_error = false}, - errors = []} - = Pkt, + errors = []}, #state{dictionary = Dict0} = S) -> {SupportedApps, RCaps, CEA} = recv_CER(CER, S), @@ -984,25 +998,25 @@ build_answer('CER', orelse ?THROW(4003), %% DIAMETER_ELECTION_LOST caps_cb(Caps) of - N -> {cea(CEA, N, Dict0), [fun open/5, Pkt, + N -> {cea(CEA, N, Dict0), [fun open/5, DecPkt, SupportedApps, Caps, {accept, inband_security(IS)}]} catch ?FAILURE(Reason) -> - rejected(Reason, {'CER', Reason, Caps, Pkt}, S) + rejected(Reason, {'CER', Reason, Caps, DecPkt}, S) end; %% The error checks below are similar to those in diameter_traffic for %% other messages. Should factor out the commonality. build_answer(Type, + DecPkt, #diameter_packet{header = H, - errors = Es} - = Pkt, + errors = Es}, S) -> {RC, FailedAVP} = result_code(Type, H, Es), - {answer(Type, RC, FailedAVP, S), post(Type, RC, Pkt, S)}. + {answer(Type, RC, FailedAVP, S), post(Type, RC, DecPkt, S)}. inband_security([]) -> ?NO_INBAND_SECURITY; @@ -1174,12 +1188,10 @@ handle_CEA(#diameter_packet{header = H} = S) -> incr(recv, H, Dict0), - #diameter_packet{} - = DPkt - = diameter_codec:decode(Dict0, Opts, Pkt), + {DecPkt, RecPkt} = decode(Dict0, Opts, Pkt), - RC = result_code(incr_rc(recv, DPkt, Dict0)), - {SApps, IS, RCaps} = recv_CEA(DPkt, S), + RC = result_code(incr_rc(recv, RecPkt, Dict0)), + {SApps, IS, RCaps} = recv_CEA(RecPkt, S), #diameter_caps{origin_host = {OH, DH}} = Caps @@ -1202,9 +1214,9 @@ handle_CEA(#diameter_packet{header = H} orelse ?THROW(election_lost), caps_cb(Caps) of - _ -> open(DPkt, SApps, Caps, {connect, hd([_] = IS)}, S) + _ -> open(DecPkt, SApps, Caps, {connect, hd([_] = IS)}, S) catch - ?FAILURE(Reason) -> close({'CEA', Reason, Caps, DPkt}) + ?FAILURE(Reason) -> close({'CEA', Reason, Caps, DecPkt}) end. %% Check more than the result code since the peer could send success %% regardless. If not 2001 then a peer_up callback could do anything diff --git a/lib/diameter/src/base/diameter_service.erl b/lib/diameter/src/base/diameter_service.erl index c7b0e706a5..802e6e7243 100644 --- a/lib/diameter/src/base/diameter_service.erl +++ b/lib/diameter/src/base/diameter_service.erl @@ -529,6 +529,13 @@ transition({tc_timeout, T}, S) -> tc_timeout(T, S), ok; +transition({nodeup, Node, _}, S) -> + nodeup(Node, S), + ok; + +transition({nodedown, _Node, _}, _) -> + ok; + transition(Req, S) -> unexpected(handle_info, [Req], S), ok. @@ -742,6 +749,8 @@ mref(P) -> init_shared(#state{options = #{use_shared_peers := T}, service_name = Svc}) -> + T == false orelse net_kernel:monitor_nodes(true, [{node_type, visible}, + nodedown_reason]), notify(T, Svc, {service, self()}). init_mod(#diameter_app{alias = Alias, @@ -757,6 +766,11 @@ notify(Share, SvcName, T) -> %% Test for the empty list for upgrade reasons: there's no %% diameter_peer:notify/3 in old code. +nodeup(Node, #state{options = #{share_peers := SP}, + service_name = SvcName}) -> + lists:member(Node, remotes(SP)) + andalso diameter_peer:notify([Node], SvcName, {service, self()}). + remotes(false) -> []; @@ -1433,9 +1447,15 @@ is_remote(Pid, T) -> %% # remote_peer_up/4 %% --------------------------------------------------------------------------- -remote_peer_up(TPid, Aliases, Caps, #state{options = #{use_shared_peers := T}} +remote_peer_up(TPid, Aliases, Caps, #state{options = #{use_shared_peers := T}, + remote = {PeerT, _, _}} = S) -> - is_remote(TPid, T) andalso rpu(TPid, Aliases, Caps, S). + is_remote(TPid, T) + andalso not ets:member(PeerT, TPid) + andalso rpu(TPid, Aliases, Caps, S). + +%% Notification can be duplicate since remote nodes push and the local +%% node pulls. rpu(TPid, Aliases, Caps, #state{service = Svc, remote = RT}) -> #diameter_service{applications = Apps} = Svc, @@ -1445,6 +1465,7 @@ rpu(TPid, Aliases, Caps, #state{service = Svc, remote = RT}) -> rpu(_, [] = No, _, _) -> No; + rpu(TPid, Aliases, Caps, {PeerT, _, _} = RT) -> monitor(process, TPid), ets:insert(PeerT, #peer{pid = TPid, diff --git a/lib/diameter/src/base/diameter_traffic.erl b/lib/diameter/src/base/diameter_traffic.erl index 27a41d6eb0..b89c4648d1 100644 --- a/lib/diameter/src/base/diameter_traffic.erl +++ b/lib/diameter/src/base/diameter_traffic.erl @@ -92,6 +92,7 @@ caller :: pid() | undefined, %% calling process handler :: pid(), %% request process peer :: undefined | {pid(), #diameter_caps{}}, + caps :: undefined, %% no longer used packet :: #diameter_packet{} | undefined}). %% of request %% --------------------------------------------------------------------------- diff --git a/lib/diameter/src/compiler/diameter_exprecs.erl b/lib/diameter/src/compiler/diameter_exprecs.erl index 9a0cb6baf2..143dede037 100644 --- a/lib/diameter/src/compiler/diameter_exprecs.erl +++ b/lib/diameter/src/compiler/diameter_exprecs.erl @@ -110,9 +110,9 @@ %% parse_transform/2 parse_transform(Forms, _Options) -> - Rs = [R || {attribute, _, record, R} <- Forms], - Es = lists:append([E || {attribute, _, export_records, E} <- Forms]), {H,T} = lists:splitwith(fun is_head/1, Forms), + Rs = [R || {attribute, _, record, R} <- H], + Es = lists:append([E || {attribute, _, export_records, E} <- H]), H ++ [a_export(Es) | f_accessors(Es, Rs)] ++ T. is_head(T) -> diff --git a/lib/diameter/src/transport/diameter_sctp.erl b/lib/diameter/src/transport/diameter_sctp.erl index 4f56475529..8de19e904e 100644 --- a/lib/diameter/src/transport/diameter_sctp.erl +++ b/lib/diameter/src/transport/diameter_sctp.erl @@ -112,7 +112,7 @@ {transport :: pid(), ack = false :: boolean(), socket :: gen_sctp:sctp_socket(), - assoc_id :: gen_sctp:assoc_id()}). %% next output stream + assoc_id :: gen_sctp:assoc_id()}). %% Listener process state. -record(listener, @@ -156,12 +156,7 @@ start(T, Svc, Opts) = Svc, diameter_sctp_sup:start(), %% start supervisors on demand Addrs = Caps#diameter_caps.host_ip_address, - s(T, Addrs, Pid, lists:map(fun ip/1, Opts)). - -ip({ifaddr, A}) -> - {ip, A}; -ip(T) -> - T. + s(T, Addrs, Pid, Opts). %% A listener spawns transports either as a consequence of this call %% when there is not yet an association to assign it, or at comm_up on @@ -354,23 +349,35 @@ l([], Ref, T) -> %% open/3 open(Addrs, Opts, PortNr) -> - {LAs, Os} = addrs(Addrs, Opts), - {LAs, case gen_sctp:open(gen_opts(portnr(Os, PortNr))) of - {ok, Sock} -> - Sock; - {error, Reason} -> - x({open, Reason}) - end}. + case gen_sctp:open(gen_opts(portnr(addrs(Addrs, Opts), PortNr))) of + {ok, Sock} -> + {addrs(Sock), Sock}; + {error, Reason} -> + x({open, Reason}) + end. addrs(Addrs, Opts) -> - case proplists:split(Opts, [ip]) of - {[[]], _} -> - {Addrs, Opts ++ [{ip, A} || A <- Addrs]}; - {[As], Os} -> - LAs = [diameter_lib:ipaddr(A) || {ip, A} <- As], - {LAs, Os ++ [{ip, A} || A <- LAs]} + case lists:mapfoldl(fun ipaddr/2, false, Opts) of + {Os, true} -> + Os; + {_, false} -> + Opts ++ [{ip, A} || A <- Addrs] end. +ipaddr({K,A}, _) + when K == ifaddr; + K == ip -> + {{ip, ipaddr(A)}, true}; +ipaddr(T, B) -> + {T, B}. + +ipaddr(A) + when A == loopback; + A == any -> + A; +ipaddr(A) -> + diameter_lib:ipaddr(A). + portnr(Opts, PortNr) -> case proplists:get_value(port, Opts) of undefined -> @@ -379,6 +386,14 @@ portnr(Opts, PortNr) -> Opts end. +addrs(Sock) -> + case inet:socknames(Sock) of + {ok, As} -> + [A || {A,_} <- As]; + {error, Reason} -> + x({socknames, Reason}) + end. + %% x/1 x(Reason) -> @@ -565,7 +580,7 @@ transition(Msg, S) %% Deferred actions from a message_cb. transition({actions, Dir, Acts}, S) -> - actions(Acts, Dir, S); + setopts(ok, actions(Acts, Dir, S)); %% Request to close the transport connection. transition({diameter, {close, Pid}}, #transport{parent = Pid}) -> diff --git a/lib/diameter/src/transport/diameter_tcp.erl b/lib/diameter/src/transport/diameter_tcp.erl index ac55d722fa..a8639baa11 100644 --- a/lib/diameter/src/transport/diameter_tcp.erl +++ b/lib/diameter/src/transport/diameter_tcp.erl @@ -142,8 +142,7 @@ -> {ok, pid(), [inet:ip_address()]} when Ref :: diameter:transport_ref(); ({connect, Ref}, #diameter_service{}, [connect_option()]) - -> {ok, pid(), [inet:ip_address()]} - | {ok, pid()} + -> {ok, pid()} when Ref :: diameter:transport_ref(). start({T, Ref}, Svc, Opts) -> @@ -258,22 +257,14 @@ i(#monitor{parent = Pid, transport = TPid} = S) -> i({listen, Ref, {Mod, Opts, Addrs}}) -> [_] = diameter_config:subscribe(Ref, transport), %% assert existence - {[LA, LP], Rest} = proplists:split(Opts, [ip, port]), - LAddrOpt = get_addr(LA, Addrs), - LPort = get_port(LP), - {ok, LSock} = Mod:listen(LPort, gen_opts(LAddrOpt, Rest)), - LAddr = laddr(LAddrOpt, Mod, LSock), + {[LP], Rest} = proplists:split(Opts, [port]), + {ok, LSock} = Mod:listen(get_port(LP), gen_opts(Addrs, Rest)), + {ok, {LAddr, _}} = sockname(Mod, LSock), true = diameter_reg:add_new({?MODULE, listener, {Ref, {LAddr, LSock}}}), proc_lib:init_ack({ok, self(), {LAddr, LSock}}), #listener{socket = LSock, module = Mod}. -laddr([], Mod, Sock) -> - {ok, {Addr, _Port}} = sockname(Mod, Sock), - Addr; -laddr([{ip, Addr}], _, _) -> - Addr. - ssl_opts([]) -> false; ssl_opts([{ssl_options, true}]) -> @@ -308,24 +299,16 @@ init(accept = T, Ref, Mod, Pid, Opts, Addrs, SvcPid) -> Sock; init(connect = T, Ref, Mod, Pid, Opts, Addrs, _SvcPid) -> - {[LA, RA, RP], Rest} = proplists:split(Opts, [ip, raddr, rport]), - LAddrOpt = get_addr(LA, Addrs), + {[RA, RP], Rest} = proplists:split(Opts, [raddr, rport]), RAddr = get_addr(RA), RPort = get_port(RP), - proc_lib:init_ack(init_rc(LAddrOpt)), - Sock = ok(connect(Mod, RAddr, RPort, gen_opts(LAddrOpt, Rest))), + proc_lib:init_ack({ok, self()}), + Sock = ok(connect(Mod, RAddr, RPort, gen_opts(Addrs, Rest))), publish(Mod, T, Ref, Sock), - up(Pid, {RAddr, RPort}, LAddrOpt, Mod, Sock), + up(Pid, {RAddr, RPort}, Mod, Sock), Sock. -init_rc([{ip, Addr}]) -> - {ok, self(), [Addr]}; -init_rc([]) -> - {ok, self()}. - -up(Pid, Remote, [{ip, _Addr}], _, _) -> - diameter_peer:up(Pid, Remote); -up(Pid, Remote, [], Mod, Sock) -> +up(Pid, Remote, Mod, Sock) -> {Addr, _Port} = ok(sockname(Mod, Sock)), diameter_peer:up(Pid, Remote, [Addr]). @@ -382,25 +365,41 @@ l([{{?MODULE, listener, {_, AS}}, LPid}], _, _) -> l([], Ref, T) -> diameter_tcp_sup:start_child({listen, Ref, T}). -%% get_addr/1 +%% addrs/2 +%% +%% Take the first address from the service if several are specified +%% and not address is configured. + +addrs(Addrs, Opts) -> + case lists:mapfoldr(fun ipaddr/2, [], Opts) of + {Os, [_]} -> + Os; + {_, []} -> + Opts ++ [{ip, A} || [A|_] <- [Addrs]]; + {_, As} -> + ?ERROR({invalid_addrs, As, Addrs}) + end. -get_addr(As) -> - diameter_lib:ipaddr(addr(As, [])). +ipaddr({K,A}, As) + when K == ifaddr; + K == ip -> + {{ip, ipaddr(A)}, [A | As]}; +ipaddr(T, B) -> + {T, B}. -%% get_addr/2 +ipaddr(A) + when A == loopback; + A == any -> + A; +ipaddr(A) -> + diameter_lib:ipaddr(A). -get_addr([], []) -> - []; -get_addr(As, Def) -> - [{ip, diameter_lib:ipaddr(addr(As, Def))}]. +%% get_addr/1 -%% Take the first address from the service if several are unspecified. -addr([], [Addr | _]) -> - Addr; -addr([{_, Addr}], _) -> - Addr; -addr(As, Addrs) -> - ?ERROR({invalid_addrs, As, Addrs}). +get_addr([{_, Addr}]) -> + diameter_lib:ipaddr(Addr); +get_addr(Addrs) -> + ?ERROR({invalid_addrs, Addrs}). %% get_port/1 @@ -413,10 +412,15 @@ get_port(Ps) -> %% gen_opts/2 -gen_opts(LAddrOpt, Opts) -> +gen_opts(Addrs, Opts) -> + gen_opts(addrs(Addrs, Opts)). + +%% gen_opts/1 + +gen_opts(Opts) -> {L,_} = proplists:split(Opts, [binary, packet, active]), [[],[],[]] == L orelse ?ERROR({reserved_options, Opts}), - [binary, {packet, 0}, {active, false}] ++ LAddrOpt ++ Opts. + [binary, {packet, 0}, {active, false} | Opts]. %% --------------------------------------------------------------------------- %% # ports/1 @@ -640,7 +644,7 @@ transition(Msg, S) %% Deferred actions from a message_cb. transition({actions, Dir, Acts}, S) -> - actions(Acts, Dir, S); + setopts(actions(Acts, Dir, S)); %% Request to close the transport connection. transition({diameter, {close, Pid}}, #transport{parent = Pid, |