diff options
Diffstat (limited to 'lib/megaco/src/engine/megaco_messenger.erl')
| -rw-r--r-- | lib/megaco/src/engine/megaco_messenger.erl | 5158 |
1 files changed, 5158 insertions, 0 deletions
diff --git a/lib/megaco/src/engine/megaco_messenger.erl b/lib/megaco/src/engine/megaco_messenger.erl new file mode 100644 index 0000000000..a9e4fd67b2 --- /dev/null +++ b/lib/megaco/src/engine/megaco_messenger.erl @@ -0,0 +1,5158 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 1999-2009. 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% +%% + +%% +%%---------------------------------------------------------------------- +%% Purpose: Send and process a (sequence of) Megaco/H.248 transactions +%%---------------------------------------------------------------------- + +-module(megaco_messenger). + +%% Application internal export +-export([ + process_received_message/4, process_received_message/5, + receive_message/4, receive_message/5, + connect/4, connect/5, + disconnect/2, + encode_actions/3, + call/3, + cast/3, + cancel/2, + request_timeout/2, + request_keep_alive_timeout/2, + pending_timeout/3, + reply_timeout/3, + segment_timeout/3, + %% segment_reply_timeout/4, + + test_request/5, + test_reply/5 + ]). + +%% MIB stat functions +-export([ + get_stats/0, get_stats/1, get_stats/2, + reset_stats/0, reset_stats/1 + ]). + +%% Misc functions +-export([ + cleanup/2, + which_requests/1, which_replies/1 + ]). + +%% Module internal export +-export([ + process_received_message/6, + handle_request/2, + handle_long_request/2, + connect_remote/3, + disconnect_local/2, + disconnect_remote/3, + send_request_remote/4, + receive_reply_remote/2, receive_reply_remote/3 + ]). + +-include_lib("megaco/include/megaco.hrl"). +-include("megaco_message_internal.hrl"). +-include_lib("megaco/src/app/megaco_internal.hrl"). + +%% N.B. Update cancel/1 with '_' when a new field is added +-record(request, + {trans_id, + remote_mid, + timer_ref, % {short, Ref} | {long, Ref} + init_timer, + init_long_timer, + curr_timer, + version, + bytes, % {send, Data} | {no_send, Data}, Data = binary() | tuple() + send_handle, + user_mod, + user_args, + reply_action, % call | cast + reply_data, + seg_recv = [], % [integer()] (received segments) + init_seg_timer, + seg_timer_ref, + keep_alive_timer, % plain | integer() >= 0 + keep_alive_ref % undefined | ref() + }). + + +%% N.B. Update cancel/1 with '_' when a new field is added +-record(reply, + { + trans_id, + local_mid, + state = prepare, % prepare | eval_request | waiting_for_ack | aborted + pending_timer_ref, + handler = undefined, % pid of the proc executing the callback func + timer_ref, + version, + %% bytes: Sent reply data: not acknowledged + bytes, % binary() | [{integer(), binary(), timer_ref()}] + ack_action, % discard_ack | {handle_ack, Data} + send_handle, + %% segments: Not sent reply data (segments) + segments = [] % [{integer(), binary()}] + }). + +-record(trans_id, + { + mid, + serial + }). + + +-ifdef(MEGACO_TEST_CODE). +-define(SIM(Other,Where), + fun(Afun,Bfun) -> + Kfun = {?MODULE,Bfun}, + case (catch ets:lookup(megaco_test_data, Kfun)) of + [{Kfun,Cfun}] -> + Cfun(Afun); + _ -> + Afun + end + end(Other,Where)). +-define(TC_AWAIT_CANCEL_EVENT(), + case megaco_tc_controller:lookup(block_on_cancel) of + {value, {Tag, Pid}} when is_pid(Pid) -> + Pid ! {Tag, self()}, + receive + {Tag, Pid} -> + ok + end; + {value, {sleep, To}} when is_integer(To) andalso (To > 0) -> + receive after To -> ok end; + _ -> + ok + end). +-define(TC_AWAIT_REPLY_EVENT(Info), + case megaco_tc_controller:lookup(block_on_reply) of + {value, {Tag, Pid}} when is_pid(Pid) -> + Pid ! {Tag, self(), Info}, + receive + {Tag, Pid} -> + ok + end; + _Whatever -> + %% io:format("Whatever: ~p~n", [Whatever]), + ok + end). +-else. +-define(SIM(Other,Where),Other). +-define(TC_AWAIT_CANCEL_EVENT(),ok). +-define(TC_AWAIT_REPLY_EVENT(_),ok). +-endif. + + +-define(report_pending_limit_exceeded(ConnData), + ?report_important(ConnData, "<ERROR> pending limit exceeded", [])). + +-ifdef(megaco_extended_trace). +-define(rt1(T,F,A),?report_trace(T,F,A)). +-define(rt2(F,A), ?rt1(ignore,F,A)). +-define(rt3(F), ?rt2(F,[])). +-else. +-define(rt1(T,F,A),ok). +-define(rt2(F,A), ok). +-define(rt3(F), ok). +-endif. + + +%%---------------------------------------------------------------------- +%% SNMP statistics handling functions +%%---------------------------------------------------------------------- + +%%----------------------------------------------------------------- +%% Func: get_stats/0, get_stats/1, get_stats/2 +%% Description: Retreive statistics (counters) for TCP +%%----------------------------------------------------------------- + +get_stats() -> + megaco_stats:get_stats(megaco_stats). + +get_stats(ConnHandleOrCounter) -> + megaco_stats:get_stats(megaco_stats, ConnHandleOrCounter). + +get_stats(ConnHandle, Counter) -> + megaco_stats:get_stats(megaco_stats, ConnHandle, Counter). + + +%%----------------------------------------------------------------- +%% Func: reset_stats/0, reaet_stats/1 +%% Description: Reset statistics (counters) +%%----------------------------------------------------------------- + +reset_stats() -> + megaco_stats:reset_stats(megaco_stats). + +reset_stats(ConnHandleOrCounter) -> + megaco_stats:reset_stats(megaco_stats, ConnHandleOrCounter). + + + +%%---------------------------------------------------------------------- +%% cleanup utility functions +%%---------------------------------------------------------------------- + +cleanup(#megaco_conn_handle{local_mid = LocalMid}, Force) + when (Force =:= true) orelse (Force =:= false) -> + Pat = #reply{trans_id = '$1', + local_mid = LocalMid, + state = '$2', + _ = '_'}, + do_cleanup(Pat, Force); +cleanup(LocalMid, Force) + when (Force =:= true) orelse (Force =:= false) -> + Pat = #reply{trans_id = '$1', + local_mid = LocalMid, + state = '$2', + _ = '_'}, + do_cleanup(Pat, Force). + +do_cleanup(Pat, Force) -> + Match = megaco_monitor:which_replies(Pat), + Reps = [{V1, V2} || [V1, V2] <- Match], + do_cleanup2(Reps, Force). + +do_cleanup2([], _) -> + ok; +do_cleanup2([{TransId, aborted}|T], Force = false) -> + megaco_monitor:delete_reply(TransId), + do_cleanup2(T, Force); +do_cleanup2([_|T], Force = false) -> + do_cleanup2(T, Force); +do_cleanup2([{TransId, _State}|T], Force = true) -> + megaco_monitor:delete_reply(TransId), + do_cleanup2(T, Force). + + +%%---------------------------------------------------------------------- +%% which_requests and which_replies utility functions +%%---------------------------------------------------------------------- + +which_requests(#megaco_conn_handle{local_mid = LocalMid, + remote_mid = RemoteMid}) -> + Pat1 = #trans_id{mid = LocalMid, + serial = '$1', _ = '_'}, + Pat2 = #request{trans_id = Pat1, + remote_mid = RemoteMid, + _ = '_'}, + Match = megaco_monitor:which_requests(Pat2), + [S || [S] <- Match]; +which_requests(LocalMid) -> + Pat1 = #trans_id{mid = LocalMid, + serial = '$1', _ = '_'}, + Pat2 = #request{trans_id = Pat1, + remote_mid = '$2', _ = '_'}, + Match0 = megaco_monitor:which_requests(Pat2), + Match1 = [{mk_ch(LocalMid, V2), V1} || [V1, V2] <- Match0], + which_requests1(lists:sort(Match1)). + +which_requests1([]) -> + []; +which_requests1([{CH, S}|T]) -> + which_requests2(T, CH, [S], []). + +which_requests2([], CH, Serials, Reqs) -> + lists:reverse([{CH, Serials}|Reqs]); +which_requests2([{CH, S}|T], CH, Serials, Reqs) -> + which_requests2(T, CH, [S|Serials], Reqs); +which_requests2([{CH1, S}|T], CH2, Serials, Reqs) -> + which_requests2(T, CH1, [S], [{CH2, lists:reverse(Serials)}| Reqs]). + + +which_replies(#megaco_conn_handle{local_mid = LocalMid, + remote_mid = RemoteMid}) -> + Pat1 = #trans_id{mid = RemoteMid, + serial = '$1', _ = '_'}, + Pat2 = #reply{trans_id = Pat1, + local_mid = LocalMid, + state = '$2', + handler = '$3', _ = '_'}, + Match = megaco_monitor:which_replies(Pat2), + [{V1, V2, V3} || [V1, V2, V3] <- Match]; +which_replies(LocalMid) -> + Pat1 = #trans_id{mid = '$1', + serial = '$2', _ = '_'}, + Pat2 = #reply{trans_id = Pat1, + local_mid = LocalMid, + state = '$3', + handler = '$4', _ = '_'}, + Match0 = megaco_monitor:which_replies(Pat2), + Match1 = [{mk_ch(LocalMid,V1),{V2,V3,V4}} || [V1, V2, V3, V4] <- Match0], + which_replies1(lists:sort(Match1)). + +which_replies1([]) -> + []; +which_replies1([{CH, Data}|T]) -> + which_replies2(T, CH, [Data], []). + +which_replies2([], CH, Data, Reps) -> + lists:reverse([{CH, Data}|Reps]); +which_replies2([{CH, Data}|T], CH, Datas, Reps) -> + which_replies2(T, CH, [Data|Datas], Reps); +which_replies2([{CH1, Data}|T], CH2, Datas, Reps) -> + which_replies2(T, CH1, [Data], [{CH2, lists:reverse(Datas)}| Reps]). + + +mk_ch(LM, RM) -> + #megaco_conn_handle{local_mid = LM, remote_mid = RM}. + + +%%---------------------------------------------------------------------- +%% Register/unreister connections +%%---------------------------------------------------------------------- + +%% Returns {ok, ConnHandle} | {error, Reason} +autoconnect(RH, RemoteMid, SendHandle, ControlPid, Extra) + when is_record(RH, megaco_receive_handle) -> + ?rt2("autoconnect", [RH, RemoteMid, SendHandle, ControlPid]), + case megaco_config:autoconnect(RH, RemoteMid, SendHandle, ControlPid) of + {ok, ConnData} -> + do_connect(ConnData, Extra); + {error, Reason} -> + {error, Reason} + end; +autoconnect(BadHandle, _CH, _SendHandle, _ControlPid, _Extra) -> + {error, {bad_receive_handle, BadHandle}}. + +connect(RH, RemoteMid, SendHandle, ControlPid) -> + Extra = ?default_user_callback_extra, + connect(RH, RemoteMid, SendHandle, ControlPid, Extra). +connect(RH, RemoteMid, SendHandle, ControlPid, Extra) + when is_record(RH, megaco_receive_handle) -> + ?rt2("connect", [RH, RemoteMid, SendHandle, ControlPid, Extra]), + + %% The purpose of this is to have a temoporary process, to + %% which one can set up a monitor or link and get a + %% notification when process exits. The entire connect is + %% done in the temporary worker process. + %% When it exits, the connect is either successfully done + %% or it failed. + + ConnectorFun = + fun() -> + + ConnectResult = + case megaco_config:connect(RH, RemoteMid, + SendHandle, ControlPid) of + {ok, ConnData} -> + do_connect(ConnData, Extra); + {error, Reason} -> + {error, Reason} + end, + ?rt2("connector: connected", [self(), ConnectResult]), + exit({result, ConnectResult}) + end, + Flag = process_flag(trap_exit, true), + Connector = erlang:spawn_link(ConnectorFun), + receive + {'EXIT', Connector, {result, ConnectResult}} -> + ?rt2("connect result: received expected connector exit signal", + [Connector, ConnectResult]), + process_flag(trap_exit, Flag), + ConnectResult; + {'EXIT', Connector, OtherReason} -> + ?rt2("connect exit: received unexpected connector exit signal", + [Connector, OtherReason]), + process_flag(trap_exit, Flag), + {error, OtherReason} + end; +connect(BadHandle, _CH, _SendHandle, _ControlPid, _Extra) -> + {error, {bad_receive_handle, BadHandle}}. + +do_connect(CD, Extra) -> + CH = CD#conn_data.conn_handle, + Version = CD#conn_data.protocol_version, + UserMod = CD#conn_data.user_mod, + UserArgs = CD#conn_data.user_args, + Args = + case Extra of + ?default_user_callback_extra -> + [CH, Version | UserArgs]; + _ -> + [CH, Version, Extra | UserArgs] + end, + ?report_trace(CD, "callback: connect", [Args]), + Res = (catch apply(UserMod, handle_connect, Args)), + ?report_debug(CD, "return: connect", [{return, Res}]), + case Res of + ok -> + ?SIM(ok, do_connect), % do_encode), + monitor_process(CH, CD#conn_data.control_pid); + error -> + megaco_config:disconnect(CH), + {error, {connection_refused, CD, error}}; + {error, ED} when is_record(ED,'ErrorDescriptor') -> + megaco_config:disconnect(CH), + {error, {connection_refused, CD, ED}}; + _Error -> + warning_msg("connect callback failed: ~w", [Res]), + megaco_config:disconnect(CH), + {error, {connection_refused, CD, Res}} + end. + +finish_connect(#conn_data{control_pid = ControlPid} = CD) + when is_pid(ControlPid) andalso (node(ControlPid) =:= node()) -> + ?rt1(CD, "finish local connect", [ControlPid]), + do_finish_connect(CD); +finish_connect(#conn_data{conn_handle = CH, + control_pid = ControlPid} = CD) + when is_pid(ControlPid) andalso (node(ControlPid) =/= node()) -> + ?rt1(CD, "finish remote connect", [ControlPid]), + RemoteNode = node(ControlPid), + UserMonitorPid = whereis(megaco_monitor), + Args = [CH, ControlPid, UserMonitorPid], + case rpc:call(RemoteNode, ?MODULE, connect_remote, Args) of + {ok, ControlMonitorPid} -> + do_finish_connect(CD#conn_data{control_pid = ControlMonitorPid}); + {error, Reason} -> + disconnect(CH, {connect_remote, Reason}), + {error, Reason}; + {badrpc, Reason} -> + Reason2 = {'EXIT', Reason}, + disconnect(CH, {connect_remote, Reason2}), + {error, Reason2} + end. + +do_finish_connect(#conn_data{conn_handle = CH, + send_handle = SendHandle, + control_pid = ControlPid} = CD) -> + M = ?MODULE, + F = disconnect_local, + A = [CH], + MFA = {M, F, A}, + case megaco_config:finish_connect(CH, SendHandle, ControlPid, MFA) of + {ok, Ref} -> + {ok, CD#conn_data{monitor_ref = Ref}}; + {error, Reason} -> + {error, {config_update, Reason}} + end. + + +monitor_process(CH, ControlPid) + when is_pid(ControlPid) andalso (node(ControlPid) =:= node()) -> + M = ?MODULE, + F = disconnect_local, + A = [CH], + Ref = megaco_monitor:apply_at_exit(M, F, A, ControlPid), + case megaco_config:update_conn_info(CH, monitor_ref, Ref) of + ok -> + ?SIM({ok, CH}, monitor_process_local); + {error, Reason} -> + disconnect(CH, {config_update, Reason}), + {error, Reason} + end; +monitor_process(CH, ControlPid) + when is_pid(ControlPid) andalso (node(ControlPid) =/= node()) -> + RemoteNode = node(ControlPid), + UserMonitorPid = whereis(megaco_monitor), + Args = [CH, ControlPid, UserMonitorPid], + case rpc:call(RemoteNode, ?MODULE, connect_remote, Args) of + {ok, ControlMonitorPid} -> + M = ?MODULE, + F = disconnect_local, + A = [CH], + Ref = megaco_monitor:apply_at_exit(M, F, A, ControlMonitorPid), + case megaco_config:update_conn_info(CH, monitor_ref, Ref) of + ok -> + ?SIM({ok, CH}, monitor_process_remote); + {error, Reason} -> + disconnect(CH, {config_update, Reason}), + {error, Reason} + end; + {error, Reason} -> + disconnect(CH, {connect_remote, Reason}), + {error, Reason}; + {badrpc, Reason} -> + Reason2 = {'EXIT', Reason}, + disconnect(CH, {connect_remote, Reason2}), + {error, Reason2} + end; +monitor_process(CH, undefined = _ControlPid) -> + %% We have to do this later (setting up the monitor), + %% when the first message arrives. The 'connected' atom is + %% the indication for the first arriving message to finish + %% the connect. + %% This may be the case when an MGC performs a pre-connect + %% in order to speed up the handling of an (expected) connecting + %% MG. + case megaco_config:update_conn_info(CH, monitor_ref, connected) of + ok -> + ?SIM({ok, CH}, monitor_process_local); + {error, Reason} -> + disconnect(CH, {config_update, Reason}), + {error, Reason} + end. + +connect_remote(CH, ControlPid, UserMonitorPid) + when node(ControlPid) =:= node() andalso node(UserMonitorPid) =/= node() -> + case megaco_config:lookup_local_conn(CH) of + [_ConnData] -> + UserNode = node(UserMonitorPid), + M = ?MODULE, + F = disconnect_remote, + A = [CH, UserNode], + Ref = megaco_monitor:apply_at_exit(M, F, A, UserMonitorPid), + case megaco_config:connect_remote(CH, UserNode, Ref) of + ok -> + ControlMonitorPid = whereis(megaco_monitor), + ?SIM({ok, ControlMonitorPid}, connect_remote); + {error, Reason} -> + {error, Reason} + end; + [] -> + {error, {no_connection, CH}} + end. + +cancel_apply_at_exit({connecting, _ConnectorPid}) -> + ok; +cancel_apply_at_exit(connected) -> + ok; +cancel_apply_at_exit(ControlRef) -> + megaco_monitor:cancel_apply_at_exit(ControlRef). + +node_of_control_pid(Pid) when is_pid(Pid) -> + node(Pid); +node_of_control_pid(_) -> + node(). + +disconnect(ConnHandle, DiscoReason) + when is_record(ConnHandle, megaco_conn_handle) -> + case megaco_config:disconnect(ConnHandle) of + {ok, ConnData, RemoteConnData} -> + ControlRef = ConnData#conn_data.monitor_ref, + cancel_apply_at_exit(ControlRef), + handle_disconnect_callback(ConnData, DiscoReason), + ControlNode = node_of_control_pid(ConnData#conn_data.control_pid), + case ControlNode =:= node() of + true -> + %% Propagate to remote users + CancelFun = + fun(RCD) -> + UserRef = RCD#remote_conn_data.monitor_ref, + cancel_apply_at_exit(UserRef), + RCD#remote_conn_data.user_node + end, + Nodes = lists:map(CancelFun, RemoteConnData), + %% io:format("NODES: ~p~n", [Nodes]), + M = ?MODULE, + F = disconnect, + A = [ConnHandle, DiscoReason], + case rpc:multicall(Nodes, M, F, A) of + {Res, []} -> + Check = fun(ok) -> false; + ({error, {no_connection, _CH}}) -> false; + (_) -> true + end, + case lists:filter(Check, Res) of + [] -> + ok; + Bad -> + {error, {remote_disconnect_error, ConnHandle, Bad}} + end; + {_Res, Bad} -> + {error, {remote_disconnect_crash, ConnHandle, Bad}} + end; + false when (RemoteConnData =:= []) -> + %% Propagate to remote control node + M = ?MODULE, + F = disconnect_remote, + A = [DiscoReason, ConnHandle, node()], + case rpc:call(ControlNode, M, F, A) of + {badrpc, Reason} -> + {error, {'EXIT', Reason}}; + Other -> + Other + end + end; + {error, Reason} -> + {error, Reason} + end; +disconnect(BadHandle, Reason) -> + {error, {bad_conn_handle, BadHandle, Reason}}. + +disconnect_local(Reason, ConnHandle) -> + disconnect(ConnHandle, {no_controlling_process, Reason}). + +disconnect_remote(_Reason, ConnHandle, UserNode) -> + case megaco_config:disconnect_remote(ConnHandle, UserNode) of + [RCD] -> + Ref = RCD#remote_conn_data.monitor_ref, + cancel_apply_at_exit(Ref), + ok; + [] -> + {error, {no_connection, ConnHandle}} + end. + + +%%---------------------------------------------------------------------- +%% Handle incoming message +%%---------------------------------------------------------------------- + +receive_message(ReceiveHandle, ControlPid, SendHandle, Bin) -> + Extra = ?default_user_callback_extra, + receive_message(ReceiveHandle, ControlPid, SendHandle, Bin, Extra). + +receive_message(ReceiveHandle, ControlPid, SendHandle, Bin, Extra) -> + Opts = [link , {min_heap_size, 5000}], + spawn_opt(?MODULE, + process_received_message, + [ReceiveHandle, ControlPid, SendHandle, Bin, self(), Extra], Opts), + ok. + +%% This function is called via the spawn_opt function with the link +%% option, therefor the unlink before the exit. +process_received_message(ReceiveHandle, ControlPid, SendHandle, Bin, Receiver, + Extra) -> + process_received_message(ReceiveHandle, ControlPid, SendHandle, Bin, Extra), + unlink(Receiver), + exit(normal). + +process_received_message(ReceiveHandle, ControlPid, SendHandle, Bin) -> + Extra = ?default_user_callback_extra, + process_received_message(ReceiveHandle, ControlPid, SendHandle, Bin, Extra). + +process_received_message(ReceiveHandle, ControlPid, SendHandle, Bin, Extra) -> + Flag = process_flag(trap_exit, true), + case prepare_message(ReceiveHandle, SendHandle, Bin, ControlPid, Extra) of + {ok, ConnData, MegaMsg} when is_record(MegaMsg, 'MegacoMessage') -> + ?rt1(ConnData, "message prepared", [MegaMsg]), + Mess = MegaMsg#'MegacoMessage'.mess, + case Mess#'Message'.messageBody of + {transactions, Transactions} -> + {AckList, ReqList} = + prepare_trans(ConnData, Transactions, [], [], Extra), + handle_acks(AckList, Extra), + case ReqList of + [] -> + ?rt3("no transaction requests"), + ignore; + [Req|Reqs] when (ConnData#conn_data.threaded =:= true) -> + ?rt3("handle requests (spawned)"), + lists:foreach( + fun(R) -> + spawn(?MODULE, handle_request, [R, Extra]) + end, + Reqs), + handle_request(Req, Extra); + _ -> + ?rt3("handle requests"), + case handle_requests(ReqList, [], Extra) of + [] -> + ignore; + [LongRequest | More] -> + lists:foreach( + fun(LR) -> + spawn(?MODULE, handle_long_request, [LR, Extra]) + end, + More), + handle_long_request(LongRequest, Extra) + end + end; + {messageError, Error} -> + handle_message_error(ConnData, Error, Extra) + end; + {silent_fail, ConnData, {_Code, Reason, Error}} -> + ?report_debug(ConnData, Reason, [no_reply, Error]), + ignore; + {verbose_fail, ConnData, {Code, Reason, Error}} -> + ?report_debug(ConnData, Reason, [Error]), + send_message_error(ConnData, Code, Reason) + end, + process_flag(trap_exit, Flag), + ok. + +prepare_message(RH, SH, Bin, Pid, Extra) + when is_record(RH, megaco_receive_handle) andalso is_pid(Pid) -> + ?report_trace(RH, "receive bytes", [{bytes, Bin}]), + EncodingMod = RH#megaco_receive_handle.encoding_mod, + EncodingConfig = RH#megaco_receive_handle.encoding_config, + ProtVersion = RH#megaco_receive_handle.protocol_version, + case (catch EncodingMod:decode_message(EncodingConfig, ProtVersion, Bin)) of + {ok, MegaMsg} when is_record(MegaMsg, 'MegacoMessage') -> + ?report_trace(RH, "receive message", [{message, MegaMsg}]), + Mess = MegaMsg#'MegacoMessage'.mess, + RemoteMid = Mess#'Message'.mId, + Version = Mess#'Message'.version, + LocalMid = RH#megaco_receive_handle.local_mid, + CH = #megaco_conn_handle{local_mid = LocalMid, + remote_mid = RemoteMid}, + case megaco_config:lookup_local_conn(CH) of + + %% + %% Message is not of the negotiated version + %% + + [#conn_data{protocol_version = NegVersion, + strict_version = true} = ConnData] + when NegVersion =/= Version -> + %% Use already established connection, + %% but incorrect version + ?rt1(ConnData, "not negotiated version", [Version]), + Error = {error, {not_negotiated_version, + NegVersion, Version}}, + handle_syntax_error_callback(RH, ConnData, + prepare_error(Error), + Extra); + + + [ConnData] -> + + %% + %% Use an already established connection + %% + %% This *may* have been set up in the + %% "non-official" way, so we may need to + %% create the monitor to the control process + %% and store the SendHandle (which is normally + %% done when creating the "temporary" connection). + %% + + ?rt1(ConnData, "use already established connection", []), + ConnData2 = ConnData#conn_data{send_handle = SH, + control_pid = Pid, + protocol_version = Version}, + check_message_auth(CH, ConnData2, MegaMsg, Bin); + + [] -> + %% Setup a temporary connection + ?rt3("setup a temporary connection"), + case autoconnect(RH, RemoteMid, SH, Pid, Extra) of + {ok, _} -> + do_prepare_message(RH, CH, SH, MegaMsg, Pid, Bin); + {error, {already_connected, _ConnHandle}} -> + do_prepare_message(RH, CH, SH, MegaMsg, Pid, Bin); + {error, {connection_refused, ConnData, Reason}} -> + Error = prepare_error({error, {connection_refused, Reason}}), + {verbose_fail, ConnData, Error}; + {error, Reason} -> + ConnData = fake_conn_data(RH, RemoteMid, SH, Pid), + ConnData2 = ConnData#conn_data{protocol_version = Version}, + Error = prepare_error({error, Reason}), + {verbose_fail, ConnData2, Error} + end + end; + Error -> + ?rt2("decode error", [Error]), + ConnData = handle_decode_error(Error, + RH, SH, Bin, Pid, + EncodingMod, + EncodingConfig, + ProtVersion), + handle_syntax_error_callback(RH, ConnData, prepare_error(Error), Extra) + end; +prepare_message(RH, SendHandle, _Bin, ControlPid, _Extra) -> + ConnData = fake_conn_data(RH, SendHandle, ControlPid), + Error = prepare_error({'EXIT', {bad_receive_handle, RH}}), + {verbose_fail, ConnData, Error}. + + +handle_decode_error({error, {unsupported_version, _}}, + #megaco_receive_handle{local_mid = LocalMid} = RH, SH, + Bin, Pid, + EM, EC, V) -> + case (catch EM:decode_mini_message(EC, V, Bin)) of + {ok, #'MegacoMessage'{mess = #'Message'{version = _Ver, + mId = RemoteMid}}} -> + ?rt2("erroneous message received", [SH, RemoteMid, _Ver]), + CH = #megaco_conn_handle{local_mid = LocalMid, + remote_mid = RemoteMid}, + incNumErrors(CH), + %% We cannot put the version into conn-data, that will + %% make the resulting error message impossible to sent + %% (unsupported version) + case megaco_config:lookup_local_conn(CH) of + [ConnData] -> + ?rt3("known to us"), + ConnData#conn_data{send_handle = SH}; + [] -> + ?rt3("unknown to us"), + ConnData = fake_conn_data(RH, SH, Pid), + ConnData#conn_data{conn_handle = CH} + end; + + _ -> + ?rt2("erroneous message received", [SH]), + incNumErrors(), + fake_conn_data(RH, SH, Pid) + end; + +handle_decode_error(_, + #megaco_receive_handle{local_mid = LocalMid} = RH, SH, + Bin, Pid, + EM, EC, V) -> + case (catch EM:decode_mini_message(EC, V, Bin)) of + {ok, #'MegacoMessage'{mess = #'Message'{version = Ver, + mId = RemoteMid}}} -> + ?rt2("erroneous message received", [SH, Ver, RemoteMid]), + CH = #megaco_conn_handle{local_mid = LocalMid, + remote_mid = RemoteMid}, + incNumErrors(CH), + case megaco_config:lookup_local_conn(CH) of + [ConnData] -> + ?rt3("known to us"), + ConnData#conn_data{send_handle = SH, + protocol_version = Ver}; + [] -> + ?rt3("unknown to us"), + ConnData = fake_conn_data(RH, SH, Pid), + ConnData#conn_data{conn_handle = CH, + protocol_version = Ver} + end; + + _ -> + ?rt2("erroneous message received", [SH]), + incNumErrors(), + fake_conn_data(RH, SH, Pid) + end. + + +do_prepare_message(RH, CH, SendHandle, MegaMsg, ControlPid, Bin) -> + case megaco_config:lookup_local_conn(CH) of + [ConnData] -> + case check_message_auth(CH, ConnData, MegaMsg, Bin) of + {ok, ConnData2, MegaMsg} -> + %% Let the connection be permanent + {ok, ConnData2, MegaMsg}; + {ReplyTag, ConnData, Reason} -> + %% Remove the temporary connection + disconnect(CH, {bad_auth, Reason}), + {ReplyTag, ConnData, Reason} + end; + [] -> + Reason = no_connection, + disconnect(CH, Reason), + RemoteMid = CH#megaco_conn_handle.remote_mid, + ConnData = fake_conn_data(RH, RemoteMid, SendHandle, ControlPid), + Error = prepare_error({error, Reason}), + {silent_fail, ConnData, Error} + end. + +check_message_auth(_ConnHandle, ConnData, MegaMsg, Bin) -> + MsgAuth = MegaMsg#'MegacoMessage'.authHeader, + Mess = MegaMsg#'MegacoMessage'.mess, + Version = Mess#'Message'.version, + ConnData2 = ConnData#conn_data{protocol_version = Version}, + ConnAuth = ConnData2#conn_data.auth_data, + ?report_trace(ConnData2, "check message auth", [{bytes, Bin}]), + if + (MsgAuth =:= asn1_NOVALUE) andalso (ConnAuth =:= asn1_NOVALUE) -> + ?SIM({ok, ConnData2, MegaMsg}, check_message_auth); + true -> + ED = #'ErrorDescriptor'{errorCode = ?megaco_unauthorized, + errorText = "Autentication is not supported"}, + {verbose_fail, ConnData2, prepare_error({error, ED})} + end. + +handle_syntax_error_callback(ReceiveHandle, ConnData, PrepError, Extra) -> + {Code, Reason, Error} = PrepError, + ErrorDesc = #'ErrorDescriptor'{errorCode = Code, errorText = Reason}, + Version = + case Error of + {error, {unsupported_version, UV}} -> + UV; + _ -> + ConnData#conn_data.protocol_version + end, + UserMod = ConnData#conn_data.user_mod, + UserArgs = ConnData#conn_data.user_args, + ?report_trace(ReceiveHandle, "callback: syntax error", [ErrorDesc, Error]), + Args = + case Extra of + ?default_user_callback_extra -> + [ReceiveHandle, Version, ErrorDesc | UserArgs]; + _ -> + [ReceiveHandle, Version, ErrorDesc, Extra | UserArgs] + end, + Res = (catch apply(UserMod, handle_syntax_error, Args)), + ?report_debug(ReceiveHandle, "return: syntax error", + [{return, Res}, ErrorDesc, Error]), + case Res of + reply -> + {verbose_fail, ConnData, PrepError}; + {reply,#'ErrorDescriptor'{errorCode = Code1, errorText = Reason1}} -> + {verbose_fail, ConnData, {Code1,Reason1,Error}}; + no_reply -> + {silent_fail, ConnData, PrepError}; + {no_reply,#'ErrorDescriptor'{errorCode=Code2,errorText=Reason2}} -> + {silent_fail, ConnData, {Code2,Reason2,Error}}; %%% OTP-???? + _ -> + warning_msg("syntax error callback failed: ~w", [Res]), + {verbose_fail, ConnData, PrepError} + end. + +fake_conn_data(CH) when is_record(CH, megaco_conn_handle) -> + case (catch megaco_config:conn_info(CH, receive_handle)) of + RH when is_record(RH, megaco_receive_handle) -> + RemoteMid = CH#megaco_conn_handle.remote_mid, + ConnData = + fake_conn_data(RH, RemoteMid, no_send_handle, no_control_pid), + ConnData#conn_data{conn_handle = CH}; + {'EXIT', _} -> + UserMid = CH#megaco_conn_handle.local_mid, + case catch megaco_config:user_info(UserMid, receive_handle) of + {'EXIT', _} -> % No such user + #conn_data{conn_handle = CH, + serial = undefined_serial, + control_pid = no_control_pid, + monitor_ref = undefined_monitor_ref, + send_mod = no_send_mod, + send_handle = no_send_handle, + encoding_mod = no_encoding_mod, + encoding_config = no_encoding_config, + reply_action = undefined, + sent_pending_limit = infinity, + recv_pending_limit = infinity}; + RH -> + ConnData = + fake_conn_data(RH, no_send_handle, no_control_pid), + ConnData#conn_data{conn_handle = CH} + end + end. + +fake_conn_data(RH, SendHandle, ControlPid) -> + fake_conn_data(RH, unknown_remote_mid, SendHandle, ControlPid). + +fake_conn_data(RH, RemoteMid, SendHandle, ControlPid) -> + case catch megaco_config:init_conn_data(RH, RemoteMid, SendHandle, ControlPid) of + {'EXIT', _} -> % No such user + fake_user_data(RH, RemoteMid, SendHandle, ControlPid); + ConnData -> + ConnData + end. + +fake_user_data(RH, RemoteMid, SendHandle, ControlPid) -> + LocalMid = RH#megaco_receive_handle.local_mid, + RH2 = RH#megaco_receive_handle{local_mid = default}, + case catch megaco_config:init_conn_data(RH2, RemoteMid, SendHandle, ControlPid) of + {'EXIT', _} -> % Application stopped? + ConnHandle = #megaco_conn_handle{local_mid = LocalMid, + remote_mid = RemoteMid}, + EncodingMod = RH#megaco_receive_handle.encoding_mod, + EncodingConfig = RH#megaco_receive_handle.encoding_config, + SendMod = RH#megaco_receive_handle.send_mod, + #conn_data{conn_handle = ConnHandle, + serial = undefined_serial, + control_pid = ControlPid, + monitor_ref = undefined_monitor_ref, + send_mod = SendMod, + send_handle = SendHandle, + encoding_mod = EncodingMod, + encoding_config = EncodingConfig, + reply_action = undefined, + sent_pending_limit = infinity, + recv_pending_limit = infinity}; + ConnData -> + ConnData + end. + +prepare_error(Error) -> + case Error of + {error, ED} when is_record(ED, 'ErrorDescriptor') -> + Code = ED#'ErrorDescriptor'.errorCode, + Reason = ED#'ErrorDescriptor'.errorText, + {Code, Reason, Error}; + {error, [{reason, {bad_token, [BadToken, _Acc]}, Line}]} when is_integer(Line) -> + Reason = + lists:flatten( + io_lib:format("Illegal token (~p) on line ~w", [BadToken, Line])), + Code = ?megaco_bad_request, + {Code, Reason, Error}; + {error, [{reason, {bad_token, _}, Line}]} when is_integer(Line) -> + Reason = lists:concat(["Illegal token on line ", Line]), + Code = ?megaco_bad_request, + {Code, Reason, Error}; + {error, [{reason, {Line, _ParserMod, RawReasonString}} | _]} when is_integer(Line) andalso is_list(RawReasonString) -> + Reason = + case RawReasonString of + [[$s, $y, $n, $t, $a, $x | _], TokenString] -> + lists:flatten( + io_lib:format("Syntax error on line ~w before token ~s", [Line, TokenString])); + _ -> + lists:flatten(io_lib:format("Syntax error on line ~w", [Line])) + end, + Code = ?megaco_bad_request, + {Code, Reason, Error}; + {error, [{reason, {Line, _, _}} | _]} when is_integer(Line) -> + Reason = lists:concat(["Syntax error on line ", Line]), + Code = ?megaco_bad_request, + {Code, Reason, Error}; + {error, {connection_refused, ED}} when is_record(ED,'ErrorDescriptor') -> + Code = ED#'ErrorDescriptor'.errorCode, + Reason = ED#'ErrorDescriptor'.errorText, + {Code, Reason, Error}; + {error, {connection_refused, _}} -> + Reason = "Connection refused by user", + Code = ?megaco_unauthorized, + {Code, Reason, Error}; + {error, {unsupported_version, V}} -> + Reason = + lists:flatten(io_lib:format("Unsupported version: ~w",[V])), + Code = ?megaco_version_not_supported, + {Code, Reason, Error}; + {error, {not_negotiated_version, NegV, MsgV}} -> + Reason = + lists:flatten( + io_lib:format("Not negotiated version: ~w [negotiated ~w]", + [MsgV, NegV])), + Code = ?megaco_version_not_supported, + {Code, Reason, Error}; + {error, _} -> + Reason = "Syntax error", + Code = ?megaco_bad_request, + {Code, Reason, Error}; + {ok, MegaMsg} when is_record(MegaMsg, 'MegacoMessage') -> + Reason = "MID does not match config", + Code = ?megaco_incorrect_identifier, + {Code, Reason, Error}; + _ -> + Reason = "Fatal syntax error", + Code = ?megaco_internal_gateway_error, + {Code, Reason, Error} + end. + +prepare_trans(_ConnData, [], AckList, ReqList, _Extra) -> + ?SIM({AckList, ReqList}, prepare_trans_done); + +prepare_trans(ConnData, Trans, AckList, ReqList, Extra) + when ConnData#conn_data.monitor_ref =:= undefined_auto_monitor_ref -> + + ?rt3("prepare_trans - autoconnect"), + + %% <BUGBUG> + %% Do we need something here, if we send more then one + %% trans per message? + %% </BUGBUG> + + %% May occur if another process has already setup a + %% temporary connection, but the handle_connect callback + %% function has not yet returned before the eager MG + %% re-sends its initial service change message. + + prepare_autoconnecting_trans(ConnData, Trans, AckList, ReqList, Extra); + +prepare_trans(#conn_data{monitor_ref = connected} = ConnData, + Trans, AckList, ReqList, Extra) -> + + ?rt3("prepare_trans - connected"), + + %% + %% This will happen when the "MGC" user performs a "pre" connect, + %% instead of waiting for the auto-connect (which normally + %% happen when the MGC receives the first message from the + %% MG). + %% + + %% + %% The monitor_ref will have this value when the pre-connect + %% is complete, so we finish it here and then continue with the + %% normal transaction prepare. + %% + + case finish_connect(ConnData) of + {ok, CD} -> + prepare_normal_trans(CD, Trans, AckList, ReqList, Extra); + {error, Reason} -> + disconnect(ConnData#conn_data.conn_handle, Reason), + {[], []} + end; + +prepare_trans(#conn_data{monitor_ref = {connecting, _}} = _ConnData, + _Trans, _AckList, _ReqList, _Extra) -> + + ?rt3("prepare_trans - connecting"), + + %% + %% This will happen when the "MGC" user performs a "pre" connect, + %% instead of waiting for the auto-connect (which normally + %% happen when the MGC receives the first message from the + %% MG). + %% + + %% + %% The monitor_ref will have this value when the pre-connect + %% is in progress. We drop (ignore) this message and hope the + %% other side (MG) will resend. + %% + + %% prepare_connecting_trans(ConnData, Trans, AckList, ReqList, Extra); + {[], []}; + +prepare_trans(ConnData, Trans, AckList, ReqList, Extra) -> + + ?rt3("prepare_trans - normal"), + + %% Handle transaction in the normal case + + prepare_normal_trans(ConnData, Trans, AckList, ReqList, Extra). + + +prepare_autoconnecting_trans(_ConnData, [], AckList, ReqList, _Extra) -> + ?SIM({AckList, ReqList}, prepare_autoconnecting_trans_done); + +prepare_autoconnecting_trans(ConnData, [Trans | Rest], AckList, ReqList, Extra) -> + ?rt1(ConnData, "[autoconnecting] prepare trans", [Trans]), + case Trans of + {transactionRequest, T} when is_record(T, 'TransactionRequest') -> + + Serial = T#'TransactionRequest'.transactionId, + ConnData2 = ConnData#conn_data{serial = Serial}, + ?report_trace(ConnData2, "Pending handle_connect", [T]), + + %% ------------------------------------------ + %% + %% Check pending limit + %% + %% ------------------------------------------ + + Limit = ConnData#conn_data.sent_pending_limit, + TransId = to_remote_trans_id(ConnData2), + case check_and_maybe_incr_pending_limit(Limit, sent, TransId) of + ok -> + send_pending(ConnData2); + error -> + %% Pending limit: + %% In this (granted, highly hypothetical case) + %% we would make the user very confused if we + %% called the abort callback function, since + %% the request callback function has not yet + %% been called. Alas, we skip this call here. + send_pending_limit_error(ConnData); + aborted -> + ignore + end, + prepare_autoconnecting_trans(ConnData2, Rest, AckList, ReqList, + Extra); + _ -> + prepare_autoconnecting_trans(ConnData, Rest, AckList, ReqList, + Extra) + end. + + +%% ================================================================= +%% +%% Note that the TransactionReply record was changed i v3 (two +%% new fields where added), and since we don't know which version, +%% we cannot use the record definition of TransactionReply. +%% Instead we transform the record into our own internal format +%% #megaco_transaction_reply{} +%% +%% ================================================================= + +prepare_normal_trans(_ConnData, [], AckList, ReqList, _Extra) -> + ?SIM({AckList, ReqList}, prepare_normal_trans_done); + +prepare_normal_trans(ConnData, [Trans | Rest], AckList, ReqList, Extra) -> + ?rt1(ConnData, "prepare [normal] trans", [Trans]), + case Trans of + {transactionRequest, #'TransactionRequest'{transactionId = asn1_NOVALUE}} -> + ConnData2 = ConnData#conn_data{serial = 0}, + Code = ?megaco_bad_request, + Reason = "Syntax error in message: transaction id missing", + send_trans_error(ConnData2, Code, Reason), + prepare_normal_trans(ConnData2, Rest, AckList, ReqList, Extra); + {transactionRequest, T} when is_record(T, 'TransactionRequest') -> + Serial = T#'TransactionRequest'.transactionId, + ConnData2 = ConnData#conn_data{serial = Serial}, + prepare_request(ConnData2, T, Rest, AckList, ReqList, Extra); + {transactionPending, T} when is_record(T, 'TransactionPending') -> + Serial = T#'TransactionPending'.transactionId, + ConnData2 = ConnData#conn_data{serial = Serial}, + handle_pending(ConnData2, T, Extra), + prepare_normal_trans(ConnData2, Rest, AckList, ReqList, Extra); + {transactionReply, T} when is_tuple(T) andalso + (element(1, T) == 'TransactionReply') -> + T2 = transform_transaction_reply_dec(T), + Serial = T2#megaco_transaction_reply.transactionId, + ConnData2 = ConnData#conn_data{serial = Serial}, + handle_reply(ConnData2, T2, Extra), + prepare_normal_trans(ConnData2, Rest, AckList, ReqList, Extra); + {transactionResponseAck, List} when is_list(List) -> + prepare_ack(ConnData, List, Rest, AckList, ReqList, Extra); + {segmentReply, SR} when is_record(SR, 'SegmentReply') -> + handle_segment_reply(ConnData, SR, Extra), + prepare_normal_trans(ConnData, Rest, AckList, ReqList, Extra) + + end. + + +prepare_request(ConnData, T, Rest, AckList, ReqList, Extra) -> + ?rt2("prepare request", [T]), + ConnHandle = ConnData#conn_data.conn_handle, + LocalMid = ConnHandle#megaco_conn_handle.local_mid, + TransId = to_remote_trans_id(ConnData), + ?rt2("prepare request", [LocalMid, TransId]), + case megaco_monitor:lookup_reply(TransId) of + [] -> + ?rt3("brand new request"), + + %% Brand new request + + %% Check pending limit: + %% + %% We should actually check the pending limit here + %% but since we have to do it later in the + %% handle_request function (just before we call + %% the handle_trans_request callback function) we + %% can just as well wait (this is after all a very + %% unlikely case: see function prepare_trans when + %% monitor_ref == undefined_auto_monitor_ref). + %% + + #conn_data{send_handle = SendHandle, + pending_timer = InitTimer, + protocol_version = Version} = ConnData, + {WaitFor, CurrTimer} = megaco_timer:init(InitTimer), + M = ?MODULE, + F = pending_timeout, + A = [ConnHandle, TransId, CurrTimer], + PendingRef = megaco_monitor:apply_after(M, F, A, WaitFor), + Rep = #reply{send_handle = SendHandle, + trans_id = TransId, + local_mid = LocalMid, + pending_timer_ref = PendingRef, + handler = self(), + version = Version}, + case megaco_monitor:insert_reply_new(Rep) of + true -> + prepare_normal_trans(ConnData, Rest, AckList, + [{ConnData, TransId, T} | ReqList], + Extra); + false -> + %% Oups - someone got there before we did... + ?report_debug(ConnData, + "prepare request: conflicting requests", + [TransId]), + send_pending(ConnData), + megaco_monitor:cancel_apply_after(PendingRef), + prepare_normal_trans(ConnData, Rest, AckList, ReqList, + Extra) + end; + + [#reply{state = State, + handler = Pid, + pending_timer_ref = Ref} = Rep] + when (State =:= prepare) orelse (State =:= eval_request) -> + + ?rt2("request resend", [State, Pid, Ref]), + + %% Pending limit: + %% We are still preparing/evaluating the request + %% Check if the pending limit has been exceeded... + %% If the pending limit is _not_ exceeded then + %% we shall send a pending (and actually restart + %% the pending timer, but that we cannot do). + %% Don't care about Msg and Rep version diff + + #conn_data{sent_pending_limit = Limit} = ConnData, + + case check_and_maybe_incr_pending_limit(Limit, sent, TransId) of + ok -> + + %% ------------------------------------------ + %% + %% Pending limit not exceeded + %% + %% 1) Increment number of pendings sent + %% (done in the check function above) + %% 2) Send pending message + %% (We should really restart the pending + %% timer, but we have no way of doing that). + %% + %% ------------------------------------------ + + send_pending(ConnData), + prepare_normal_trans(ConnData, Rest, AckList, ReqList, + Extra); + + + error -> + + %% ------------------------------------------- + %% + %% Pending limit exceeded + %% + %% 1) Cancel pending timer + %% 2) Send 506 error message to other side + %% 3) Inform user (depends on state) + %% 4) Set reply in aborted state + %% + %% ------------------------------------------- + + %% + %% State == eval_request: + %% This means that the request is currently beeing + %% evaluated by the user, and the reply timer has + %% not yet been started. + %% Either: + %% a) The "other side" will resend (which will + %% trigger a pending message send) until we pass the + %% pending limit + %% b) We will send pending messages (when the pending + %% timer expire) until we pass the pending limit. + %% In any event, we cannot delete the reply record + %% or the pending counter in this case. Is there + %% a risk we accumulate aborted reply records? + %% + %% State == prepare: + %% The user does not know about this request + %% so we can safely perform cleanup. + %% + megaco_monitor:cancel_apply_after(Ref), + send_pending_limit_error(ConnData), + if + State == eval_request -> + %% + %% What if the user never replies? + %% In that case we will have a record + %% (and counters) that is never cleaned up... + NewFields = + [{#reply.state, aborted}, + {#reply.pending_timer_ref, undefined}], + megaco_monitor:update_reply_fields(TransId, + NewFields), + handle_request_abort_callback(ConnData, + TransId, Pid, Extra); + true -> + %% Since the user does not know about + %% this call yet, it is safe to cleanup. + %% Should we inform? + Rep2 = Rep#reply{state = aborted}, + cancel_reply(ConnData, Rep2, aborted), + ok + end, + prepare_normal_trans(ConnData, Rest, AckList, ReqList, + Extra); + + + aborted -> + + %% ------------------------------------------- + %% + %% Pending limit already exceeded + %% + %% Cleanup, just to make sure: + %% reply record & pending counter + %% + %% ------------------------------------------- + + Rep2 = Rep#reply{state = aborted}, + cancel_reply(ConnData, Rep2, aborted), + prepare_normal_trans(ConnData, Rest, AckList, ReqList, + Extra) + + end; + + [#reply{state = waiting_for_ack, + bytes = Bin, + version = Version} = Rep] -> + ?rt3("request resend when waiting for ack"), + + %% We have already sent a reply, but the receiver + %% has obviously not got it. Resend the reply but + %% don't restart the reply_timer. + ConnData2 = ConnData#conn_data{protocol_version = Version}, + ?report_trace(ConnData2, + "re-send trans reply", [T | {bytes, Bin}]), + case megaco_messenger_misc:send_message(ConnData2, true, Bin) of + {ok, _} -> + ok; + {error, Reason} -> + %% Pass it on to the user (via handle_ack) + cancel_reply(ConnData2, Rep, Reason) + end, + prepare_normal_trans(ConnData2, Rest, AckList, ReqList, Extra); + + [#reply{state = aborted} = Rep] -> + ?rt3("request resend when already in aborted state"), + + %% OTP-4956: + %% Already aborted so ignore. + %% This furthermore means that the abnoxious user at the + %% other end has already been informed (pending-limit + %% passed => error descriptor sent), but keeps sending... + %% + %% Shall we perform a cleanup? + cancel_reply(ConnData, Rep, aborted), + prepare_normal_trans(ConnData, Rest, AckList, ReqList, Extra) + end. + +prepare_ack(ConnData, [TA | T], Rest, AckList, ReqList, Extra) + when is_record(TA, 'TransactionAck') -> + First = TA#'TransactionAck'.firstAck, + Last = TA#'TransactionAck'.lastAck, + TA2 = TA#'TransactionAck'{lastAck = asn1_NOVALUE}, + ConnData2 = ConnData#conn_data{serial = First}, + AckList2 = do_prepare_ack(ConnData2, TA2, AckList), + if + Last =:= asn1_NOVALUE -> + prepare_ack(ConnData, T, Rest, AckList2, ReqList, Extra); + First < Last -> + TA3 = TA#'TransactionAck'{firstAck = First + 1}, + prepare_ack(ConnData, [TA3 | T], Rest, AckList2, ReqList, Extra); + First =:= Last -> + prepare_ack(ConnData, T, Rest, AckList2, ReqList, Extra); + First > Last -> + %% Protocol violation from the sender of this ack + ?report_important(ConnData, "<ERROR> discard trans", + [TA, {error, "firstAck > lastAck"}]), + prepare_ack(ConnData, T, Rest, AckList2, ReqList, Extra) + end; +prepare_ack(ConnData, [], Rest, AckList, ReqList, Extra) -> + prepare_normal_trans(ConnData, Rest, AckList, ReqList, Extra). + +do_prepare_ack(ConnData, T, AckList) -> + TransId = to_remote_trans_id(ConnData), + case megaco_monitor:lookup_reply(TransId) of + [] -> + %% The reply has already been garbage collected. Ignore. + ?report_trace(ConnData, "discard ack (no receiver)", [T]), + AckList; + [Rep] when Rep#reply.state =:= waiting_for_ack -> + %% Don't care about Msg and Rep version diff + [{ConnData, Rep, T} | AckList]; + [_Rep] -> + %% Protocol violation from the sender of this ack + ?report_important(ConnData, "<ERROR> discard trans", + [T, {error, "got ack before reply was sent"}]), + AckList + end. + + +increment_request_keep_alive_counter(#conn_data{conn_handle = CH}, TransId) -> + ?rt1(CH, "increment request keep alive counter", [TransId]), + megaco_config:incr_reply_counter(CH, TransId). + +create_or_maybe_increment_request_keep_alive_counter( + #conn_data{conn_handle = CH}, TransId) -> + ?rt1(CH, "create or maybe increment request keep alive counter", + [TransId]), + try + begin + megaco_config:cre_reply_counter(CH, TransId) + end + catch + _:_ -> + megaco_config:incr_reply_counter(CH, TransId) + end. + + +check_and_maybe_create_pending_limit(infinity, _, _) -> + ok; +check_and_maybe_create_pending_limit(Limit, Direction, TransId) -> + ?rt2("check and maybe create pending limit counter", + [Limit, Direction, TransId]), + try megaco_config:get_pending_counter(Direction, TransId) of + Val when Val =< Limit -> + %% Since we have no intention to increment here, it + %% is ok to be _at_ the limit + ok; + _ -> + aborted + catch + _:_ -> + %% Has not been created yet (connect). + megaco_config:cre_pending_counter(Direction, TransId, 0), + ok + end. + +%% check_and_maybe_create_pending_limit(infinity, _, _) -> +%% ok; +%% check_and_maybe_create_pending_limit(Limit, Direction, TransId) -> +%% case (catch megaco_config:get_pending_counter(Direction, TransId)) of +%% {'EXIT', _} -> +%% %% Has not been created yet (connect). +%% megaco_config:cre_pending_counter(Direction, TransId, 0), +%% ok; +%% Val when Val =< Limit -> +%% %% Since we have no intention to increment here, it +%% %% is ok to be _at_ the limit +%% ok; +%% _ -> +%% aborted +%% end. + + +check_pending_limit(infinity, _, _) -> + {ok, 0}; +check_pending_limit(Limit, Direction, TransId) -> + ?rt2("check pending limit", [Direction, Limit, TransId]), + try megaco_config:get_pending_counter(Direction, TransId) of + Val when Val =< Limit -> + %% Since we have no intention to increment here, it + %% is ok to be _at_ the limit + ?rt2("check pending limit - ok", [Val]), + {ok, Val}; + _Val -> + ?rt2("check pending limit - aborted", [_Val]), + aborted + catch + _:_ -> + %% This function is only called when we "know" the + %% counter to exist. So, the only reason that this + %% would happen is of the counter has been removed. + %% This only happen if the pending limit has been + %% reached. In any case, this is basically the same + %% as aborted! + ?rt2("check pending limit - exit", []), + aborted + end. + +%% check_pending_limit(infinity, _, _) -> +%% {ok, 0}; +%% check_pending_limit(Limit, Direction, TransId) -> +%% ?rt2("check pending limit", [Direction, Limit, TransId]), +%% case (catch megaco_config:get_pending_counter(Direction, TransId)) of +%% {'EXIT', _} -> +%% %% This function is only called when we "know" the +%% %% counter to exist. So, the only reason that this +%% %% would happen is of the counter has been removed. +%% %% This only happen if the pending limit has been +%% %% reached. In any case, this is basically the same +%% %% as aborted! +%% ?rt2("check pending limit - exit", []), +%% aborted; +%% Val when Val =< Limit -> +%% %% Since we have no intention to increment here, it +%% %% is ok to be _at_ the limit +%% ?rt2("check pending limit - ok", [Val]), +%% {ok, Val}; +%% _Val -> +%% ?rt2("check pending limit - aborted", [_Val]), +%% aborted +%% end. + + +check_and_maybe_incr_pending_limit(infinity, _, _) -> + ok; +check_and_maybe_incr_pending_limit(Limit, Direction, TransId) -> + %% + %% We need this kind of test to detect when we _pass_ the limit + %% + ?rt2("check and maybe incr pending limit", [Direction, Limit, TransId]), + try megaco_config:get_pending_counter(Direction, TransId) of + Val when Val > Limit -> + ?rt2("check and maybe incr - aborted", [Direction, Val, Limit]), + aborted; % Already passed the limit + Val -> + ?rt2("check and maybe incr - incr", [Direction, Val, Limit]), + megaco_config:incr_pending_counter(Direction, TransId), + if + Val < Limit -> + ok; % Still within the limit + true -> + ?rt2("check and maybe incr - error", + [Direction, Val, Limit]), + error % Passed the limit + end + catch + _:_ -> + %% Has not been created yet (connect). + megaco_config:cre_pending_counter(Direction, TransId, 1), + ok + end. + + +%% check_and_maybe_incr_pending_limit(infinity, _, _) -> +%% ok; +%% check_and_maybe_incr_pending_limit(Limit, Direction, TransId) -> +%% %% +%% %% We need this kind of test to detect when we _pass_ the limit +%% %% +%% ?rt2("check and maybe incr pending limit", [Direction, Limit, TransId]), +%% case (catch megaco_config:get_pending_counter(Direction, TransId)) of +%% {'EXIT', _} -> +%% %% Has not been created yet (connect). +%% megaco_config:cre_pending_counter(Direction, TransId, 1), +%% ok; +%% Val when Val > Limit -> +%% ?rt2("check and maybe incr - aborted", [Direction, Val, Limit]), +%% aborted; % Already passed the limit +%% Val -> +%% ?rt2("check and maybe incr - incr", [Direction, Val, Limit]), +%% megaco_config:incr_pending_counter(Direction, TransId), +%% if +%% Val < Limit -> +%% ok; % Still within the limit +%% true -> +%% ?rt2("check and maybe incr - error", +%% [Direction, Val, Limit]), +%% error % Passed the limit +%% end +%% end. + + +%% BUGBUG BUGBUG BUGBUG +%% +%% Do we know that the Rep is still valid? A previous transaction +%% could have taken a lot of time. +%% +handle_request({ConnData, TransId, T}, Extra) -> + case handle_request(ConnData, TransId, T, Extra) of + {pending, _RequestData} -> + handle_long_request(ConnData, TransId, T, Extra); + Else -> + Else + end. + +handle_request(ConnData, TransId, T, Extra) -> + ?report_trace(ConnData, "handle request", [TransId, T]), + + %% Pending limit: + %% Ok, before we begin, lets check that this request + %% has not been aborted. I.e. exceeded the pending + %% limit, so go check it... + + #conn_data{sent_pending_limit = Limit} = ConnData, + + case check_and_maybe_create_pending_limit(Limit, sent, TransId) of + ok -> + %% Ok so far, now update state + case megaco_monitor:update_reply_field(TransId, + #reply.state, + eval_request) of + true -> + Actions = T#'TransactionRequest'.actions, + {AckAction, SendReply} = + handle_request_callback(ConnData, TransId, Actions, + T, Extra), + + %% Next step, while we where in the callback function, + %% the pending limit could have been exceeded, so check + %% it again... + do_handle_request(AckAction, SendReply, + ConnData, TransId); + + false -> + %% Ugh? + ignore + end; + + aborted -> + %% Pending limit + %% Already exceeded the limit + %% The user does not yet know about this request, so + %% don't bother telling that it has been aborted... + %% Furthermore, the reply timer has not been started, + %% so do the cleanup now + ?rt1(ConnData, "pending limit already passed", [TransId]), + case megaco_monitor:lookup_reply(TransId) of + [Rep] -> + cancel_reply(ConnData, Rep, aborted); + _ -> + ok + end, + ignore + end. + +do_handle_request(_, ignore, _ConnData, _TransId) -> + ?rt1(_ConnData, "ignore: don't reply", [_TransId]), + ignore; +do_handle_request(_, ignore_trans_request, ConnData, TransId) -> + ?rt1(ConnData, "ignore trans request: don't reply", [TransId]), + case megaco_monitor:lookup_reply(TransId) of + [#reply{} = Rep] -> + cancel_reply(ConnData, Rep, ignore); + _ -> + ignore + end; +do_handle_request({pending, _RequestData}, {aborted, ignore}, _, _) -> + ?rt2("handle request: pending - aborted - ignore => don't reply", []), + ignore; +do_handle_request({pending, _RequestData}, {aborted, _SendReply}, _, _) -> + ?rt2("handle request: pending - aborted => don't reply", []), + ignore; +do_handle_request({pending, RequestData}, _SendReply, _ConnData, _) -> + ?rt2("handle request: pending", [RequestData]), + {pending, RequestData}; +do_handle_request(AckAction, {ok, Bin}, ConnData, TransId) + when is_binary(Bin) -> + ?rt1(ConnData, "handle request - ok", [AckAction, TransId]), + case megaco_monitor:lookup_reply(TransId) of + [#reply{pending_timer_ref = PendingRef} = Rep] -> + + #conn_data{reply_timer = InitTimer, + conn_handle = ConnHandle} = ConnData, + + %% Pending limit update: + %% - Cancel the pending timer, if running + %% - Delete the pending counter + %% + + megaco_monitor:cancel_apply_after(PendingRef), + megaco_config:del_pending_counter(sent, TransId), + + Method = timer_method(AckAction), + {WaitFor, CurrTimer} = megaco_timer:init(InitTimer), + OptBin = opt_garb_binary(CurrTimer, Bin), + M = ?MODULE, + F = reply_timeout, + A = [ConnHandle, TransId, CurrTimer], + Ref = megaco_monitor:apply_after(Method, M, F, A, + WaitFor), + Rep2 = Rep#reply{pending_timer_ref = undefined, + handler = undefined, + bytes = OptBin, + state = waiting_for_ack, + timer_ref = Ref, + ack_action = AckAction}, + megaco_monitor:insert_reply(Rep2), % Timing problem? + ignore; + + _ -> + %% Been removed already? + ignore + end; +do_handle_request(AckAction, {ok, {Sent, NotSent}}, ConnData, TransId) + when is_list(Sent) andalso is_list(NotSent) -> + ?rt1(ConnData, "handle request - ok [segmented reply]", + [AckAction, TransId]), + + case megaco_monitor:lookup_reply(TransId) of + [#reply{pending_timer_ref = PendingRef} = Rep] -> + + %% d("do_handle_request -> found reply record:" + %% "~n Rep: ~p", [Rep]), + + #conn_data{reply_timer = InitTimer, + conn_handle = ConnHandle} = ConnData, + + %% Pending limit update: + %% - Cancel the pending timer, if running + %% - Delete the pending counter + %% + + megaco_monitor:cancel_apply_after(PendingRef), + megaco_config:del_pending_counter(sent, TransId), + + Method = timer_method(AckAction), + {WaitFor, CurrTimer} = megaco_timer:init(InitTimer), + Garb = fun(Bin) -> opt_garb_binary(CurrTimer, Bin) end, + OptBins = [{SN, Garb(Bin), undefined} || {SN, Bin} <- Sent], + + M = ?MODULE, + F = reply_timeout, + A = [ConnHandle, TransId, CurrTimer], + Ref = megaco_monitor:apply_after(Method, M, F, A, WaitFor), + + Rep2 = Rep#reply{pending_timer_ref = undefined, + handler = undefined, + bytes = OptBins, + state = waiting_for_ack, + timer_ref = Ref, + ack_action = AckAction, + segments = NotSent}, + megaco_monitor:insert_reply(Rep2), % Timing problem? + + ignore; + _ -> + %% Been removed already? + ignore + end; +do_handle_request(_, {error, aborted}, ConnData, TransId) -> + ?report_trace(ConnData, "aborted during our absence", [TransId]), + case megaco_monitor:lookup_reply(TransId) of + [Rep] -> + cancel_reply(ConnData, Rep, aborted); + _ -> + ok + end, + ignore; +do_handle_request(AckAction, {error, Reason}, ConnData, TransId) -> + ?report_trace(ConnData, "error", [TransId, Reason]), + case megaco_monitor:lookup_reply(TransId) of + [Rep] -> + Rep2 = Rep#reply{state = waiting_for_ack, + ack_action = AckAction}, + cancel_reply(ConnData, Rep2, Reason); + _ -> + ok + end, + ignore; +do_handle_request(AckAction, SendReply, ConnData, TransId) -> + ?report_trace(ConnData, "unknown send trans reply result", + [TransId, AckAction, SendReply]), + ignore. + + +handle_requests([{ConnData, TransId, T} | Rest], Pending, Extra) -> + ?rt2("handle requests", [TransId]), + case handle_request(ConnData, TransId, T, Extra) of + {pending, RequestData} -> + handle_requests(Rest, [{ConnData,TransId,RequestData} | Pending], Extra); + _ -> + handle_requests(Rest, Pending, Extra) + end; +handle_requests([], Pending, _Extra) -> + ?rt2("handle requests - done", [Pending]), + Pending. + +%% opt_garb_binary(timeout, _Bin) -> garb_binary; % Need msg at restart of timer +opt_garb_binary(_Timer, Bin) -> Bin. + +timer_method(discard_ack) -> + apply_method; +timer_method(_) -> + spawn_method. + + +handle_long_request({ConnData, TransId, RequestData}, Extra) -> + + ?rt2("handle long request", [TransId, RequestData]), + + %% Pending limit: + %% We need to check the pending limit, in case it was + %% exceeded before we got this far... + %% We dont need to be able to create the counter here, + %% since that was done in the handle_request function. + + #conn_data{sent_pending_limit = Limit} = ConnData, + + case check_pending_limit(Limit, sent, TransId) of + {ok, _} -> + handle_long_request(ConnData, TransId, RequestData, Extra); + _ -> + %% Already exceeded the limit + ignore + end. + +handle_long_request(ConnData, TransId, RequestData, Extra) -> + ?report_trace(ConnData, "callback: trans long request", + [TransId, {request_data, RequestData}]), + + %% Attempt to update the handler field for this reply record + %% (if there is one). + case megaco_monitor:update_reply_field(TransId, #reply.handler, self()) of + true -> + {AckAction, Res} = + handle_long_request_callback(ConnData, TransId, + RequestData, Extra), + do_handle_long_request(AckAction, Res, ConnData, TransId); + false -> + %% Been removed already? + ignore + end. + + +do_handle_long_request(AckAction, {ok, Bin}, ConnData, TransId) -> + case megaco_monitor:lookup_reply_field(TransId, #reply.trans_id) of + {ok, _} -> + Method = timer_method(AckAction), + InitTimer = ConnData#conn_data.reply_timer, + {WaitFor, CurrTimer} = megaco_timer:init(InitTimer), + OptBin = opt_garb_binary(CurrTimer, Bin), + ConnHandle = ConnData#conn_data.conn_handle, + M = ?MODULE, + F = reply_timeout, + A = [ConnHandle, TransId, CurrTimer], + Ref = megaco_monitor:apply_after(Method, M, F, A, WaitFor), + NewFields = + [{#reply.bytes, OptBin}, + {#reply.state, waiting_for_ack}, + {#reply.timer_ref, Ref}, + {#reply.ack_action, AckAction}], + megaco_monitor:update_reply_fields(TransId, NewFields); % Timing problem? + _ -> + %% Been removed already? + ignore + end; +do_handle_long_request(_, {error, Reason}, ConnData, TransId) -> + ?report_trace(ConnData, "send trans reply", [TransId, {error, Reason}]), + ignore. + +handle_request_abort_callback(ConnData, TransId, Pid) -> + Extra = ?default_user_callback_extra, + handle_request_abort_callback(ConnData, TransId, Pid, Extra). + +handle_request_abort_callback(ConnData, TransId, Pid, Extra) -> + ?report_trace(ConnData, "callback: trans request aborted", [TransId, Pid]), + ConnHandle = ConnData#conn_data.conn_handle, + Version = ConnData#conn_data.protocol_version, + UserMod = ConnData#conn_data.user_mod, + UserArgs = ConnData#conn_data.user_args, + Serial = TransId#trans_id.serial, + Args = + case Extra of + ?default_user_callback_extra -> + [ConnHandle, Version, Serial, Pid | UserArgs]; + _ -> + [ConnHandle, Version, Serial, Pid, Extra | UserArgs] + end, + Res = (catch apply(UserMod, handle_trans_request_abort, Args)), + ?report_debug(ConnData, "return: trans request aborted", + [TransId, {return, Res}]), + case Res of + ok -> + ok; + _ -> + warning_msg("transaction request abort callback failed: ~w", + [Res]), + ok + end. + +handle_request_callback(ConnData, TransId, Actions, T, Extra) -> + ?report_trace(ConnData, "callback: trans request", [T]), + ConnHandle = ConnData#conn_data.conn_handle, + Version = ConnData#conn_data.protocol_version, + UserMod = ConnData#conn_data.user_mod, + UserArgs = ConnData#conn_data.user_args, + Args = + case Extra of + ?default_user_callback_extra -> + [ConnHandle, Version, Actions | UserArgs]; + _ -> + [ConnHandle, Version, Actions, Extra | UserArgs] + end, + Res = (catch apply(UserMod, handle_trans_request, Args)), + ?report_debug(ConnData, "return: trans request", [T, {return, Res}]), + case Res of + ignore -> %% NOTE: Only used for testing!! + {discard_ack, ignore}; + + ignore_trans_request -> + {discard_ack, ignore_trans_request}; + + {discard_ack, Replies} when is_list(Replies) -> + Reply = {actionReplies, Replies}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + [], asn1_NOVALUE), + {discard_ack, SendReply}; + {discard_ack, Error} when is_record(Error, 'ErrorDescriptor') -> + Reply = {transactionError, Error}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + [], asn1_NOVALUE), + {discard_ack, SendReply}; + {discard_ack, Replies, SendOpts} when is_list(Replies) andalso + is_list(SendOpts) -> + Reply = {actionReplies, Replies}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + SendOpts, asn1_NOVALUE), + {discard_ack, SendReply}; + {discard_ack, Error, SendOpts} + when is_record(Error, 'ErrorDescriptor') andalso + is_list(SendOpts) -> + Reply = {transactionError, Error}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + SendOpts, asn1_NOVALUE), + {discard_ack, SendReply}; + + {{handle_pending_ack, AckData}, Replies} when is_list(Replies) -> + Reply = {actionReplies, Replies}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + [], when_pending_sent), + {{handle_ack, AckData}, SendReply}; + {{handle_pending_ack, AckData}, Error} + when is_record(Error, 'ErrorDescriptor') -> + Reply = {transactionError, Error}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + [], when_pending_sent), + {{handle_ack, AckData}, SendReply}; + {{handle_pending_ack, AckData}, Replies, SendOpts} + when is_list(Replies) andalso + is_list(SendOpts) -> + Reply = {actionReplies, Replies}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + SendOpts, when_pending_sent), + {{handle_ack, AckData}, SendReply}; + {{handle_pending_ack, AckData}, Error, SendOpts} + when is_record(Error, 'ErrorDescriptor') andalso + is_list(SendOpts) -> + Reply = {transactionError, Error}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + SendOpts, when_pending_sent), + {{handle_ack, AckData}, SendReply}; + + {{handle_ack, AckData}, Replies} when is_list(Replies) -> + Reply = {actionReplies, Replies}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + [], 'NULL'), + {{handle_ack, AckData}, SendReply}; + {{handle_ack, AckData}, Error} + when is_record(Error, 'ErrorDescriptor') -> + Reply = {transactionError, Error}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + [], 'NULL'), + {{handle_ack, AckData}, SendReply}; + {{handle_ack, AckData}, Replies, SendOpts} + when is_list(Replies) andalso + is_list(SendOpts) -> + Reply = {actionReplies, Replies}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + SendOpts, 'NULL'), + {{handle_ack, AckData}, SendReply}; + {{handle_ack, AckData}, Error, SendOpts} + when is_record(Error, 'ErrorDescriptor') andalso + is_list(SendOpts) -> + Reply = {transactionError, Error}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + SendOpts, 'NULL'), + {{handle_ack, AckData}, SendReply}; + + {{handle_sloppy_ack, AckData}, Replies} when is_list(Replies) -> + Reply = {actionReplies, Replies}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + [], asn1_NOVALUE), + {{handle_ack, AckData}, SendReply}; + {{handle_sloppy_ack, AckData}, Error} + when is_record(Error, 'ErrorDescriptor') -> + Reply = {transactionError, Error}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + [], asn1_NOVALUE), + {{handle_ack, AckData}, SendReply}; + {{handle_sloppy_ack, AckData}, Replies, SendOpts} + when is_list(Replies) andalso + is_list(SendOpts) -> + Reply = {actionReplies, Replies}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + SendOpts, asn1_NOVALUE), + {{handle_ack, AckData}, SendReply}; + {{handle_sloppy_ack, AckData}, Error, SendOpts} + when is_record(Error, 'ErrorDescriptor') andalso + is_list(SendOpts) -> + Reply = {transactionError, Error}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + SendOpts, asn1_NOVALUE), + {{handle_ack, AckData}, SendReply}; + + {pending, RequestData} -> + %% The user thinks that this request will take + %% quite a while to evaluate. Maybe respond with + %% a pending trans (depends on the pending limit) + SendReply = maybe_send_pending(ConnData, TransId), + {{pending, RequestData}, SendReply}; + + Error -> + ErrorText = atom_to_list(UserMod), + ED = #'ErrorDescriptor'{ + errorCode = ?megaco_internal_gateway_error, + errorText = ErrorText}, + ?report_important(ConnData, + "callback: <ERROR> trans request", + [ED, {error, Error}]), + error_msg("transaction request callback failed: ~w", [Error]), + Reply = {transactionError, ED}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + [], asn1_NOVALUE), + {discard_ack, SendReply} + end. + +handle_long_request_callback(ConnData, TransId, RequestData, Extra) -> + ?report_trace(ConnData, "callback: trans long request", [RequestData]), + ConnHandle = ConnData#conn_data.conn_handle, + Version = ConnData#conn_data.protocol_version, + UserMod = ConnData#conn_data.user_mod, + UserArgs = ConnData#conn_data.user_args, + Args = + case Extra of + ?default_user_callback_extra -> + [ConnHandle, Version, RequestData | UserArgs]; + _ -> + [ConnHandle, Version, RequestData, Extra | UserArgs] + end, + Res = (catch apply(UserMod, handle_trans_long_request, Args)), + ?report_debug(ConnData, "return: trans long request", + [{request_data, RequestData}, {return, Res}]), + case Res of + ignore -> + {discard_ack, ignore}; + + {discard_ack, Replies} when is_list(Replies) -> + Reply = {actionReplies, Replies}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + [], asn1_NOVALUE), + {discard_ack, SendReply}; + {discard_ack, Replies, SendOpts} when is_list(Replies) andalso + is_list(SendOpts) -> + Reply = {actionReplies, Replies}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + SendOpts, asn1_NOVALUE), + {discard_ack, SendReply}; + + {{handle_ack, AckData}, Replies} when is_list(Replies) -> + Reply = {actionReplies, Replies}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + [], 'NULL'), + {{handle_ack, AckData}, SendReply}; + {{handle_ack, AckData}, Replies, SendOpts} when is_list(Replies) + andalso + is_list(SendOpts) -> + Reply = {actionReplies, Replies}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + SendOpts, 'NULL'), + {{handle_ack, AckData}, SendReply}; + + Error -> + ErrorText = atom_to_list(UserMod), + ED = #'ErrorDescriptor'{errorCode = ?megaco_internal_gateway_error, + errorText = ErrorText}, + ?report_important(ConnData, "callback: <ERROR> trans long request", + [ED, {error, Error}]), + error_msg("long transaction request callback failed: ~w", [Error]), + Reply = {transactionError, ED}, + SendReply = maybe_send_reply(ConnData, TransId, Reply, + [], asn1_NOVALUE), + {discard_ack, SendReply} + end. + +handle_pending(ConnData, T, Extra) -> + TransId = to_local_trans_id(ConnData), + ?rt2("handle pending", [T, TransId]), + case megaco_monitor:lookup_request(TransId) of + [Req] -> + + %% ------------------------------------------ + %% + %% Check received pending limit + %% + %% ------------------------------------------ + + Limit = ConnData#conn_data.recv_pending_limit, + case check_and_maybe_incr_pending_limit(Limit, + recv, TransId) of + + ok -> + %% ---------------------------------------------------- + %% + %% Received pending limit not exceeded + %% + %% ---------------------------------------------------- + + handle_recv_pending(ConnData, TransId, Req, T); + + error -> + %% ---------------------------------------------------- + %% + %% Received pending limit exceeded + %% + %% Time to give up on this transaction + %% 1) Delete request record + %% 2) Cancel timers + %% 3) Delete the (receive) pending counter + %% 4) Inform the user (handle_trans_reply) + %% + %% ---------------------------------------------------- + + handle_recv_pending_error(ConnData, TransId, Req, T, Extra); + + + aborted -> + %% ---------------------------------------------------- + %% + %% Received pending limit already exceeded + %% + %% BMK BMK BMK -- can this really happen? + %% + %% The user has already been notified about this + %% (see error above) + %% + %% ---------------------------------------------------- + + ok + + end; + + [] -> + ?report_trace(ConnData, "remote pending (no receiver)", [T]), + return_unexpected_trans(ConnData, T, Extra) + end. + +handle_recv_pending(#conn_data{long_request_resend = LRR, + conn_handle = ConnHandle} = ConnData, + TransId, + #request{timer_ref = {short, Ref}, + init_long_timer = InitTimer}, T) -> + + ?rt2("handle pending - long request", [LRR, InitTimer]), + + %% The request seems to take a while, + %% let's reset our transmission timer. + %% We now know the other side has got + %% the request and is working on it, + %% so there is no need to keep the binary + %% message for re-transmission. + + %% Start using the long timer. + %% We can now drop the "bytes", since we will + %% not resend from now on. + + megaco_monitor:cancel_apply_after(Ref), + {WaitFor, CurrTimer} = megaco_timer:init(InitTimer), + ConnHandle = ConnData#conn_data.conn_handle, + M = ?MODULE, + F = request_timeout, + A = [ConnHandle, TransId], + Ref2 = megaco_monitor:apply_after(M, F, A, WaitFor), + NewFields = + case LRR of + true -> + [{#request.timer_ref, {long, Ref2}}, + {#request.curr_timer, CurrTimer}]; + false -> + [{#request.bytes, {no_send, garb_binary}}, + {#request.timer_ref, {long, Ref2}}, + {#request.curr_timer, CurrTimer}] + end, + ?report_trace(ConnData, "trans pending (timer restarted)", [T]), + megaco_monitor:update_request_fields(TransId, NewFields); % Timing problem? + +handle_recv_pending(_ConnData, _TransId, + #request{timer_ref = {long, _Ref}, + curr_timer = timeout}, _T) -> + + ?rt3("handle pending - timeout"), + + %% The request seems to take a while, + %% let's reset our transmission timer. + %% We now know the other side has got + %% the request and is working on it, + %% so there is no need to keep the binary + %% message for re-transmission. + + %% This can happen if the timer is running for the last + %% time. I.e. next time it expires, will be the last. + %% Therefor we really do not need to do anything here. + %% The cleanup will be done in request_timeout. + + ok; + +handle_recv_pending(#conn_data{conn_handle = ConnHandle} = ConnData, TransId, + #request{timer_ref = {long, Ref}, + curr_timer = CurrTimer}, T) -> + + ?rt2("handle pending - still waiting", [CurrTimer]), + + %% The request seems to take a while, + %% let's reset our transmission timer. + %% We now know the other side has got + %% the request and is working on it, + %% so there is no need to keep the binary + %% message for re-transmission. + + %% We just need to recalculate the timer, i.e. + %% increment the timer (one "slot" has been consumed). + + megaco_monitor:cancel_apply_after(Ref), + {WaitFor, Timer2} = megaco_timer:restart(CurrTimer), + ConnHandle = ConnData#conn_data.conn_handle, + M = ?MODULE, + F = request_timeout, + A = [ConnHandle, TransId], + Ref2 = megaco_monitor:apply_after(M, F, A, WaitFor), + NewFields = + [{#request.timer_ref, {long, Ref2}}, + {#request.curr_timer, Timer2}], + ?report_trace(ConnData, + "long trans pending" + " (timer restarted)", [T]), + %% Timing problem? + megaco_monitor:update_request_fields(TransId, NewFields). + + +handle_recv_pending_error(ConnData, TransId, Req, T, Extra) -> + %% 1) Delete the request record + megaco_monitor:delete_request(TransId), + + %% 2) Possibly cancel the timer + case Req#request.timer_ref of + {_, Ref} -> + megaco_monitor:cancel_apply_after(Ref); + _ -> + ok + end, + + %% 3) Delete the (receive) pending counter + megaco_config:del_pending_counter(recv, TransId), + + %% 4) Inform the user that his/her request reached + %% the receive pending limit + UserMod = Req#request.user_mod, + UserArgs = Req#request.user_args, + Action = Req#request.reply_action, + UserData = Req#request.reply_data, + UserReply = {error, exceeded_recv_pending_limit}, + ConnData2 = ConnData#conn_data{user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData}, + + ?report_trace(ConnData, "receive pending limit reached", [T]), + return_reply(ConnData2, TransId, UserReply, Extra). + + +%% +%% This _is_ a segmented message. +%% +%% Since this is not the last segment, we shall not send any ack. +%% (even if three-way-handshake has been configured). +%% +handle_reply( + ConnData, + #megaco_transaction_reply{segmentNumber = SN, + segmentationComplete = asn1_NOVALUE} = T, Extra) + when is_integer(SN) -> + TransId = to_local_trans_id(ConnData), + ?rt2("handle segmented reply", [T, TransId, SN]), + case megaco_monitor:lookup_request(TransId) of + + %% --------------------------------------------------------- + %% The first segment, so stop the request timer. No longer + %% needed when the segment(s) start to arrive. + + [#request{timer_ref = {_Type, Ref}, + seg_recv = [], + seg_timer_ref = undefined} = Req] -> + + %% Don't care about Req and Rep version diff + ?report_trace(ConnData, "[segmented] trans reply - first seg", + [T]), + + %% Stop the request timer + megaco_monitor:cancel_apply_after(Ref), %% OTP-4843 + + %% Acknowledge the segment + send_segment_reply(ConnData, SN), + + %% First segment for this reply + NewFields = + [{#request.timer_ref, undefined}, + {#request.seg_recv, [SN]}], + megaco_monitor:update_request_fields(TransId, NewFields), + + %% Handle the reply + UserMod = Req#request.user_mod, + UserArgs = Req#request.user_args, + Action = Req#request.reply_action, + UserData = Req#request.reply_data, + UserReply = + case T#megaco_transaction_reply.transactionResult of + {transactionError, Reason} -> + {error, {SN, false, Reason}}; + {actionReplies, Replies} -> + {ok, {SN, false, Replies}} + end, + ConnData2 = ConnData#conn_data{user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData}, + return_reply(ConnData2, TransId, UserReply, Extra); + + + %% --------------------------------------------------------- + %% This is not the first segment. + %% The segment timer has not been started, so the last + %% segment have been received. + %% We must check that this is not a re-transmission! + + [#request{seg_recv = Segs, + seg_timer_ref = undefined} = Req] -> + %% Don't care about Req and Rep version diff + ?report_trace(ConnData, "[segmented] trans reply - no seg acc", + [T]), + + %% Acknowledge the segment + send_segment_reply(ConnData, SN), + + %% Updated/handle received segment + case lists:member(SN, Segs) of + true -> + %% This is a re-transmission, so we shall not pass + %% it on to the user (or update the request record). + ok; + false -> + %% First time for this segment + megaco_monitor:update_request_field(TransId, + #request.seg_recv, + [ SN | Segs ]), + + %% Handle the reply + UserMod = Req#request.user_mod, + UserArgs = Req#request.user_args, + Action = Req#request.reply_action, + UserData = Req#request.reply_data, + UserReply = + case T#megaco_transaction_reply.transactionResult of + {transactionError, Reason} -> + {error, {SN, false, Reason}}; + {actionReplies, Replies} -> + {ok, {SN, false, Replies}} + end, + ConnData2 = ConnData#conn_data{user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData}, + return_reply(ConnData2, TransId, UserReply, Extra) + + end; + + + %% --------------------------------------------------------- + %% The segment timer is running! + %% This could be the last (out-of-order) segment! + %% We must check that this is not a re-transmission! + + [#request{seg_recv = Segs, + seg_timer_ref = SegRef} = Req] -> + %% Don't care about Req and Rep version diff + ?report_trace(ConnData, "[segmented] trans reply - no seg acc", + [T]), + + %% Acknowledge the segment + send_segment_reply(ConnData, SN), + + %% Updated received segments + case lists:member(SN, Segs) of + true -> + %% This is a re-transmission + ok; + false -> + %% First time for this segment, + %% we may now have a complete set + Last = + case is_all_segments([SN | Segs]) of + {true, _Sorted} -> + megaco_monitor:cancel_apply_after(SegRef), + megaco_monitor:delete_request(TransId), + send_ack(ConnData), + true; + {false, Sorted} -> + megaco_monitor:update_request_field(TransId, + #request.seg_recv, + Sorted), + false + end, + + %% Handle the reply + UserMod = Req#request.user_mod, + UserArgs = Req#request.user_args, + Action = Req#request.reply_action, + UserData = Req#request.reply_data, + UserReply = + case T#megaco_transaction_reply.transactionResult of + {transactionError, Reason} -> + {error, {SN, Last, Reason}}; + {actionReplies, Replies} -> + {ok, {SN, Last, Replies}} + end, + ConnData2 = ConnData#conn_data{user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData}, + return_reply(ConnData2, TransId, UserReply, Extra) + + end; + + + [] -> + ?report_trace(ConnData, "trans reply (no receiver)", [T]), + return_unexpected_trans(ConnData, T, Extra) + end; + + +%% +%% This _is_ a segmented message and it's the last segment of the +%% message. +%% +handle_reply( + ConnData, + #megaco_transaction_reply{segmentNumber = SN, + segmentationComplete = 'NULL'} = T, Extra) + when is_integer(SN) -> + TransId = to_local_trans_id(ConnData), + ?rt2("handle (last) segmented reply", [T, TransId, SN]), + case megaco_monitor:lookup_request(TransId) of + + %% --------------------------------------------------------- + %% The first segment, so stop the request timer. No longer + %% needed when the segment(s) start to arrive. + + [#request{timer_ref = {_Type, Ref}, + seg_recv = [], + seg_timer_ref = undefined} = Req] -> + + %% Don't care about Req and Rep version diff + ?report_trace(ConnData, "[segmented] trans reply - " + "first/complete seg", [T]), + + %% Stop the request timer + megaco_monitor:cancel_apply_after(Ref), %% OTP-4843 + + %% Acknowledge the ("last") segment + send_segment_reply_complete(ConnData, SN), + + %% It is ofcourse pointless to split + %% a transaction into just one segment, + %% but just to be sure, we handle that + %% case also + Last = + if + SN > 1 -> + %% More then one segment + %% First time for this segment + ConnHandle = ConnData#conn_data.conn_handle, + InitSegTmr = Req#request.init_seg_timer, + {WaitFor, CurrTimer} = megaco_timer:init(InitSegTmr), + M = ?MODULE, + F = segment_timeout, + A = [ConnHandle, TransId, CurrTimer], + SegRef = + megaco_monitor:apply_after(M, F, A, WaitFor), + NewFields = + [{#request.timer_ref, undefined}, + {#request.seg_recv, [SN]}, + {#request.seg_timer_ref, SegRef}], + megaco_monitor:update_request_fields(TransId, NewFields), + false; + true -> + %% Just one segment! + megaco_monitor:delete_request(TransId), + send_ack(ConnData), + true + end, + + %% Handle the reply + UserMod = Req#request.user_mod, + UserArgs = Req#request.user_args, + Action = Req#request.reply_action, + UserData = Req#request.reply_data, + UserReply = + case T#megaco_transaction_reply.transactionResult of + {transactionError, Reason} -> + {error, {SN, Last, Reason}}; + {actionReplies, Replies} -> + {ok, {SN, Last, Replies}} + end, + ConnData2 = ConnData#conn_data{user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData}, + return_reply(ConnData2, TransId, UserReply, Extra); + + + [#request{seg_recv = Segs} = Req] -> + %% Don't care about Req and Rep version diff + ?report_trace(ConnData, "[segmented] trans reply - no seg acc", + [T]), + + %% Acknowledge the ("last") segment + send_segment_reply_complete(ConnData, SN), + + %% Updated received segments + %% This is _probably_ the last segment, but some of + %% the previous segments may have been lost, so we + %% may not have a complete set! + case lists:member(SN, Segs) of + true -> + %% This is a re-transmission + ok; + false -> + Last = + case is_all_segments([SN | Segs]) of + {true, _Sorted} -> + ?report_trace(ConnData, + "[segmented] trans reply - " + "complete set", [T]), + megaco_monitor:delete_request(TransId), + send_ack(ConnData), + true; + {false, Sorted} -> + ConnHandle = ConnData#conn_data.conn_handle, + InitSegTmr = Req#request.init_seg_timer, + {WaitFor, CurrTimer} = + megaco_timer:init(InitSegTmr), + M = ?MODULE, + F = segment_timeout, + A = [ConnHandle, TransId, CurrTimer], + SegRef = + megaco_monitor:apply_after(M, F, A, + WaitFor), + NewFields = + [{#request.seg_recv, Sorted}, + {#request.seg_timer_ref, SegRef}], + megaco_monitor:update_request_fields(TransId, NewFields), + false + end, + + %% Handle the reply + UserMod = Req#request.user_mod, + UserArgs = Req#request.user_args, + Action = Req#request.reply_action, + UserData = Req#request.reply_data, + UserReply = + case T#megaco_transaction_reply.transactionResult of + {transactionError, Reason} -> + {error, {SN, Last, Reason}}; + {actionReplies, Replies} -> + {ok, {SN, Last, Replies}} + end, + ConnData2 = ConnData#conn_data{user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData}, + return_reply(ConnData2, TransId, UserReply, Extra) + + end; + + [] -> + ?report_trace(ConnData, "trans reply (no receiver)", [T]), + return_unexpected_trans(ConnData, T, Extra) + end; + + +%% +%% This is _not_ a segmented message, +%% i.e. it's an ordinary transaction reply +%% +handle_reply(#conn_data{conn_handle = CH} = CD, T, Extra) -> + TransId = to_local_trans_id(CD), + ?rt2("handle reply", [T, TransId]), + case megaco_monitor:lookup_request(TransId) of + [Req] when (is_record(Req, request) andalso + (CD#conn_data.cancel =:= true)) -> + ?TC_AWAIT_REPLY_EVENT(true), + do_handle_reply_cancel(CD, Req, T); + + [#request{remote_mid = RMid} = Req] when ((RMid =:= preliminary_mid) orelse + (RMid =:= CH#megaco_conn_handle.remote_mid)) -> + ?TC_AWAIT_REPLY_EVENT(false), + %% Just in case conn_data got update after our lookup + %% but before we looked up the request record, we + %% check the cancel field again. + case megaco_config:conn_info(CD, cancel) of + true -> + do_handle_reply_cancel(CD, Req, T); + false -> + do_handle_reply(CD, Req, TransId, T, Extra) + end; + + [#request{user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData, + remote_mid = RMid}] -> + ?report_trace(CD, + "received trans reply with invalid remote mid", + [T, RMid]), + WrongMid = CH#megaco_conn_handle.remote_mid, + T2 = transform_transaction_reply_enc(CD#conn_data.protocol_version, + T), + UserReply = {error, {wrong_mid, WrongMid, RMid, T2}}, + CD2 = CD#conn_data{user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData}, + return_reply(CD2, TransId, UserReply, Extra); + + [] -> + ?TC_AWAIT_REPLY_EVENT(undefined), + ?report_trace(CD, "trans reply (no receiver)", [T]), + return_unexpected_trans(CD, T, Extra) + end. + +do_handle_reply_cancel(CD, #request{user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData}, T) -> + CD2 = CD#conn_data{user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData}, + return_unexpected_trans(CD2, T). + +%% Plain old handling of incomming replies +do_handle_reply(CD, + #request{timer_ref = {_Type, Ref}, % OTP-4843 + user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData, + keep_alive_timer = RKAT}, + TransId, T, Extra) + when ((RKAT =:= plain) orelse (Action =:= call)) -> + %% Don't care about Req and Rep version diff + ?report_trace(CD, "trans reply", [T]), + + %% This is the first reply (maybe of many) + megaco_monitor:delete_request(TransId), + megaco_monitor:cancel_apply_after(Ref), % OTP-4843 + megaco_config:del_pending_counter(recv, TransId), % OTP-7189 + + %% Send acknowledgement + maybe_send_ack(T#megaco_transaction_reply.immAckRequired, CD), + + UserReply = + case T#megaco_transaction_reply.transactionResult of + {transactionError, Reason} -> + {error, Reason}; + {actionReplies, Replies} -> + {ok, Replies} + end, + CD2 = CD#conn_data{user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData}, + return_reply(CD2, TransId, UserReply, Extra); + +%% This may be the first reply (of maybe many) +do_handle_reply(CD, + #request{user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData, + keep_alive_ref = undefined} = Req, + TransId, T, Extra) -> + %% Don't care about Req and Rep version diff + ?report_trace(CD, "trans reply", [T]), + + %% Could be the first reply, in which case we shall start the + %% Request Keep Alive timer... + %% This could happen for more than one (1) reply though, so + %% we need to check if the counter value actually equals one (1)! + + ReplyNo = + create_or_maybe_increment_request_keep_alive_counter(CD, TransId), + if + (ReplyNo =:= 1) -> + %% This *is* the first reply!! + %% 1) Stop resend timer + {_Type, Ref} = Req#request.timer_ref, % OTP-4843 + megaco_monitor:cancel_apply_after(Ref), % OTP-4843 + + %% 2) Delete pending counter + megaco_config:del_pending_counter(recv, TransId), % OTP-7189 + + %% 3) Start request keep alive timer + ConnHandle = CD#conn_data.conn_handle, + RKATimer = Req#request.keep_alive_timer, + {RKAWaitFor, _} = megaco_timer:init(RKATimer), + RKARef = megaco_monitor:apply_after(?MODULE, + request_keep_alive_timeout, + [ConnHandle, TransId], + RKAWaitFor), + + %% 4) Maybe send acknowledgement (three-way-handshake) + maybe_send_ack(T#megaco_transaction_reply.immAckRequired, CD), + + %% 5) And finally store the updated request record + Req2 = Req#request{keep_alive_ref = RKARef}, + megaco_monitor:insert_request(Req2); + + true -> + ok + end, + + UserReply = + case T#megaco_transaction_reply.transactionResult of + {transactionError, Reason} -> + {error, ReplyNo, Reason}; + {actionReplies, Replies} -> + {ok, ReplyNo, Replies} + end, + CD2 = CD#conn_data{user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData}, + return_reply(CD2, TransId, UserReply, Extra); + +%% This is *not* the first reply (of many) +do_handle_reply(CD, #request{user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData}, TransId, T, Extra) -> + %% Don't care about Req and Rep version diff + ?report_trace(CD, "trans reply (first reply already delivered)", [T]), + + ReplyNo = increment_request_keep_alive_counter(CD, TransId), + + UserReply = + case T#megaco_transaction_reply.transactionResult of + {transactionError, Reason} -> + {error, ReplyNo, Reason}; + {actionReplies, Replies} -> + {ok, ReplyNo, Replies} + end, + CD2 = CD#conn_data{user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData}, + return_reply(CD2, TransId, UserReply, Extra). + +is_all_segments(Segs) -> + Sorted = lists:sort(Segs), + {is_all_segments(Sorted, 1, lists:last(Sorted)), Sorted}. + +is_all_segments([Last], Last, Last) -> + true; +is_all_segments([_], _, _) -> + false; +is_all_segments([SN|Segs], SN, Last) when (SN < Last) -> + is_all_segments(Segs, SN+1, Last); +is_all_segments([SN1|_], SN2, _Last) when SN1 =/= SN2 -> + false. + + +handle_segment_reply(CD, + #'SegmentReply'{transactionId = TransId, + segmentNumber = SN, + segmentationComplete = SC}, Extra) -> + ?rt2("handle segment reply", [{trans_id, TransId}, + {segment_no, SN}, + {segmentation_complete, SC}]), + TransId2 = to_remote_trans_id(CD#conn_data{serial = TransId}), + case megaco_monitor:lookup_reply(TransId2) of + [#reply{bytes = Sent, + segments = []} = Rep] when is_list(Sent) -> + ?rt2("no unsent segments", [Sent]), + handle_segment_reply_callback(CD, TransId, SN, SC, Extra), + case lists:keysearch(SN, 1, Sent) of + {value, {SN, _Bin, SegTmr}} -> + megaco_monitor:cancel_apply_after(SegTmr), %% BMK BMK + case lists:keydelete(SN, 1, Sent) of + [] -> %% We are done + Ref = Rep#reply.timer_ref, + megaco_monitor:cancel_apply_after(Ref), + megaco_monitor:update_reply_field(TransId2, + #reply.bytes, + []), + ok; + NewSent -> + megaco_monitor:update_reply_field(TransId2, + #reply.bytes, + NewSent), + ok + end; + _ -> + ok + end; + + [#reply{bytes = Sent, + segments = NotSent}] when is_list(Sent) andalso + is_list(NotSent) -> + ?rt2("unsent segments", [Sent, NotSent]), + handle_segment_reply_callback(CD, TransId, SN, SC, Extra), + case lists:keysearch(SN, 1, Sent) of + {value, {SN, _Bin, SegTmr}} -> + megaco_monitor:cancel_apply_after(SegTmr), %% BMK BMK + NewSent = lists:keydelete(SN, 1, Sent), + [{SN2, Bin2}|NewNotSent] = NotSent, + case send_reply_segment(CD, "send trans reply segment", + SN2, Bin2) of + {ok, Bin3} -> + ?rt2("another segment sent", [Bin3]), + NewSent2 = [{SN2, Bin3, undefined}|NewSent], + NewFields = + [{#reply.bytes, NewSent2}, + {#reply.segments, NewNotSent}], + megaco_monitor:update_reply_fields(TransId2, + NewFields), + ok; + Error -> + incNumErrors(CD#conn_data.conn_handle), + ?report_important(CD, "failed sending segment", + [{segment_no, SN2}, + {error, Error}]), + error_msg("failed sending transaction reply [~w] " + "segment [~w]: ~w", + [TransId, SN2, Error]), + megaco_monitor:update_reply_field(TransId2, + #reply.bytes, + NewSent), + ok + end; + _ -> + ok + end; + + [#reply{state = State}] -> + %% We received a segment reply for a segmented reply we have + %% not yet sent? This is either some sort of race condition + %% or the "the other side" is really confused. + %% Ignore the message but issue a warning just in case... + warning_msg("received unexpected segment reply: " + "~n Transaction Id: ~p" + "~n Segment Number: ~p" + "~n Segmentation Complete: ~p" + "~n Reply state: ~p", + [TransId2, SN, SC, State]), + ignore; + + [] -> + ignore + + end. + + +%% +%% This should be passed on to the user only if the user wish it +%% (sri = segment reply indication) +%% +handle_segment_reply_callback(#conn_data{segment_reply_ind = true, + conn_handle = ConnHandle, + protocol_version = Version, + user_mod = UserMod, + user_args = UserArgs}, + TransId, SN, SC, Extra) -> + Args = + case Extra of + ?default_user_callback_extra -> + [ConnHandle, Version, TransId, SN, SC | UserArgs]; + _ -> + [ConnHandle, Version, TransId, SN, SC, Extra | UserArgs] + end, + (catch apply(UserMod, handle_segment_reply, Args)); +handle_segment_reply_callback(_CD, _TransId, _SN, _SC, _Extra) -> + ok. + + +handle_acks([{ConnData, Rep, T} | Rest], Extra) + when Rep#reply.state == waiting_for_ack -> + handle_ack(ConnData, ok, Rep, T, Extra), + handle_acks(Rest, Extra); +handle_acks([], _Extra) -> + ok. + +%% If the reply to which this is the ack was segmented, +%% then we also need to check that we have received all +%% the segment-replies. If not, an error callback call +%% shall be made instead. +handle_ack(ConnData, AckStatus, + #reply{trans_id = TransId, + bytes = Bytes, + timer_ref = ReplyRef, + pending_timer_ref = PendingRef, %% BMK Still running? + ack_action = AckAction}, T, Extra) + when is_binary(Bytes) orelse (Bytes =:= undefined) -> + handle_ack_cleanup(TransId, ReplyRef, PendingRef), + handle_ack_callback(ConnData, AckStatus, AckAction, T, Extra); + +handle_ack(ConnData, AckStatus, + #reply{trans_id = TransId, + bytes = [], + segments = [], + timer_ref = ReplyRef, + pending_timer_ref = PendingRef, %% BMK Still running? + ack_action = AckAction}, T, Extra) -> + handle_ack_cleanup(TransId, ReplyRef, PendingRef), + handle_ack_callback(ConnData, AckStatus, AckAction, T, Extra); + +handle_ack(ConnData, OrigAckStatus, + #reply{trans_id = TransId, + bytes = SegSent, + segments = NotSent, + timer_ref = ReplyRef, + pending_timer_ref = PendingRef, %% BMK Still running? + ack_action = OrigAckAction}, T, Extra) + when is_list(SegSent) andalso is_list(NotSent) -> + SN_NotAcked = [SN || {SN, _, _} <- SegSent], + SN_NotSent = [SN || {SN, _} <- NotSent], + AckStatus = {error, {segment_failure, + [{original_ack_status, OrigAckStatus}, + {segments_not_acked, SN_NotAcked}, + {segments_not_sent, SN_NotSent}]}}, + AckAction = + case OrigAckAction of + {handle_ack, _} -> + OrigAckAction; + _ -> + {handle_ack, segmented_reply} + end, + cancel_segment_timers(SegSent), + handle_ack_cleanup(TransId, ReplyRef, PendingRef), + handle_ack_callback(ConnData, AckStatus, AckAction, T, Extra). + +handle_ack_cleanup(TransId, ReplyRef, PendingRef) -> + megaco_monitor:cancel_apply_after(ReplyRef), + megaco_monitor:cancel_apply_after(PendingRef), + megaco_monitor:delete_reply(TransId), + megaco_config:del_pending_counter(sent, TransId). %% BMK: Still existing? + +cancel_segment_timers(SegSent) when is_list(SegSent) -> + Cancel = fun({_, _, Ref}) -> + megaco_monitor:cancel_apply_after(Ref) + end, + lists:foreach(Cancel, SegSent); +cancel_segment_timers(_) -> + ok. + +handle_ack_callback(_CD, ok = _AS, discard_ack = _AA, _T, _Extra) -> + ok; +handle_ack_callback(ConnData, {error, Reason}, discard_ack = AckAction, T, Extra) -> + ?report_trace(ConnData, "handle ack (no callback)", + [T, AckAction, {error, Reason}, Extra]); +handle_ack_callback(ConnData, AckStatus, {handle_ack, AckData}, T, Extra) -> + ?report_trace(ConnData, "callback: trans ack", [{ack_data, AckData}]), + ConnHandle = ConnData#conn_data.conn_handle, + Version = ConnData#conn_data.protocol_version, + UserMod = ConnData#conn_data.user_mod, + UserArgs = ConnData#conn_data.user_args, + Args = + case Extra of + ?default_user_callback_extra -> + [ConnHandle, Version, AckStatus, AckData | UserArgs]; + _ -> + [ConnHandle, Version, AckStatus, AckData, Extra | UserArgs] + end, + Res = (catch apply(UserMod, handle_trans_ack, Args)), + ?report_debug(ConnData, "return: trans ack", [T, AckData, {return, Res}]), + case Res of + ok -> + ok; + _ -> + warning_msg("transaction ack callback failed: ~w", [Res]), + ok + end, + Res. + + +handle_message_error(ConnData, _Error, _Extra) + when ConnData#conn_data.monitor_ref == undefined_monitor_ref -> + %% May occur if another process already has setup a + %% temporary connection, but the handle_connect callback + %% function has not yet returned before the eager MG + %% re-sends its initial service change message. + ignore; +handle_message_error(ConnData, Error, Extra) -> + ?report_trace(ConnData, "callback: message error", [Error]), + ConnHandle = ConnData#conn_data.conn_handle, + Version = ConnData#conn_data.protocol_version, + UserMod = ConnData#conn_data.user_mod, + UserArgs = ConnData#conn_data.user_args, + Args = + case Extra of + ?default_user_callback_extra -> + [ConnHandle, Version, Error | UserArgs]; + _ -> + [ConnHandle, Version, Error, Extra | UserArgs] + end, + Res = (catch apply(UserMod, handle_message_error, Args)), + ?report_debug(ConnData, "return: message error", [Error, {return, Res}]), + case Res of + ok -> + ok; + _ -> + warning_msg("message error callback failed: ~w", [Res]), + ok + end, + Res. + +handle_disconnect_callback(ConnData, UserReason) + when is_record(ConnData, conn_data) -> + ?report_trace(ConnData, "callback: disconnect", [{reason, UserReason}]), + ConnHandle = ConnData#conn_data.conn_handle, + Version = ConnData#conn_data.protocol_version, + UserMod = ConnData#conn_data.user_mod, + UserArgs = ConnData#conn_data.user_args, + Args = [ConnHandle, Version, UserReason | UserArgs], + Res = (catch apply(UserMod, handle_disconnect, Args)), + ?report_debug(ConnData, "return: disconnect", [{reason, UserReason}, {return, Res}]), + case Res of + ok -> + ok; + _ -> + warning_msg("disconnect callback failed: ~w", [Res]), + ok + end, + Res. + + +%%---------------------------------------------------------------------- +%% Test "outgoing" messages +%%---------------------------------------------------------------------- + +%% test_request/5 -> {MegacoMessage, EncodingRes} +%% +%% This function is only intended for testing +%% (e.g. answer the question: have I constructed a valid action request?) +%% +%% It's not exactly the same code as a call to 'call' +%% or 'cast' but close enough. +%% +test_request(ConnHandle, Actions, + Version, EncodingMod, EncodingConfig) + when is_record(ConnHandle, megaco_conn_handle) and + is_integer(Version) andalso is_atom(EncodingMod) -> + %% Create a fake conn_data structure + ConnData = #conn_data{serial = 1, + protocol_version = Version, + conn_handle = ConnHandle, + auth_data = asn1_NOVALUE, + encoding_mod = EncodingMod, + encoding_config = EncodingConfig}, + + TRs = test_req_compose_transactions(ConnData, Actions), + Body = {transactions, TRs}, + MegaMsg = megaco_messenger_misc:compose_message(ConnData, Version, Body), + EncodeRes = megaco_messenger_misc:encode_message(ConnData, MegaMsg), + {MegaMsg, EncodeRes}. + + +test_req_compose_transactions(ConnData, [A|_] = ActionsList) when is_list(A) -> + LastSerial = ConnData#conn_data.serial, + test_req_compose_transactions(LastSerial, lists:reverse(ActionsList), []); +test_req_compose_transactions(#conn_data{serial = Serial}, Actions) -> + TR = #'TransactionRequest'{transactionId = Serial, + actions = Actions}, + [{transactionRequest, TR}]. + +test_req_compose_transactions(_Serial, [], Acc) -> + lists:reverse(Acc); +test_req_compose_transactions(Serial, [A|As], Acc) -> + TR = #'TransactionRequest'{transactionId = Serial, + actions = A}, + test_req_compose_transactions(Serial, As, [{transactionRequest, TR}|Acc]). + + +test_reply(ConnHandle, Version, EncodingMod, EncodingConfig, Error) + when is_record(Error, 'ErrorDescriptor') -> + Reply = {transactionError, Error}, + test_reply_encode(ConnHandle, Version, EncodingMod, EncodingConfig, Reply); +test_reply(ConnHandle, Version, EncodingMod, EncodingConfig, Replies) + when is_list(Replies) -> + Reply = {actionReplies, Replies}, + test_reply_encode(ConnHandle, Version, EncodingMod, EncodingConfig, Reply). + +test_reply_encode(ConnHandle, Version, EncodingMod, EncodingConfig, Reply) -> + ImmAck = asn1_NOVALUE, + Serial = 1, + %% Create a fake conn_data structure + CD = #conn_data{serial = Serial, + protocol_version = Version, + conn_handle = ConnHandle, + auth_data = asn1_NOVALUE, + encoding_mod = EncodingMod, + encoding_config = EncodingConfig}, + TR0 = #megaco_transaction_reply{transactionId = Serial, + immAckRequired = ImmAck, + transactionResult = Reply}, + TR = megaco_messenger_misc:transform_transaction_reply(CD, TR0), + Body = {transactions, [{transactionReply, TR}]}, + MegaMsg = megaco_messenger_misc:compose_message(CD, Version, Body), + EncodeRes = megaco_messenger_misc:encode_message(CD, MegaMsg), + {MegaMsg, EncodeRes}. + + +%%---------------------------------------------------------------------- +%% Send (or prepare) outgoing messages +%%---------------------------------------------------------------------- + +%% Description: +%% Encode a list of actions or a list of list of actions for +%% later sending (using call or cast). +%% +%% encode_actions(CH, Acts, Opts) -> {ok, encoded_actions()} | {error, Reason} +%% CH -> connection_handle() +%% Acts -> action_reqs() | [action_reqs()] +%% action_reqs() -> [action_req()] +%% action_req() -> #'ActionRequest'{} +%% Opts -> [option()] +%% option() -> {Tab, Val} +%% Tag -> atom() +%% Val -> term() +%% encoded_actions() -> binary() | [binary()] +%% Reason -> term() +encode_actions(CH, [A|_] = ActionsList, Opts) + when is_record(CH, megaco_conn_handle) andalso is_list(A) -> + (catch encode_multi_actions(CH, ActionsList, Opts)); + +encode_actions(CH, [A|_] = Actions, Opts) + when is_record(CH, megaco_conn_handle) andalso is_tuple(A) -> + do_encode_actions(CH, Actions, Opts). + +encode_multi_actions(CH, ActionsList, Opts) -> + case prepare_req_send_options(CH, Opts) of + {ok, CD} -> + ActsList = [encode_multi_actions(CD, Acts) || Acts <- ActionsList], + {ok, ActsList}; + Error -> + Error + end. + +encode_multi_actions(CD, Actions) -> + case megaco_messenger_misc:encode_actions(CD, + "encode multi actions", + Actions) of + {ok, Bin} -> + Bin; + Error -> + throw(Error) + end. + +do_encode_actions(CH, Actions, Opts) + when is_record(CH, megaco_conn_handle) -> + case prepare_req_send_options(CH, Opts) of + {ok, CD} -> + megaco_messenger_misc:encode_actions(CD, + "encode actions", + Actions); + Error -> + Error + end. + +prepare_req_send_options(CH, Opts) -> + case megaco_config:lookup_local_conn(CH) of + [CD] -> + override_req_send_options(any, CD, Opts); + [] -> + {error, {not_found, conn_data}} + end. + + +call(ConnHandle, Actions, Options) -> + case lists:keymember(reply_data, 1, Options) of + true -> + {error, {bad_option, reply_data}}; + false -> + Self = self(), + ProxyFun = fun() -> call_proxy(Self) end, + {Proxy, MRef} = erlang:spawn_monitor(ProxyFun), + Options2 = [{reply_data, Proxy} | Options], + call_or_cast(call, ConnHandle, Actions, Options2, MRef) + end. + +cast(ConnHandle, Actions, Options) -> + call_or_cast(cast, ConnHandle, Actions, Options, undefined). + +%% In a transaction there can be several actions, so if the +%% First element of the Actions list is an ''ActionRequest'' +%% record this a list of ActionRequest's for one Transaction +%% request. If on the other hand this is not the case, then +%% the Actions list is assumed to be a list of list of +%% ActionRequest. That is, action requests for several transactions. +%% It could also be a binary or a list of binaries (if +%% the actions has already been encoded). +call_or_cast(CallOrCast, ConnHandle, [A|_] = Actions, Options, ProxyMon) + when is_tuple(A) -> + %% Just one transaction + case call_or_cast(CallOrCast, ConnHandle, [Actions], Options, ProxyMon) of + ok -> + ok; + {error, Reason} -> + {error, Reason}; + {Version, [Reply]} when is_integer(Version) -> + {Version, Reply}; + {Version, Error} when is_integer(Version) -> + {Version, Error} + end; + +call_or_cast(CallOrCast, ConnHandle, Actions, Options, ProxyMon) + when is_binary(Actions) -> + %% Just one transaction (although the actions has already been encoded) + case call_or_cast(CallOrCast, ConnHandle, [Actions], Options, ProxyMon) of + ok -> + ok; + {error, Reason} -> + {error, Reason}; + {Version, [Reply]} when is_integer(Version) -> + {Version, Reply}; + {Version, Error} when is_integer(Version) -> + {Version, Error} + end; + +call_or_cast(CallOrCast, ConnHandle, ActionsList, Options, ProxyMon) + when is_record(ConnHandle, megaco_conn_handle) -> + case prepare_req_send_options(CallOrCast, + ConnHandle, Options, ActionsList) of + {ok, ConnData} -> + ?report_trace(ConnData, "call_or_cast - options prepared", []), + case encode_requests(ConnData, ActionsList) of + {ok, TRs, BinOrBins} -> + ?report_trace(ConnData, + "call_or_cast - request encoded", []), + send_request(ConnData, ConnHandle, + TRs, CallOrCast, BinOrBins), + case CallOrCast of + call -> + TransIds = to_local_trans_id(ConnData, TRs), + wait_for_reply(ConnData, TransIds, ProxyMon); + cast -> + ok + end; + {error, Reason} -> + call_proxy_cleanup(ConnData, ProxyMon), + Version = ConnData#conn_data.protocol_version, + return_error(CallOrCast, Version, {error, Reason}) + end; + {error, Reason} -> + call_proxy_cleanup(Options, ProxyMon), + return_error(CallOrCast, 1, {error, Reason}) + end; +call_or_cast(CallOrCast, ConnHandle, _Actions, Options, ProxyMon) -> + call_proxy_cleanup(Options, ProxyMon), + return_error(CallOrCast, 1, {error, {bad_megaco_conn_handle, ConnHandle}}). + + +return_error(Action, Version, Error) -> + case Action of + call -> {Version, Error}; + cast -> Error + end. + +wait_for_reply(CD, TransIds, ProxyMon) -> + ProxyPid = CD#conn_data.reply_data, + ProxyPid ! {go, self(), CD, TransIds}, + receive + {reply, ProxyPid, Reply} -> + erlang:demonitor(ProxyMon, [flush]), + Reply; + {'DOWN', ProxyMon, process, ProxyPid, Info} -> + UserReply = {error, {call_proxy_crash, Info}}, + {CD#conn_data.protocol_version, UserReply} + end. + + +call_proxy_cleanup(#conn_data{reply_data = ProxyPid}, ProxyMon) -> + do_call_proxy_cleanup(ProxyPid, ProxyMon); +call_proxy_cleanup(Options, ProxyMon) when is_list(Options) -> + ProxyPid = + case lists:keysearch(reply_data, 1, Options) of + {value, {reply_data, Data}} -> + Data; + _ -> + undefined + end, + do_call_proxy_cleanup(ProxyPid, ProxyMon); +call_proxy_cleanup(ProxyPid, ProxyMon) -> + do_call_proxy_cleanup(ProxyPid, ProxyMon). + +do_call_proxy_cleanup(ProxyPid, ProxyMon) -> + maybe_demonitor(ProxyMon), + maybe_stop_proxy(ProxyPid), + ok. + +maybe_demonitor(undefined) -> + ok; +maybe_demonitor(Mon) -> + (catch erlang:demonitor(Mon, [flush])), + ok. + +maybe_stop_proxy(Pid) when is_pid(Pid) -> + Pid ! {stop, self()}, + ok; +maybe_stop_proxy(_) -> + ok. + + +call_proxy(Parent) -> + receive + {go, Parent, CD, TransIds} -> + call_proxy(Parent, CD, TransIds); + {stop, Parent} -> + exit(normal) + end. + +call_proxy(Parent, CD, TransIds) -> + Reply = proxy_wait_for_reply(CD, TransIds, []), + Parent ! {reply, self(), Reply}, + call_proxy_gc(CD, CD#conn_data.call_proxy_gc_timeout). + +call_proxy_gc(CD, Timeout) when (Timeout > 0) -> + T = t(), + receive + {?MODULE, TransId, Version, Result} -> % Old format + CD2 = CD#conn_data{protocol_version = Version}, + Extra = ?default_user_callback_extra, + return_unexpected_trans_reply(CD2, TransId, Result, Extra), + call_proxy_gc(CD, Timeout - (t() - T)); + + {?MODULE, TransId, Version, Result, Extra} -> + CD2 = CD#conn_data{protocol_version = Version}, + return_unexpected_trans_reply(CD2, TransId, Result, Extra), + call_proxy_gc(CD, Timeout - (t() - T)) + + after Timeout -> + exit(normal) + end; +call_proxy_gc(_CD, _Timeout) -> + exit(normal). + +proxy_wait_for_reply(_CD, [], Replies0) -> + % Make sure they come in the same order as the requests where sent + Replies1 = lists:keysort(2, Replies0), + %% Must all be the same version + [{Version, _, _}|_] = Replies1, + Replies2 = [Result || {_Version, _TransId, Result} <- Replies1], + {Version, Replies2}; +proxy_wait_for_reply(CD, TransIds, Replies) -> + receive + {?MODULE, TransId, Version, Reply} -> % Old format + {TransIds2, Replies2} = + wfr_handle_reply(CD, + TransIds, TransId, + Version, Replies, Reply), + proxy_wait_for_reply(CD, TransIds2, Replies2); + + {?MODULE, TransId, Version, Reply, Extra} -> + {TransIds2, Replies2} = + wfr_handle_reply(CD, + TransIds, TransId, + Version, Replies, Reply, Extra), + proxy_wait_for_reply(CD, TransIds2, Replies2) + end. + +wfr_handle_reply(CD, TransIds, TransId, Version, Replies, Reply) -> + Extra = ?default_user_callback_extra, + wfr_handle_reply(CD, TransIds, TransId, Version, Replies, Reply, Extra). + +wfr_handle_reply(CD, TransIds, TransId, Version, Replies, Reply, Extra) -> + %% Is this meant for us? + case lists:member(TransId, TransIds) of + true -> % Yep + wfr_update(TransIds, TransId, Version, Replies, Reply, Extra); + false -> % Nop + CD2 = CD#conn_data{protocol_version = Version}, + return_unexpected_trans_reply(CD2, TransId, Reply, Extra), + {TransIds, Replies} + end. + +wfr_mk_reply(Version, TransId, Result, ?default_user_callback_extra = _Extra) -> + {Version, TransId, Result}; +wfr_mk_reply(Version, TransId, Result0, Extra) -> + Result = list_to_tuple(lists:append(tuple_to_list(Result0), [Extra])), + {Version, TransId, Result}. + +%% Last segment of a reply +%% transactionResult "=" actionReplies +wfr_update(TransIds, TransId, Version, Results, {ok, {SegNo, Last, ARs}}, Extra) + when is_integer(SegNo) andalso (Last == true) -> + TransIds2 = lists:delete(TransId, TransIds), + case lists:keysearch(TransId, 2, Results) of + + %% All segments ok (actionReplies) + {value, {V, TransId, {ok, SegReps}}} -> + SegReps2 = lists:keysort(1, [{SegNo, ARs}|SegReps]), + Rep = wfr_mk_reply(V, TransId, {ok, SegReps2}, Extra), + Results2 = lists:keyreplace(TransId, 2, Results, Rep), + {TransIds2, Results2}; + + %% Atleast one segment error (transactionError) + {value, {V, TransId, {error, {segment, OkSegs, ErrSegs}}}} -> + OkSegs2 = lists:keysort(1, [{SegNo, ARs}|OkSegs]), + ErrSegs2 = lists:keysort(1, ErrSegs), + Error = {error, {segment, OkSegs2, ErrSegs2}}, + Rep = wfr_mk_reply(V, TransId, Error, Extra), + Results2 = lists:keyreplace(TransId, 2, Results, Rep), + {TransIds2, Results2}; + + false -> + %% First and only segment + Rep = wfr_mk_reply(Version, TransId, {ok, [{SegNo, ARs}]}, Extra), + {TransIds2, [Rep | Results]} + + end; + +%% Last segment of a reply +%% transactionResult "=" transactionError +wfr_update(TransIds, TransId, Version, Results, {error, {SegNo, Last, ED}}, Extra) + when is_integer(SegNo) andalso (Last == true) -> + TransIds2 = lists:delete(TransId, TransIds), + case lists:keysearch(TransId, 2, Results) of + + %% First segment with error (transactionError) + {value, {V, TransId, {ok, SegReps}}} -> + OkSegs = lists:keysort(1, [{SegNo, ED}|SegReps]), + ErrSegs = [{SegNo, ED}], + Error = {error, {segment, OkSegs, ErrSegs}}, + Rep = wfr_mk_reply(V, TransId, Error, Extra), + Results2 = lists:keyreplace(TransId, 2, Results, Rep), + {TransIds2, Results2}; + + %% Another segment with error (transactionError) + {value, {V, TransId, {error, {segment, OkSegs, ErrSegs}}}} -> + OkSegs2 = lists:keysort(1, OkSegs), + ErrSegs2 = lists:keysort(1, [{SegNo, ED}|ErrSegs]), + Error = {error, {segment, OkSegs2, ErrSegs2}}, + Rep = wfr_mk_reply(V, TransId, Error, Extra), + Results2 = lists:keyreplace(TransId, 2, Results, Rep), + {TransIds2, Results2}; + + false -> + %% First and only segment + OkSegs = [], + ErrSegs = [{SegNo, ED}], + Error = {error, {segment, OkSegs, ErrSegs}}, + Rep = wfr_mk_reply(Version, TransId, Error, Extra), + {TransIds2, [Rep]} + + end; + +%% One segment of a reply +%% transactionResult "=" actionReplies +wfr_update(TransIds, TransId, Version, Results, {ok, {SegNo, _Last, ARs}}, Extra) + when is_integer(SegNo) -> + case lists:keysearch(TransId, 2, Results) of + + %% All segments ok (actionReplies) + {value, {V, TransId, {ok, SegReps}}} -> + SegReps2 = [{SegNo, ARs}|SegReps], + Rep = wfr_mk_reply(V, TransId, {ok, SegReps2}, Extra), + Results2 = lists:keyreplace(TransId, 2, Results, Rep), + {TransIds, Results2}; + + %% Atleast one segment error (transactionError) + {value, {V, TransId, {error, {segment, OkSegs, ErrSegs}}}} -> + OkSegs2 = [{SegNo, ARs}|OkSegs], + Error = {error, {segment, OkSegs2, ErrSegs}}, + Rep = wfr_mk_reply(V, TransId, Error, Extra), + Results2 = lists:keyreplace(TransId, 2, Results, Rep), + {TransIds, Results2}; + + false -> + %% First and only segment + Rep = wfr_mk_reply(Version, TransId, {ok, [{SegNo, ARs}]}, Extra), + {TransIds, [Rep | Results]} + + end; + +%% One segment of a reply +%% transactionResult "=" transactionError +wfr_update(TransIds, TransId, Version, Results, {error, {SegNo, _Last, ED}}, Extra) + when is_integer(SegNo) -> + case lists:keysearch(TransId, 2, Results) of + + %% First segment with error (transactionError) + {value, {V, TransId, {ok, OkSegs}}} -> + ErrSegs = [{SegNo, ED}], + Error = {error, {segment, OkSegs, ErrSegs}}, + Rep = wfr_mk_reply(V, TransId, Error, Extra), + Results2 = lists:keyreplace(TransId, 2, Results, Rep), + {TransIds, Results2}; + + %% Another segment with error (transactionError) + {value, {V, TransId, {error, {OkSegs, ErrSegs}}}} -> + ErrSegs2 = [{SegNo, ED}|ErrSegs], + Error = {error, {segment, OkSegs, ErrSegs2}}, + Rep = wfr_mk_reply(V, TransId, Error, Extra), + Results2 = lists:keyreplace(TransId, 2, Results, Rep), + {TransIds, Results2}; + + false -> + %% First segment + OkSegs = [], + ErrSegs = [{SegNo, ED}], + Error = {error, {segment, OkSegs, ErrSegs}}, + Rep = wfr_mk_reply(Version, TransId, Error, Extra), + {TransIds, [Rep]} + + end; + +%% This means that some segments did not make it in time +wfr_update(TransIds, TransId, Version, Results, + {error, {segment_timeout, Missing}}, Extra) -> + TransIds2 = lists:delete(TransId, TransIds), + case lists:keysearch(TransId, 2, Results) of + + %% First segment with error (transactionError) + {value, {V, TransId, {ok, OkSegs}}} -> + Error = {error, {segment_timeout, Missing, OkSegs, []}}, + Rep = wfr_mk_reply(V, TransId, Error, Extra), + Results2 = lists:keyreplace(TransId, 2, Results, Rep), + {TransIds2, Results2}; + + %% Another segment with error (transactionError) + {value, {V, TransId, {error, {segment, OkSegs, ErrSegs}}}} -> + Error = {error, {segment_timeout, Missing, OkSegs, ErrSegs}}, + Rep = wfr_mk_reply(V, TransId, Error, Extra), + Results2 = lists:keyreplace(TransId, 2, Results, Rep), + {TransIds2, Results2}; + + false -> + %% First segment + Error = {error, {segment_timeout, Missing, [], []}}, + Rep = wfr_mk_reply(Version, TransId, Error, Extra), + {TransIds2, [Rep]} + + end; + +%% And all other results (presumably results without segments). +wfr_update(TransIds, TransId, Version, Results, Result, Extra) -> + TransIds2 = lists:delete(TransId, TransIds), + Results2 = [wfr_mk_reply(Version, TransId, Result, Extra)|Results], + {TransIds2, Results2}. + + +%% TransInfo is either [trans_id()] or a [trans_req()] + +%% This is the normal case where we have just one +%% transaction to be sent (using call or cast) using +%% the transaction sender process. +send_request(#conn_data{control_pid = CP, + trans_req = true, + trans_sender = Pid} = CD, + CH, [Serial], Action, [Bin]) + when is_pid(Pid) andalso + is_integer(Serial) andalso + (node(CP) =:= node()) -> + + ?report_trace(CD, + "send_request - one transaction via trans-sender", + [Serial]), + + #conn_data{request_timer = InitTimer, + long_request_timer = LongTimer} = CD, + TransId = to_local_trans_id(CH, Serial), + insert_request(CD, CH, TransId, Action, {Serial, Bin}, + InitTimer, LongTimer), + megaco_trans_sender:send_req(Pid, Serial, Bin); + +%% This is the general case where we have several transactions +%% beeing sent (using call or cast) at once using +%% the transaction sender process. +send_request(#conn_data{control_pid = CP, + trans_req = true, + trans_sender = Pid} = CD, + CH, TransInfo, Action, Bins) + when is_pid(Pid) andalso + is_list(Bins) andalso + (node(CP) =:= node()) -> + + ?report_trace(CD, + "send_request - multi transactions via trans_sender", + [TransInfo, Pid]), + + #conn_data{request_timer = InitTimer, + long_request_timer = LongTimer} = CD, + insert_requests(CD, CH, TransInfo, Action, Bins, + InitTimer, LongTimer), + megaco_trans_sender:send_reqs(Pid, TransInfo, Bins); + +%% This is the case when one or more transactions is +%% beeing sent in one message immediatelly (not using +%% the transaction sender process. E.g. the binary is +%% this encoded message. +send_request(#conn_data{control_pid = CP} = CD, + CH, TRs, Action, Bin) + when is_list(TRs) andalso + is_binary(Bin) andalso + (node(CP) =:= node()) -> + + %% d("send_request -> entry with" + %% "~n TRs: ~p", [TRs]), + + ?report_trace(CD, "send_request - multi transaction", [TRs]), + + #conn_data{request_timer = InitTimer, + long_request_timer = LongTimer} = CD, + insert_requests(CD, CH, TRs, Action, Bin, + InitTimer, LongTimer), + case megaco_messenger_misc:send_message(CD, false, Bin) of + {error, Reason} -> + cancel_requests(CD, TRs, Reason); + {ok, _} -> + ignore + end; + +%% This is the case where we are not on the node where the +%% transport process run. +send_request(#conn_data{control_pid = CP} = CD, + CH, TransInfo, Action, Bin) + when node(CP) =/= node() -> + + ?report_trace(CD, "send_request - remote", [TransInfo]), + + InitTimer = infinity, + LongTimer = infinity, + insert_requests(CD, CH, TransInfo, Action, Bin, + InitTimer, LongTimer), + Node = node(CP), + Args = [node(), CD, TransInfo, Bin], + rpc:cast(Node, ?MODULE, send_request_remote, Args). + + +insert_requests(_, _, [], _, _, _, _) -> + ok; + +insert_requests(ConnData, ConnHandle, [Serial|Serials], + Action, [Bin|Bins], InitTimer, LongTimer) + when is_integer(Serial) andalso is_binary(Bin) -> + TransId = to_local_trans_id(ConnHandle, Serial), + insert_request(ConnData, ConnHandle, + TransId, Action, Bin, InitTimer, LongTimer), + + insert_requests(ConnData, ConnHandle, Serials, Action, Bins, + InitTimer, LongTimer); + +insert_requests(ConnData, ConnHandle, + [{transactionRequest, TR}|TRs], + Action, Bin, InitTimer, LongTimer) + when is_record(TR, 'TransactionRequest') andalso is_binary(Bin) -> + #'TransactionRequest'{transactionId = Serial} = TR, + TransId = to_local_trans_id(ConnHandle, Serial), + insert_request(ConnData, ConnHandle, + TransId, Action, TR, InitTimer, LongTimer), + + insert_requests(ConnData, ConnHandle, TRs, Action, Bin, + InitTimer, LongTimer). + + +insert_request(ConnData, ConnHandle, TransId, + Action, Data, InitTimer, LongTimer) -> + #megaco_conn_handle{remote_mid = RemoteMid} = ConnHandle, + #conn_data{protocol_version = Version, + user_mod = UserMod, + user_args = UserArgs, + send_handle = SendHandle, + reply_data = ReplyData, + segment_recv_timer = InitSegTimer, + request_keep_alive_timeout = RKATimer} = ConnData, + {WaitFor, CurrTimer} = megaco_timer:init(InitTimer), + M = ?MODULE, + F = request_timeout, + A = [ConnHandle, TransId], + Ref = megaco_monitor:apply_after(M, F, A, WaitFor), + Req = #request{trans_id = TransId, + remote_mid = RemoteMid, + timer_ref = ?SIM({short, Ref}, init_request_timer), + init_timer = InitTimer, + init_long_timer = LongTimer, + curr_timer = CurrTimer, + version = Version, + bytes = {send, Data}, + send_handle = SendHandle, + user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = ReplyData, + init_seg_timer = InitSegTimer, + keep_alive_timer = RKATimer}, + megaco_monitor:insert_request(Req). % Timing problem? + + +send_request_remote(ReplyNode, ConnData, TransInfo, Bin) -> + Action = remote, + ConnHandle = ConnData#conn_data.conn_handle, + ConnData2 = ConnData#conn_data{reply_data = ReplyNode}, + send_request(ConnData2, ConnHandle, TransInfo, Action, Bin). + +prepare_req_send_options(CallOrCast, ConnHandle, Options, Actions) -> + %% Ensures that two processes cannot get same transaction id. + %% Bad send options may cause spurious transaction id to be consumed. + Incr = number_of_transactions(Actions), + case megaco_config:incr_trans_id_counter(ConnHandle, Incr) of + {ok, ConnData} -> + override_req_send_options(CallOrCast, ConnData, Options); + {error, Reason} -> + {error, Reason} + end. + +number_of_transactions([Action|_]) when is_tuple(Action) -> + 1; +number_of_transactions(ActionsList) -> + length(ActionsList). + +override_req_send_options(ReplyAction, ConnData, [{Key, Val} | Tail]) -> + case Key of + protocol_version -> + ConnData2 = ConnData#conn_data{protocol_version = Val}, + override_req_send_options(ReplyAction, ConnData2, Tail); + send_handle -> + ConnData2 = ConnData#conn_data{send_handle = Val}, + override_req_send_options(ReplyAction, ConnData2, Tail); + request_timer -> + case megaco_config:verify_val(Key, Val) of + true -> + ConnData2 = ConnData#conn_data{request_timer = Val}, + override_req_send_options(ReplyAction, ConnData2, Tail); + false -> + {error, {bad_send_option, {Key, Val}}} + end; + long_request_timer -> + case megaco_config:verify_val(Key, Val) of + true -> + ConnData2 = ConnData#conn_data{long_request_timer = Val}, + override_req_send_options(ReplyAction, ConnData2, Tail); + false -> + {error, {bad_send_option, {Key, Val}}} + end; + call_proxy_gc_timeout when (ReplyAction =:= call) orelse + (ReplyAction =:= any) -> + case megaco_config:verify_val(Key, Val) of + true -> + ConnData2 = + ConnData#conn_data{call_proxy_gc_timeout = Val}, + override_req_send_options(ReplyAction, ConnData2, Tail); + false -> + {error, {bad_send_option, {Key, Val}}} + end; + request_keep_alive_timeout when (ReplyAction =:= cast) orelse + (ReplyAction =:= any) -> + case megaco_config:verify_val(Key, Val) of + true -> + ConnData2 = + ConnData#conn_data{request_keep_alive_timeout = Val}, + override_req_send_options(ReplyAction, ConnData2, Tail); + false -> + {error, {bad_send_option, {Key, Val}}} + end; + reply_data -> + ConnData2 = ConnData#conn_data{reply_data = Val}, + override_req_send_options(ReplyAction, ConnData2, Tail); + user_mod when is_atom(Val) -> + ConnData2 = ConnData#conn_data{user_mod = Val}, + override_req_send_options(ReplyAction, ConnData2, Tail); + user_args when is_list(Val) -> + ConnData2 = ConnData#conn_data{user_args = Val}, + override_req_send_options(ReplyAction, ConnData2, Tail); + trans_req when Val =:= false -> + %% We only allow turning the transaction-sender off, since + %% the opposite (turning it on) would causing to much headake... + %% This will allow not using the transaction sender for + %% occasional messages + ConnData2 = ConnData#conn_data{trans_req = Val, + trans_sender = undefined}, + override_req_send_options(ReplyAction, ConnData2, Tail); + _Bad -> + {error, {bad_send_option, {Key, Val}}} + end; +override_req_send_options(_ReplyAction, ConnData, []) -> + {ok, ConnData}. + +override_rep_send_options(ConnData, [{Key, Val} | Tail]) -> + case Key of + protocol_version -> + ConnData2 = ConnData#conn_data{protocol_version = Val}, + override_rep_send_options(ConnData2, Tail); + send_handle -> + ConnData2 = ConnData#conn_data{send_handle = Val}, + override_rep_send_options(ConnData2, Tail); + reply_timer -> + case megaco_config:verify_val(Key, Val) of + true -> + ConnData2 = ConnData#conn_data{reply_timer = Val}, + override_rep_send_options(ConnData2, Tail); + false -> + {error, {bad_send_option, {Key, Val}}} + end; + trans_req when Val =:= false -> + %% We only allow turning the transaction-sender off, since + %% the opposite (turning it on) would causing to much headake... + %% This will allow not using the transaction sender for + %% occasional messages + ConnData2 = ConnData#conn_data{trans_req = Val, + trans_sender = undefined}, + override_rep_send_options(ConnData2, Tail); + _Bad -> + {error, {bad_send_option, {Key, Val}}} + end; +override_rep_send_options(ConnData, []) -> + {ok, ConnData}. + + +%% ---- +%% This list is allways atleast one (list of actions) long. +%% ---- +%% The proper number of transaction id numbers has already +%% been "allocated", and the connection data record is +%% updated accordingly. +encode_requests(#conn_data{trans_req = true, + trans_sender = Pid, + serial = LastSerial} = CD, ActionsList) + when is_pid(Pid) -> + (catch encode_requests(CD, LastSerial, + lists:reverse(ActionsList), [], [])); +encode_requests(#conn_data{serial = LastSerial} = CD, ActionsList) -> + %% We shall not accumulate transactions. + %% This means that we shall not encode + %% the transactions individually (and send + %% them to the sender process, which + %% accumulate transactions for later sending), + %% Instead we encode the entire message directly. + %% => We shall return one binary, containing, + %% possibly, many transactions + encode_requests_in_msg(CD, LastSerial, lists:reverse(ActionsList)). + + +%% This means that we shall compose and encode one complete +%% megaco message, containing one or more transactions. +encode_requests_in_msg(CD, LastSerial, ActionsList) -> + TRs = compose_requests_in_msg(LastSerial, ActionsList, []), + Body = {transactions, TRs}, + Res = megaco_messenger_misc:encode_body(CD, + "encode trans request(s) msg", + Body), + case Res of + {ok, Bin} -> + {ok, TRs, Bin}; + Error -> + Error + end. + +compose_requests_in_msg(_S, [], TRs) -> + TRs; +compose_requests_in_msg(Serial, [A|As], Acc) -> + TR = #'TransactionRequest'{transactionId = Serial, + actions = A}, + compose_requests_in_msg(Serial - 1, As, [{transactionRequest, TR}|Acc]). + + +%% We have done the encoding in reverse order, so there +%% is no need to reverse now. +encode_requests(_, _, [], Serials, EncodedTRs) -> + {ok, Serials, EncodedTRs}; +encode_requests(CD, Serial, [Actions|ActionsList], Serials, EncodedTRs) -> + case do_encode_request(CD, Serial, Actions) of + {ok, Bin} -> + encode_requests(CD, Serial - 1, ActionsList, + [Serial|Serials], [Bin|EncodedTRs]); + Error -> + throw(Error) + end. + + +do_encode_request(CD, Serial, Actions) -> + TR = #'TransactionRequest'{transactionId = Serial, + actions = Actions}, + megaco_messenger_misc:encode_trans_request(CD, TR). + + +imm_ack_req(Counter, when_pending_sent) when (Counter > 0) -> 'NULL'; +imm_ack_req(_Counter, when_pending_sent) -> asn1_NOVALUE; +imm_ack_req(_Counter, ImmAck) -> ImmAck. + +maybe_send_reply(#conn_data{sent_pending_limit = Limit} = ConnData, + TransId, Result, SendOpts, ImmAck) -> + + %% d("maybe_send_reply -> entry with" + %% "~n Limit: ~p" + %% "~n TransId: ~p" + %% "~n Result: ~p" + %% "~n SendOpts: ~p" + %% "~n ImmAck: ~p", [Limit, TransId, Result, SendOpts, ImmAck]), + + %% Pending limit + %% Before we can send the reply we must check that we have + %% not passed the pending limit (and sent an error message). + case check_pending_limit(Limit, sent, TransId) of + {ok, Counter} -> + case override_rep_send_options(ConnData, SendOpts) of + {ok, ConnData2} -> + send_reply(ConnData2, Result, + imm_ack_req(Counter, ImmAck)); + Error -> + Error + end; + aborted -> + {error, aborted} + end. + +encode_reply(CD, TR) -> + megaco_messenger_misc:encode_trans_reply(CD, TR). + +send_reply(#conn_data{serial = Serial, + trans_req = TransReq, + trans_sender = TransSnd} = CD, TransRes, ImmAck) -> + + %% Encapsule the transaction result into a reply message + + %% d("send_reply -> entry with" + %% "~n Serial: ~p" + %% "~n TransRes: ~p" + %% "~n ImmAck: ~p", [Serial, TransRes, ImmAck]), + + TR = #megaco_transaction_reply{transactionId = Serial, + immAckRequired = ImmAck, + transactionResult = TransRes}, + case encode_reply(CD, TR) of + {ok, Bin} when is_binary(Bin) andalso (TransReq =:= true) -> + ?rt2("send_reply - pass it on to the transaction sender", + [size(Bin)]), + megaco_trans_sender:send_reply(TransSnd, Bin), + {ok, Bin}; + + {ok, Bin} when is_binary(Bin) -> + ?rt2("send_reply - encoded", [size(Bin)]), + TraceLabel = "send trans reply", + Body = {transactions, [Bin]}, + megaco_messenger_misc:send_body(CD, TraceLabel, Body); + + {ok, Bins} when is_list(Bins) -> + ?rt2("send_reply - encoded (segmented)", [length(Bins)]), + Res = send_reply_segments(CD, Bins), + {ok, Res}; + + {error, not_implemented} -> + %% Oups, we cannot segment regardless the config, + %% so pack it all into one message and hope for + %% the best... + ?rt2("send_reply - cannot encode separate transactions", []), + TR2 = megaco_messenger_misc:transform_transaction_reply(CD, TR), + Body = {transactions, [{transactionReply, TR2}]}, + megaco_messenger_misc:send_body(CD, "encode trans reply", Body); + + {error, Reason} = Error -> + Code = ?megaco_internal_gateway_error, + Text = "encode transaction reply", + ED = #'ErrorDescriptor'{errorCode = Code, + errorText = Text}, + Res = {transactionError, ED}, + TR2 = #megaco_transaction_reply{transactionId = Serial, + transactionResult = Res}, + TR3 = megaco_messenger_misc:transform_transaction_reply(CD, TR2), + TraceLabel = "<ERROR> encode trans reply body failed", + ?report_important(CD, TraceLabel, [TR, TR3, ED, Error]), + error_msg("failed encoding transaction reply body: ~s", + [format_encode_error_reason(Reason)]), + Body = {transactions, [{transactionReply, TR3}]}, + megaco_messenger_misc:send_body(CD, TraceLabel, Body), + Error + end. + +send_reply_segments(CD, Bins) -> + TraceLabelPre = "send segmented trans reply", + (catch send_reply_segments(CD, TraceLabelPre, Bins)). + +send_reply_segments(#conn_data{segment_send = infinity} = CD, Label, Bins) -> + send_reply_segments(CD, Label, length(Bins), Bins); + +send_reply_segments(#conn_data{segment_send = K} = CD, Label, Bins) + when is_integer(K) andalso (K =< length(Bins)) -> + send_reply_segments(CD, Label, K, Bins); + +send_reply_segments(#conn_data{segment_send = K} = CD, Label, Bins) + when is_integer(K) -> + send_reply_segments(CD, Label, length(Bins), Bins). + +send_reply_segments(CD, Label, K, Bins) -> + send_reply_segments(CD, Label, K, Bins, []). + +send_reply_segments(_CD, _Label, 0, Bins, Sent) -> + ?rt2("send_reply_segments - done", [Sent, Bins]), + {Sent, Bins}; +send_reply_segments(CD, TraceLabelPre, K, [{SN, Bin}|Bins], Sent) -> + case send_reply_segment(CD, TraceLabelPre, SN, Bin) of + {ok, Bin2} -> + ?rt2("send_reply_segments - send", [K, SN]), + send_reply_segments(CD, TraceLabelPre, K-1, + Bins, [{SN, Bin2}|Sent]); + Error -> + throw(Error) + end. + +send_reply_segment(CD, TraceLabelPre, SN, Bin) -> + Label = lists:flatten(io_lib:format("~s[~w]", [TraceLabelPre, SN])), + Body = {transactions, [Bin]}, + megaco_messenger_misc:send_body(CD, Label, Body). + + +format_encode_error_reason(Reason) -> + FS = + case Reason of + {Mod, Func, [EC, Msg], {AE, CS}} when is_atom(Mod) andalso + is_atom(Func) andalso + is_list(EC) and + is_tuple(Msg) and + is_list(CS) -> + io_lib:format("~n Encode module: ~w" + "~n Func: ~w" + "~n Encode config: ~w" + "~n Message part: ~p" + "~n Actual error: ~p" + "~n Call stack: ~w", + [Mod, Func, EC, Msg, AE, CS]); + + {Mod, Func, [EC, Msg], AE} when is_atom(Mod) andalso + is_atom(Func) andalso + is_list(EC) andalso + is_tuple(Msg) -> + io_lib:format("~n Encode module: ~w" + "~n Func: ~w" + "~n Encode config: ~w" + "~n Message part: ~p" + "~n Actual error: ~p", + [Mod, Func, EC, Msg, AE]); + + {Mod, [EC, Msg], {AE, CS}} when is_atom(Mod) andalso + is_list(EC) andalso + is_tuple(Msg) andalso + is_list(CS) -> + io_lib:format("~n Encode module: ~w" + "~n Encode config: ~w" + "~n Message part: ~p" + "~n Actual error: ~p" + "~n Call stack: ~w", + [Mod, EC, Msg, AE, CS]); + + {Mod, [EC, Msg], AE} when is_atom(Mod) andalso + is_list(EC) andalso + is_tuple(Msg) -> + io_lib:format("~n Encode module: ~w" + "~n Encode config: ~w" + "~n Message part: ~p" + "~n Actual error: ~p", + [Mod, EC, Msg, AE]); + + Error -> + io_lib:format("~n ~w", [Error]) + end, + lists:flatten(FS). + + +%% Presumably the user would return immediately (with {pending, Data}) if it +%% knows or suspects a request to take a long time to process. +%% For this reason we assume that handling a resent request +%% could not have caused an update of the pending limit counter. +maybe_send_pending(#conn_data{sent_pending_limit = Limit} = ConnData, + TransId) -> + case check_and_maybe_incr_pending_limit(Limit, sent, TransId) of + ok -> + send_pending(ConnData); + error -> + SendReply = send_pending_limit_error(ConnData), + {aborted, SendReply}; + aborted -> + {aborted, ignore} + end. + + +send_pending(#conn_data{serial = Serial, + trans_req = true, + trans_sender = Pid}) -> + megaco_trans_sender:send_pending(Pid, Serial); +send_pending(#conn_data{serial = Serial} = CD) -> + %% Encapsule the transaction result into a pending message + TP = #'TransactionPending'{transactionId = Serial}, + Body = {transactions, [{transactionPending, TP}]}, + megaco_messenger_misc:send_body(CD, "send trans pending", Body). + + +maybe_send_ack('NULL', #conn_data{serial = Serial, + trans_ack = true, + trans_sender = Pid}) -> + megaco_trans_sender:send_ack_now(Pid, Serial); +maybe_send_ack('NULL', CD) -> + send_ack(CD); +maybe_send_ack(_, #conn_data{auto_ack = false}) -> + ignore; +maybe_send_ack(_, #conn_data{serial = Serial, + trans_ack = true, + trans_sender = Pid}) + when is_pid(Pid) -> + %% Send (later) via the transaction sender + megaco_trans_sender:send_ack(Pid, Serial), + ok; +maybe_send_ack(_, CD) -> + %% Send now + send_ack(CD). + + +send_ack(#conn_data{serial = Serial} = CD) -> + %% Encapsule the transaction result into a ack message + TRA = #'TransactionAck'{firstAck = Serial}, + Body = {transactions, [{transactionResponseAck, [TRA]}]}, + megaco_messenger_misc:send_body(CD, "send trans ack", Body). + + +send_segment_reply(#conn_data{serial = Serial} = CD, SegNo) -> + SR = #'SegmentReply'{transactionId = Serial, + segmentNumber = SegNo}, + Body = {transactions, [{segmentReply, SR}]}, + megaco_messenger_misc:send_body(CD, "send segment reply", Body). + +send_segment_reply(#conn_data{serial = Serial} = CD, SegNo, Complete) -> + SR = #'SegmentReply'{transactionId = Serial, + segmentNumber = SegNo, + segmentationComplete = Complete}, + Body = {transactions, [{segmentReply, SR}]}, + megaco_messenger_misc:send_body(CD, "send segment reply", Body). + +send_segment_reply_complete(CD, SegNo) -> + send_segment_reply(CD, SegNo, 'NULL'). + + +send_pending_limit_error(ConnData) -> + ?report_pending_limit_exceeded(ConnData), + Code = ?megaco_number_of_transactionpending_exceeded, + Reason = "Pending limit exceeded", + send_trans_error(ConnData, Code, Reason). + +send_trans_error(ConnData, Code, Reason) -> + %% Encapsulate the transaction error into a reply message + ED = #'ErrorDescriptor'{errorCode = Code, errorText = Reason}, + Serial = ConnData#conn_data.serial, + %% Version = ConnData#conn_data.protocol_version, + TransRes = {transactionError, ED}, + TR = #megaco_transaction_reply{transactionId = Serial, + transactionResult = TransRes}, + TR2 = megaco_messenger_misc:transform_transaction_reply(ConnData, TR), + Body = {transactions, [{transactionReply, TR2}]}, + case megaco_messenger_misc:send_body(ConnData, "send trans error", Body) of + {error, Reason2} -> + ?report_important(ConnData, + "<ERROR> failed sending transaction error", + [Body, {error, Reason2}]), + error; + _ -> + ok + end. + + +send_message_error(ConnData, Code, Reason) -> + ED = #'ErrorDescriptor'{errorCode = Code, errorText = Reason}, + Body = {messageError, ED}, + case megaco_messenger_misc:send_body(ConnData, "send trans error", Body) of + {error, Reason2} -> + ?report_important(ConnData, + "<ERROR> failed sending message error", + [Body, {error, Reason2}]), + error; + _ -> + ok + end. + + +cancel(ConnHandle, Reason) when is_record(ConnHandle, megaco_conn_handle) -> + case megaco_config:lookup_local_conn(ConnHandle) of + [CD] -> + megaco_config:update_conn_info(CD, cancel, true), + do_cancel(ConnHandle, Reason, CD#conn_data{cancel = true}), + megaco_config:update_conn_info(CD, cancel, false), + ok; + [] -> + ConnData = fake_conn_data(ConnHandle), + do_cancel(ConnHandle, Reason, ConnData) + end. + +do_cancel(ConnHandle, Reason, ConnData) -> + ?report_trace(ConnData, "cancel", [ConnHandle, Reason]), + LocalMid = ConnHandle#megaco_conn_handle.local_mid, + RemoteMid = ConnHandle#megaco_conn_handle.remote_mid, + ReqTransIdPat = #trans_id{mid = LocalMid, _ = '_'}, + ReqPat = #request{trans_id = ReqTransIdPat, + remote_mid = RemoteMid, + _ = '_'}, + CancelReq = fun(Req) -> + cancel_request(ConnData, Req, Reason), + {_Type, Ref} = Req#request.timer_ref, %% OTP-4843 + megaco_monitor:cancel_apply_after(Ref) + end, + Requests = megaco_monitor:match_requests(ReqPat), + lists:foreach(CancelReq, Requests), + RemoteMid = ConnHandle#megaco_conn_handle.remote_mid, + RepTransIdPat = #trans_id{mid = RemoteMid, _ = '_'}, % BUGBUG List here? + RepPat = #reply{trans_id = RepTransIdPat, + local_mid = LocalMid, + _ = '_'}, + CancelRep = fun(Rep) -> + cancel_reply(ConnData, Rep, Reason) + end, + Replies = megaco_monitor:match_replies(RepPat), + lists:foreach(CancelRep, Replies), + ok. + +cancel_requests(_ConnData, [], _Reason) -> + ok; +cancel_requests(ConnData, [{transactionRequest,TR}|TRs], Reason) -> + #'TransactionRequest'{transactionId = TransId0} = TR, + TransId = to_local_trans_id(ConnData#conn_data.conn_handle, TransId0), + case megaco_monitor:lookup_request(TransId) of + [] -> + ignore; + [Req] when is_record(Req, request) -> + cancel_request(ConnData, Req, Reason) + end, + cancel_requests(ConnData, TRs, Reason). + +cancel_request(ConnData, Req, Reason) -> + ?report_trace(ignore, "cancel request", [Req]), + ?TC_AWAIT_CANCEL_EVENT(), + TransId = Req#request.trans_id, + Version = Req#request.version, + UserMod = Req#request.user_mod, + UserArgs = Req#request.user_args, + Action = Req#request.reply_action, + UserData = Req#request.reply_data, + UserReply = {error, Reason}, + ConnData2 = ConnData#conn_data{protocol_version = Version, + user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData}, + cancel_request2(ConnData2, TransId, UserReply). + +cancel_request2(ConnData, TransId, UserReply) -> + megaco_monitor:delete_request(TransId), + megaco_config:del_pending_counter(recv, TransId), % OTP-7189 + Serial = TransId#trans_id.serial, + ConnData2 = ConnData#conn_data{serial = Serial}, + return_reply(ConnData2, TransId, UserReply). + + +return_reply(ConnData, TransId, UserReply) -> + Extra = ?default_user_callback_extra, + return_reply(ConnData, TransId, UserReply, Extra). + +return_reply(ConnData, TransId, UserReply, Extra) -> + ?report_trace(ConnData, "callback: trans reply", [UserReply]), + Version = ConnData#conn_data.protocol_version, + UserData = ConnData#conn_data.reply_data, + case ConnData#conn_data.reply_action of + call when is_pid(UserData) -> + ?report_trace(ConnData, "callback: (call) trans reply", + [UserReply]), + Pid = UserData, + Pid ! {?MODULE, TransId, Version, UserReply, Extra}; + cast -> + ?report_trace(ConnData, "callback: (cast) trans reply", [UserReply]), + UserMod = ConnData#conn_data.user_mod, + UserArgs = ConnData#conn_data.user_args, + ConnHandle = ConnData#conn_data.conn_handle, + Args = + case Extra of + ?default_user_callback_extra -> + [ConnHandle, Version, UserReply, UserData | UserArgs]; + _ -> + [ConnHandle, Version, UserReply, UserData, Extra | UserArgs] + end, + Res = (catch apply(UserMod, handle_trans_reply, Args)), + ?report_debug(ConnData, "return: (cast) trans reply", + [UserReply, {return, Res}]), + case Res of + ok -> + ok; + _ -> + warning_msg("transaction reply callback failed: ~w", + [Res]), + ok + end, + Res; + remote -> + ?report_trace(ConnData, "callback: (remote) trans reply", [UserReply]), + Node = UserData, + Args = [ConnData, UserReply, Extra], + rpc:cast(Node, ?MODULE, receive_reply_remote, Args) + end. + +receive_reply_remote(ConnData, UserReply) -> + Extra = ?default_user_callback_extra, + receive_reply_remote(ConnData, UserReply, Extra). + +receive_reply_remote(ConnData, UserReply, Extra) -> + TransId = to_local_trans_id(ConnData), + case (catch megaco_monitor:lookup_request(TransId)) of + [#request{timer_ref = {_Type, Ref}} = Req] -> %% OTP-4843 + %% Don't care about Req and Rep version diff + megaco_monitor:delete_request(TransId), + megaco_monitor:cancel_apply_after(Ref), % OTP-4843 + megaco_config:del_pending_counter(recv, TransId), % OTP-7189 + + UserMod = Req#request.user_mod, + UserArgs = Req#request.user_args, + Action = Req#request.reply_action, + UserData = Req#request.reply_data, + ConnData2 = ConnData#conn_data{user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData}, + return_reply(ConnData2, TransId, UserReply, Extra); + + _ -> + ?report_trace(ConnData, "remote reply (no receiver)", + [UserReply]), + return_unexpected_trans_reply(ConnData, TransId, UserReply, Extra) + end. + +cancel_reply(ConnData, #reply{state = waiting_for_ack} = Rep, Reason) -> + ?report_trace(ignore, "cancel reply [waiting_for_ack]", [Rep]), + megaco_monitor:cancel_apply_after(Rep#reply.pending_timer_ref), + Serial = (Rep#reply.trans_id)#trans_id.serial, + ConnData2 = ConnData#conn_data{serial = Serial}, + T = #'TransactionAck'{firstAck = Serial}, + Extra = ?default_user_callback_extra, + handle_ack(ConnData2, {error, Reason}, Rep, T, Extra); + +cancel_reply(_ConnData, #reply{state = aborted} = Rep, _Reason) -> + ?report_trace(ignore, "cancel reply [aborted]", [Rep]), + #reply{trans_id = TransId, + timer_ref = ReplyRef, + pending_timer_ref = PendingRef} = Rep, + megaco_monitor:delete_reply(TransId), + megaco_monitor:cancel_apply_after(ReplyRef), + megaco_monitor:cancel_apply_after(PendingRef), % Still running? + megaco_config:del_pending_counter(sent, TransId), % Still existing? + ok; + +cancel_reply(_ConnData, Rep, ignore) -> + ?report_trace(ignore, "cancel reply [ignore]", [Rep]), + #reply{trans_id = TransId, + timer_ref = ReplyRef, + pending_timer_ref = PendingRef} = Rep, + megaco_monitor:delete_reply(TransId), + megaco_monitor:cancel_apply_after(ReplyRef), + megaco_monitor:cancel_apply_after(PendingRef), % Still running? + megaco_config:del_pending_counter(sent, TransId), % Still existing? + ok; + +cancel_reply(_CD, _Rep, _Reason) -> + ok. + + +request_keep_alive_timeout(ConnHandle, TransId) -> + megaco_config:del_pending_counter(ConnHandle, TransId), + megaco_monitor:lookup_request(TransId), + ok. + + +request_timeout(ConnHandle, TransId) -> + ?rt1(ConnHandle, "request timeout", [TransId]), + case megaco_monitor:lookup_request(TransId) of + [] -> + request_not_found_ignore; + [Req] when is_record(Req, request) -> + case megaco_config:lookup_local_conn(ConnHandle) of + [CD] when (CD#conn_data.cancel =:= true) -> + cancel_in_progress_ignore; + [CD] -> + incNumTimerRecovery(ConnHandle), + do_request_timeout(ConnHandle, TransId, CD, Req); + [] when ConnHandle#megaco_conn_handle.remote_mid =:= preliminary_mid -> + %% There are two possibillities: + %% 1) The connection has just been upgraded from a + %% preliminary to a real connection. So this timeout + %% is just a glitch. E.g. between the removel of this + %% ConnHandle and the timer. + %% 2) The first message sent, the service-change, got no + %% reply (UDP without three-way-handshake). + %% And then the other side (MGC) sends a request, + %% which causes an auto-upgrade + request_timeout_upgraded(ConnHandle, Req); + [] -> + incNumTimerRecovery(ConnHandle), + ConnData = fake_conn_data(ConnHandle), + do_request_timeout(ConnHandle, TransId, ConnData, Req) + end + end. + +request_timeout_upgraded(ConnHandle, Req) -> + CD = fake_conn_data(ConnHandle), + cancel_request(CD, Req, timeout). + +do_request_timeout(ConnHandle, TransId, ConnData, + #request{curr_timer = CurrTimer} = Req) -> + + ?rt1(ConnHandle, "process request timeout", [TransId, CurrTimer]), + + SendHandle = Req#request.send_handle, + Version = Req#request.version, + ConnData2 = ConnData#conn_data{send_handle = SendHandle, + protocol_version = Version}, + case CurrTimer of + timeout -> %%%%%%% + cancel_request(ConnData2, Req, timeout), + timeout1; + + %% Restartable timer + %% (max_retries = infinity_restartable) + {_, timeout} -> + cancel_request(ConnData2, Req, timeout), + timeout2; + + Timer -> + {SendOrNoSend, Data} = Req#request.bytes, + case SendOrNoSend of + send -> + case maybe_encode(ConnData2, Data) of + {ok, Bin} -> + ?report_trace(ConnData2, "re-send trans request", + [{bytes, Bin}]), + case maybe_send_message(ConnData2, true, Bin) of + ok -> + sent1_ignore; + {ok, _} -> + sent2_ignore; + {error, Reason} -> + ?report_important(ConnData2, + "<ERROR> " + "re-send trans " + "request failed", + [{bytes, Bin}, + {error, Reason}]) + end; + + {error, Reason} -> + %% Since it was possible to encode the original + %% message this should really never happen... + ?report_important(ConnData2, + "<ERROR> " + "re-send trans request failed", + [{transaction, + Req#request.bytes}, + {error, Reason}]) + end; + no_send -> + not_sent_ok + end, + {WaitFor, Timer2} = megaco_timer:restart(Timer), + OptBin = opt_garb_binary(Timer2, Data), + {Type, _} = Req#request.timer_ref, + M = ?MODULE, + F = request_timeout, + A = [ConnHandle, TransId], + Ref2 = megaco_monitor:apply_after(M, F, A, WaitFor), + NewFields = + [{#request.bytes, {SendOrNoSend, OptBin}}, + {#request.timer_ref, {Type, Ref2}}, + {#request.curr_timer, Timer2}], + megaco_monitor:update_request_fields(TransId, NewFields), % Timing problem + {restarted, WaitFor, Timer2} + + end. + +maybe_encode(#conn_data{trans_req = false} = CD, {_Serial, Bin}) + when is_binary(Bin) -> + Body = {transactions, [{transactionRequest, Bin}]}, + megaco_messenger_misc:encode_body(CD, "encode trans request msg", Body); +maybe_encode(_CD, {_Serial, Bin} = D) when is_binary(Bin) -> + {ok, D}; +maybe_encode(#conn_data{trans_req = true, + trans_sender = Pid} = CD, + #'TransactionRequest'{transactionId = Serial} = TR) + when is_pid(Pid) -> + case megaco_messenger_misc:encode_trans_request(CD, TR) of + {ok, Bin} -> + {ok, {Serial, Bin}}; + Error -> + Error + end; +maybe_encode(CD, TR) + when is_record(TR, 'TransactionRequest') -> + Body = {transactions, [{transactionRequest, TR}]}, + megaco_messenger_misc:encode_body(CD, "encode trans request msg", Body); +maybe_encode(_CD, Trash) -> + {error, {invalid_bin, Trash}}. + +maybe_send_message(CD, Resend, Bin) when is_binary(Bin) -> + megaco_messenger_misc:send_message(CD, Resend, Bin); +maybe_send_message(#conn_data{trans_sender = Pid}, _Resend, {Serial, Bin}) + when is_pid(Pid) andalso is_integer(Serial) andalso is_binary(Bin) -> + megaco_trans_sender:send_req(Pid, Serial, Bin). + + +reply_timeout(ConnHandle, TransId, timeout) -> + handle_reply_timer_timeout(ConnHandle, TransId); + +%% This means that infinity_restartable was used for max_retries. +%% There is currently no reason to use this for the reply_timeout, +%% since there is no external event to restart the timer! +reply_timeout(ConnHandle, TransId, {_, timeout}) -> + handle_reply_timer_timeout(ConnHandle, TransId); + +reply_timeout(ConnHandle, TransId, Timer) -> + ?report_trace(ConnHandle, "reply timeout", [Timer, TransId]), + + case megaco_monitor:lookup_reply(TransId) of + [] -> + reply_not_found_ignore; + + [#reply{state = waiting_for_ack, + ack_action = {handle_ack, _}} = Rep] -> + case megaco_config:lookup_local_conn(ConnHandle) of + [CD] when (CD#conn_data.cancel =:= true) -> + cancel_in_progress_ignore; + [CD] -> + incNumTimerRecovery(ConnHandle), + do_reply_timeout(ConnHandle, TransId, CD, Timer, Rep); + [] -> + incNumTimerRecovery(ConnHandle), + CD = fake_conn_data(ConnHandle), + do_reply_timeout(ConnHandle, TransId, CD, Timer, Rep) + end; + + [#reply{state = waiting_for_ack, + bytes = Sent} = Rep] when is_list(Sent) -> + case megaco_config:lookup_local_conn(ConnHandle) of + [ConnData] -> + incNumTimerRecovery(ConnHandle), + do_reply_timeout(ConnHandle, TransId, ConnData, + Timer, Rep); + [] -> + incNumTimerRecovery(ConnHandle), + ConnData = fake_conn_data(ConnHandle), + do_reply_timeout(ConnHandle, TransId, ConnData, + Timer, Rep) + end; + + [#reply{state = waiting_for_ack} = Rep] -> + do_reply_timeout(ConnHandle, TransId, Timer, Rep); + + [#reply{state = aborted} = Rep] -> + do_reply_timeout(ConnHandle, TransId, Timer, Rep); + + _ -> + ignore + + end. + +do_reply_timeout(ConnHandle, TransId, ConnData, Timer, + #reply{send_handle = SH, + version = V, + bytes = Bytes} = Rep) when is_binary(Bytes) -> + +%% d("do_reply_timeout -> entry with" +%% "~n ConnHandle: ~p" +%% "~n TransId: ~p" +%% "~n Timer: ~p" +%% "~n Rep: ~p" +%% "~n", [ConnHandle, TransId, Timer, Rep]), + + CD = ConnData#conn_data{send_handle = SH, + protocol_version = V}, + + ?rt1(CD, "re-send trans reply", [{bytes, Bytes}]), + case megaco_messenger_misc:send_message(CD, true, Bytes) of + {ok, _} -> + ignore; + {error, Reason} -> + ?report_important(CD, "<ERROR> re-send trans reply failed", + [{bytes, Bytes}, {error, Reason}]) + end, + do_reply_timeout(ConnHandle, TransId, Timer, Rep); + +do_reply_timeout(ConnHandle, TransId, ConnData, Timer, + #reply{send_handle = SH, + version = V, + bytes = Sent} = Rep) when is_list(Sent) -> + +%% d("do_reply_timeout -> entry with" +%% "~n ConnHandle: ~p" +%% "~n TransId: ~p" +%% "~n Timer: ~p" +%% "~n Rep: ~p" +%% "~n", [ConnHandle, TransId, Timer, Rep]), + + CD = ConnData#conn_data{send_handle = SH, + protocol_version = V}, + + ReSend = + fun({SN, Bytes}) -> + ?rt1(CD, "re-send segmented trans reply", + [{segment_no, SN}, {bytes, Bytes}]), + case megaco_messenger_misc:send_message(CD, true, Bytes) of +%% ok -> +%% ignore; + {ok, _} -> + ignore; + {error, Reason} -> + ?report_important(CD, + "<ERROR> re-send segmented " + "trans reply failed", + [{segment_no, SN}, + {bytes, Bytes}, + {error, Reason}]) + end + end, + lists:foreach(ReSend, Sent), + do_reply_timeout(ConnHandle, TransId, Timer, Rep). + +do_reply_timeout(ConnHandle, TransId, Timer, #reply{bytes = Bytes}) -> + {WaitFor, Timer2} = megaco_timer:restart(Timer), + OptBin = case Bytes of + Bin when is_binary(Bin) -> + opt_garb_binary(Timer2, Bin); + Sent when is_list(Sent) -> + Garb = fun(Bin) -> opt_garb_binary(Timer2, Bin) end, + [{SN, Garb(Bin)} || {SN, Bin} <- Sent] + end, + M = ?MODULE, + F = reply_timeout, + A = [ConnHandle, TransId, Timer2], + Ref2 = megaco_monitor:apply_after(M, F, A, WaitFor), + NewFields = + [{#reply.bytes, OptBin}, + {#reply.timer_ref, Ref2}], + megaco_monitor:update_reply_fields(TransId, NewFields), % Timing problem? + {restarted, WaitFor, Timer2}. + + +handle_reply_timer_timeout(ConnHandle, TransId) -> + ?report_trace(ConnHandle, "handle reply timeout", [timeout, TransId]), + incNumTimerRecovery(ConnHandle), + %% OTP-4378 + case megaco_monitor:lookup_reply(TransId) of + [#reply{state = waiting_for_ack} = Rep] -> + Serial = (Rep#reply.trans_id)#trans_id.serial, + ConnData = + case megaco_config:lookup_local_conn(ConnHandle) of + [ConnData0] -> + ConnData0; + [] -> + fake_conn_data(ConnHandle) + end, + ConnData2 = ConnData#conn_data{serial = Serial}, + T = #'TransactionAck'{firstAck = Serial}, + Extra = ?default_user_callback_extra, + handle_ack(ConnData2, {error, timeout}, Rep, T, Extra); + [#reply{pending_timer_ref = Ref, % aborted? + bytes = SegSent}] -> % may be a binary + megaco_monitor:cancel_apply_after(Ref), + cancel_segment_timers(SegSent), + megaco_monitor:delete_reply(TransId), + megaco_config:del_pending_counter(sent, TransId); + [] -> + ignore_reply_removed + end. + +%% segment_reply_timeout(ConnHandle, TransId, SN, timeout) -> +%% ?report_trace(ConnHandle, "segment reply timeout", [timeout, SN, TransId]), +%% D = fun({_, _, SegRef}) -> +%% megaco_monitor:cancel_apply_after(SegRef) +%% end, +%% incNumTimerRecovery(ConnHandle), +%% %% OTP-4378 +%% case megaco_monitor:lookup_reply(TransId) of +%% [#reply{state = waiting_for_ack, +%% bytes = Sent} = Rep] -> +%% Serial = (Rep#reply.trans_id)#trans_id.serial, +%% ConnData = +%% case megaco_config:lookup_local_conn(ConnHandle) of +%% [ConnData0] -> +%% ConnData0; +%% [] -> +%% fake_conn_data(ConnHandle) +%% end, +%% ConnData2 = ConnData#conn_data{serial = Serial}, +%% T = #'TransactionAck'{firstAck = Serial}, +%% lists:foreach(D, Sent), +%% Extra = ?default_user_callback_extra, +%% handle_ack(ConnData2, {error, timeout}, Rep, T, Extra); +%% [#reply{pending_timer_ref = Ref, +%% bytes = Sent}] -> % aborted? +%% lists:foreach(D, Sent), +%% megaco_monitor:cancel_apply_after(Ref), +%% megaco_monitor:delete_reply(TransId), +%% megaco_config:del_pending_counter(sent, TransId); + +%% [] -> +%% ignore + +%% end. + +%% segment_reply_timeout(ConnHandle, TransId, SN, Timer) -> +%% ?report_trace(ConnHandle, "reply timeout", [Timer, SN, TransId]), + +%% %% d("reply_timeout -> entry with" +%% %% "~n ConnHandle: ~p" +%% %% "~n TransId: ~p" +%% %% "~n Timer: ~p", [ConnHandle, TransId, Timer]), + +%% case megaco_monitor:lookup_reply(TransId) of +%% [] -> +%% ignore; % Trace ?? + +%% [#reply{state = waiting_for_ack, +%% bytes = ack_action = {handle_ack, _}} = Rep] -> +%% case megaco_config:lookup_local_conn(ConnHandle) of +%% [ConnData] -> +%% incNumTimerRecovery(ConnHandle), +%% do_reply_timeout(ConnHandle, TransId, ConnData, +%% Timer, Rep); +%% [] -> +%% incNumTimerRecovery(ConnHandle), +%% ConnData = fake_conn_data(ConnHandle), +%% do_reply_timeout(ConnHandle, TransId, ConnData, +%% Timer, Rep) +%% end; + +%% [#reply{state = waiting_for_ack} = Rep] -> +%% do_reply_timeout(ConnHandle, TransId, Timer, Rep); + +%% [#reply{state = aborted} = Rep] -> +%% do_reply_timeout(ConnHandle, TransId, Timer, Rep); + +%% _ -> +%% ignore + +%% end. + + +%% This clause is to catch the timers started prior to the code-upgrade +pending_timeout(#conn_data{conn_handle = CH}, TransId, Timer) -> + ?report_trace(CH, "pending timeout(1)", [Timer, TransId]), + pending_timeout(CH, TransId, Timer); + +pending_timeout(ConnHandle, TransId, Timer) -> + ?report_trace(ConnHandle, "pending timeout(2)", [Timer, TransId]), + case megaco_config:lookup_local_conn(ConnHandle) of + [CD] when (CD#conn_data.cancel == true) -> + cancel_in_progress_ignore; + [CD] -> + Serial = TransId#trans_id.serial, + handle_pending_timeout(CD#conn_data{serial = Serial}, + TransId, Timer); + [] -> + no_such_connection_ignore + end. + +handle_pending_timeout(CD, TransId, Timer) -> + ?report_trace(CD, "handle pending timeout", []), + case megaco_monitor:lookup_reply(TransId) of + [#reply{state = State, + handler = Pid} = Rep] when (State =:= prepare) orelse + (State =:= eval_request) -> + + #conn_data{sent_pending_limit = Limit, + conn_handle = ConnHandle} = CD, + + %% ------------------------------------------ + %% + %% Check pending limit + %% + %% ------------------------------------------ + + case check_and_maybe_incr_pending_limit(Limit, sent, TransId) of + ok -> + + %% --------------------------------------------- + %% + %% 1) Send pending message + %% 2) Possibly restart the pending timer + %% + %% --------------------------------------------- + + send_pending(CD), + case Timer of + timeout -> + %% We are done + incNumTimerRecovery(ConnHandle), + timeout1; + {_, timeout} -> + %% We are done + incNumTimerRecovery(ConnHandle), + timeout2; + _ -> + {WaitFor, Timer2} = megaco_timer:restart(Timer), + M = ?MODULE, + F = pending_timeout, + A = [ConnHandle, TransId, Timer2], + PendingRef = + megaco_monitor:apply_after(M, F, A, WaitFor), + %% Timing problem? + megaco_monitor:update_reply_field(TransId, + #reply.pending_timer_ref, + PendingRef), + {restarted, WaitFor, Timer2} + end; + + + error -> + + %% ------------------------------------------ + %% + %% 1) Send 506 error message to other side + %% 2) Notify user + %% 3) Set reply data in aborted state + %% + %% ------------------------------------------- + + send_pending_limit_error(CD), + handle_request_abort_callback(CD, TransId, Pid), + %% Timing problem? + Rep2 = Rep#reply{state = aborted}, + cancel_reply(CD, Rep2, aborted), + pending_limit_error; + + + aborted -> + + %% ------------------------------------------ + %% + %% Pending limit already passed + %% + %% ------------------------------------------- + Rep2 = Rep#reply{state = aborted}, + cancel_reply(CD, Rep2, aborted), + pending_limit_aborted + + end; + [] -> + reply_not_found; % Trace ?? + + [#reply{state = waiting_for_ack}] -> + %% The reply has already been sent + %% No need for any pending trans reply + reply_has_been_sent; + + [#reply{state = aborted} = Rep] -> + %% glitch, but cleanup just the same + cancel_reply(CD, Rep, aborted), + reply_aborted_state + + end. + + +segment_timeout(ConnHandle, TransId, timeout = Timer) -> + ?report_trace(ConnHandle, "segment timeout", [TransId, Timer]), + incNumTimerRecovery(ConnHandle), + case megaco_monitor:lookup_request(TransId) of + [] -> + timeout_not_found_ignore; + + [#request{seg_recv = Segs} = Req] -> + ConnData = + case megaco_config:lookup_local_conn(ConnHandle) of + [ConnData0] -> + ConnData0; + [] -> + fake_conn_data(ConnHandle) + end, + Last = lists:last(lists:sort(Segs)), + All = lists:seq(1,Last), + case All -- Segs of + [] -> + %% The last segment has just arrived, ignore + ok; + Missing -> + %% Send the error message + Code = ?megaco_segments_not_received, + Reason = missing_to_str(Missing), + send_message_error(ConnData, Code, Reason), + + %% Report to the user + UserMod = Req#request.user_mod, + UserArgs = Req#request.user_args, + Action = Req#request.reply_action, + UserData = Req#request.reply_data, + UserReply = {error, {segment_timeout, Missing}}, + ConnData2 = ConnData#conn_data{user_mod = UserMod, + user_args = UserArgs, + reply_action = Action, + reply_data = UserData}, + return_reply(ConnData2, TransId, UserReply) + end + end; + +segment_timeout(ConnHandle, TransId, Timer) -> + ?report_trace(ConnHandle, "segment timeout", [TransId, Timer]), + case megaco_monitor:lookup_request_field(TransId, #request.trans_id) of + {ok, _} -> + {WaitFor, Timer2} = megaco_timer:restart(Timer), + M = ?MODULE, + F = segment_timeout, + A = [ConnHandle, TransId, Timer2], + Ref = megaco_monitor:apply_after(M, F, A, WaitFor), + %% Timing problem? + megaco_monitor:update_request_field(TransId, + #request.seg_timer_ref, + Ref), + {restarted, WaitFor, Timer2}; + _ -> + not_found_ignore + end. + +%% segment_reply_timeout() -> +%% ok. + +missing_to_str(Missing) -> + lists:flatten(missing_to_str2(Missing)). + +missing_to_str2([X]) -> + [integer_to_list(X)]; +missing_to_str2([H|T]) -> + [integer_to_list(H) , "," | missing_to_str2(T)]. + +return_unexpected_trans_reply(ConnData, TransId, + {actionReplies, _} = UserReply, Extra) -> + Trans = make_transaction_reply(ConnData, TransId, UserReply), + return_unexpected_trans(ConnData, Trans, Extra); +return_unexpected_trans_reply(ConnData, TransId, + {transactionError, _} = UserReply, Extra) -> + Trans = make_transaction_reply(ConnData, TransId, UserReply), + return_unexpected_trans(ConnData, Trans, Extra); +return_unexpected_trans_reply(CD, TransId, {error, Reason}, Extra) -> + ?report_important(CD, "unexpected trans reply with error", + [TransId, Reason, Extra]), + ok; +return_unexpected_trans_reply(CD, TransId, Crap, Extra) -> + ?report_important(CD, "unexpected trans reply with crap", + [TransId, Crap, Extra]), + ok. + +return_unexpected_trans(ConnData, Trans) -> + Extra = ?default_user_callback_extra, + return_unexpected_trans(ConnData, Trans, Extra). + +return_unexpected_trans(ConnData, Trans0, Extra) -> + UserMod = ConnData#conn_data.user_mod, + UserArgs = ConnData#conn_data.user_args, + ConnHandle = ConnData#conn_data.conn_handle, + Version = ConnData#conn_data.protocol_version, + Trans = transform_transaction_reply_enc(Version, Trans0), + Args = + case Extra of + ?default_user_callback_extra -> + [ConnHandle, Version, Trans | UserArgs]; + _ -> + [ConnHandle, Version, Trans, Extra | UserArgs] + end, + Res = (catch apply(UserMod, handle_unexpected_trans, Args)), + ?report_debug(ConnData, "return: unexpected trans", + [Trans, {return, Res}]), + case Res of + ok -> + ok; + _ -> + warning_msg("unexpected transaction callback failed: ~w", [Res]), + ok + end, + Res. + + +%%----------------------------------------------------------------- + +to_remote_trans_id(#conn_data{conn_handle = CH, serial = Serial}) -> + Mid = CH#megaco_conn_handle.remote_mid, + #trans_id{mid = Mid, serial = Serial}. + +to_local_trans_id(#conn_data{conn_handle = CH, serial = Serial}) -> + Mid = CH#megaco_conn_handle.local_mid, + #trans_id{mid = Mid, serial = Serial}. + +to_local_trans_id(#conn_data{conn_handle = CH}, [S|_] = Serials) + when is_integer(S) -> + Mid = CH#megaco_conn_handle.local_mid, + [#trans_id{mid = Mid, serial = Serial} || Serial <- Serials]; +to_local_trans_id(#conn_data{conn_handle = CH}, + [{transactionRequest, TR}|_] = TRs) + when is_record(TR, 'TransactionRequest') -> + Mid = CH#megaco_conn_handle.local_mid, + [#trans_id{mid = Mid, serial = Serial} || + {transactionRequest, + #'TransactionRequest'{transactionId = Serial}} <- TRs]; + +to_local_trans_id(#megaco_conn_handle{local_mid = Mid}, Serial) + when is_integer(Serial) -> + #trans_id{mid = Mid, serial = Serial}; +to_local_trans_id(#conn_data{conn_handle = CH}, Serial) + when is_integer(Serial) -> + Mid = CH#megaco_conn_handle.local_mid, + #trans_id{mid = Mid, serial = Serial}. + + +%%----------------------------------------------------------------- + +transform_transaction_reply_dec({'TransactionReply', + TransId, IAR, TransRes}) -> + #megaco_transaction_reply{transactionId = TransId, + immAckRequired = IAR, + transactionResult = TransRes}; +transform_transaction_reply_dec({'TransactionReply', + TransId, IAR, TransRes, + SegNo, SegComplete}) -> + #megaco_transaction_reply{transactionId = TransId, + immAckRequired = IAR, + transactionResult = TransRes, + segmentNumber = SegNo, + segmentationComplete = SegComplete}. + +transform_transaction_reply_enc( + 3, + #megaco_transaction_reply{transactionId = TransId, + immAckRequired = IAR, + transactionResult = TransRes, + segmentNumber = SegNo, + segmentationComplete = SegComplete}) -> + {'TransactionReply', TransId, IAR, TransRes, SegNo, SegComplete}; +transform_transaction_reply_enc( + Version, + #megaco_transaction_reply{transactionId = TransId, + immAckRequired = IAR, + transactionResult = TransRes}) + when (Version < 3) -> + {'TransactionReply', TransId, IAR, TransRes}; +transform_transaction_reply_enc(_, TR) -> + TR. + +make_transaction_reply(#conn_data{protocol_version = Version}, + TransId, TransRes) -> + make_transaction_reply(Version, TransId, asn1_NOVALUE, TransRes). + +%% make_transaction_reply(#conn_data{protocol_version = Version}, +%% TransId, IAR, TransRes) -> +%% make_transaction_reply(Version, TransId, IAR, TransRes); + +make_transaction_reply(3, TransId, IAR, TransRes) -> + {'TransactionReply', TransId, IAR, TransRes, asn1_NOVALUE, asn1_NOVALUE}; +make_transaction_reply(_, TransId, IAR, TransRes) -> + {'TransactionReply', TransId, IAR, TransRes}. + + +%%----------------------------------------------------------------- + +%%----------------------------------------------------------------- +warning_msg(F, A) -> + ?megaco_warning(F, A). + +error_msg(F, A) -> + ?megaco_error(F, A). + + +%%----------------------------------------------------------------- + +%% d(F) -> +%% d(F,[]). +%% +%% d(F,A) -> +%% d(true,F,A). +%% %% d(get(dbg),F,A). +%% +%% d(true,F,A) -> +%% io:format("*** [~s] ~p:~p ***" +%% "~n " ++ F ++ "~n", +%% [format_timestamp(now()), self(),?MODULE|A]); +%% d(_, _, _) -> +%% ok. +%% +%% format_timestamp({_N1, _N2, N3} = Now) -> +%% {Date, Time} = calendar:now_to_datetime(Now), +%% {YYYY,MM,DD} = Date, +%% {Hour,Min,Sec} = Time, +%% FormatDate = +%% io_lib:format("~.4w:~.2.0w:~.2.0w ~.2.0w:~.2.0w:~.2.0w 4~w", +%% [YYYY,MM,DD,Hour,Min,Sec,round(N3/1000)]), +%% lists:flatten(FormatDate). + +%% Time in milli seconds +t() -> + {A,B,C} = erlang:now(), + A*1000000000+B*1000+(C div 1000). + + +%%----------------------------------------------------------------- +%% Func: incNumErrors/0, incNumErrors/1, incNumTimerRecovery/1 +%% Description: SNMP counter increment functions +%%----------------------------------------------------------------- +incNumErrors() -> + incNum(medGwyGatewayNumErrors). + +incNumErrors(CH) -> + incNum({CH, medGwyGatewayNumErrors}). + +incNumTimerRecovery(CH) -> + incNum({CH, medGwyGatewayNumTimerRecovery}). + +incNum(Cnt) -> + case (catch ets:update_counter(megaco_stats, Cnt, 1)) of + {'EXIT', {badarg, _Reason}} -> + ets:insert(megaco_stats, {Cnt, 1}); + Old -> + Old + end. + |