megaco_mib_test.erl

来自「OTP是开放电信平台的简称」· ERL 代码 · 共 1,492 行 · 第 1/4 页

    throw({error, no_transport});
mgc_start_transports(ListenTo) ->
    mgc_start_transports(ListenTo, undefined, undefined).
    

mgc_start_transports([], TcpSup, UdpSup) ->
    {TcpSup, UdpSup};
mgc_start_transports([{Port, RH}|ListenTo], TcpSup, UdpSup) 
  when RH#megaco_receive_handle.send_mod == megaco_tcp ->
    TcpSup1 = mgc_start_tcp(RH, Port, TcpSup),
    mgc_start_transports(ListenTo, TcpSup1, UdpSup);
mgc_start_transports([{Port, RH}|ListenTo], TcpSup, UdpSup) 
  when RH#megaco_receive_handle.send_mod == megaco_udp ->
    UdpSup1 = mgc_start_udp(RH, Port, UdpSup),
    mgc_start_transports(ListenTo, TcpSup, UdpSup1);
mgc_start_transports([{_Port, RH}|_ListenTo], _TcpSup, _UdpSup) ->
    throw({error, {bad_send_mod, RH#megaco_receive_handle.send_mod}}).


mgc_start_tcp(RH, Port, undefined) ->
    d("start tcp transport"),
    case megaco_tcp:start_transport() of
	{ok, Sup} ->
	    mgc_start_tcp(RH, Port, Sup);
	Else ->
	    throw({error, {failed_starting_tcp_transport, Else}})
    end;
mgc_start_tcp(RH, Port, Sup) when pid(Sup) ->
    d("tcp listen on ~p", [Port]),
    Opts = [{port, Port}, {receive_handle, RH}],
    case megaco_tcp:listen(Sup, Opts) of
	ok ->
	    Sup;
	Else ->
	    throw({error, {failed_starting_tcp_listen, Else}})
    end.


mgc_start_udp(RH, Port, undefined) ->
    d("start udp transport"),
    case megaco_udp:start_transport() of
	{ok, Sup} ->
	    mgc_start_udp(RH, Port, Sup);
	Else ->
	    throw({error, {failed_starting_udp_transport, Else}})
    end;
mgc_start_udp(RH, Port, Sup) ->
    d("open udp ~p", [Port]),
    Opts = [{port, Port}, {receive_handle, RH}],
    case megaco_udp:open(Sup, Opts) of
	{ok, _SendHandle, _ControlPid} ->
	    Sup;
	Else ->
	    exit({error, {failed_starting_udp_listen, Else}})
    end.



%% -----------------------
%% Handle megaco callbacks
%%

mgc_handle_request({handle_connect, _CH, _PV}) ->
    ok;
mgc_handle_request({handle_disconnect, CH, _PV, R}) ->
    megaco:cancel(CH, R), % Cancel the outstanding messages
    ok;
mgc_handle_request({handle_syntax_error, _RH, _PV, _ED}) ->
    reply;
mgc_handle_request({handle_message_error, _CH, _PV, _ED}) ->
    no_reply;
mgc_handle_request({handle_trans_request, CH, PV, ARs}) ->
    ED =  #'ErrorDescriptor'{errorCode = ?megaco_not_implemented,
                             errorText = "Only single service change on null context handled"},
    case ARs of
        [AR] ->
            ContextId = AR#'ActionRequest'.contextId,
            case AR#'ActionRequest'.commandRequests of
                [CR] when ContextId == ?megaco_null_context_id ->
                    case CR#'CommandRequest'.command of
                        {serviceChangeReq, Req} ->
                            Rep    = mgc_service_change(CH, PV, Req),
			    CmdRep = [{serviceChangeReply, Rep}],
                            {discard_ack,
                             [#'ActionReply'{contextId    = ContextId,
                                             commandReply = CmdRep}]};
                        _ ->
                            {discard_ack, ED}
                    end;
                _ ->
                    {discard_ack, ED}
            end;
        _ ->
            {discard_ack, ED}
    end;
mgc_handle_request({handle_trans_long_request, _CH, _PV, _RD}) ->
    ED = #'ErrorDescriptor'{errorCode = ?megaco_not_implemented,
                            errorText = "Long transaction requests not handled"},
    {discard_ack,  ED};
mgc_handle_request({handle_trans_reply, _CH, _PV, _AR, _RD}) ->
    ok;
mgc_handle_request({handle_trans_ack, _CH, _PV, _AS, _AD}) ->
    ok.

mgc_service_change(CH, _PV, SCR) ->
    SCP = SCR#'ServiceChangeRequest'.serviceChangeParms,
    #'ServiceChangeParm'{serviceChangeAddress = Address,
                         serviceChangeProfile = Profile,
                         serviceChangeReason  = [_Reason]} = SCP,
    TermId = SCR#'ServiceChangeRequest'.terminationID,
    if
        TermId == [?megaco_root_termination_id] ->
            MyMid = CH#megaco_conn_handle.local_mid,
            Res = {serviceChangeResParms,
                   #'ServiceChangeResParm'{serviceChangeMgcId   = MyMid,
                                           serviceChangeAddress = Address,
                                           serviceChangeProfile = Profile}},
            #'ServiceChangeReply'{terminationID       = TermId,
                                  serviceChangeResult = Res};
        true ->
            Res = {errorDescriptor,
                   #'ErrorDescriptor'{errorCode = ?megaco_not_implemented,
                                      errorText = "Only handled for root"}},
            
             #'ServiceChangeReply'{terminationID       = TermId,
                                   serviceChangeResult = Res}
    end.

    
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

start_mg(Node, Mid, Encoding, Transport, Verbosity) ->
    d("start mg[~p]: ~p", [Node, Mid]),
    RI1 = 
	case Encoding of 
	    text ->
		[{encoding_module, megaco_pretty_text_encoder},
		 {encoding_config, []},
		 {port,2944}];
	    binary ->
		[{encoding_module, megaco_ber_bin_encoder},
		 {encoding_config, []},
		 {port,2945}]
	end,
    RI2 = 
	case Transport of
	    tcp ->
		[{transport_module, megaco_tcp}];
	    udp ->
		[{transport_module, megaco_udp}]
	end,
    RI = {receive_info, RI1 ++ RI2},
    Config = [{local_mid, Mid}, RI],
    Pid = spawn_link(Node, ?MODULE, mg, [self(), Verbosity, Config]),
    await_started(Pid).


mg(Parent, Verbosity, Config) ->
    process_flag(trap_exit, true),
    put(verbosity, Verbosity),
    put(sname,   "MG"),
    i("mg -> starting"),
    {Mid, ConnHandle} = mg_init(Config),
    notify_started(Parent),
    S = #mg{parent = Parent, mid = Mid, conn_handle = ConnHandle},
    i("mg -> started"),
    mg_loop(S).

mg_init(Config) ->
    d("mg_init -> entry"),
    random_init(),
    Mid = get_conf(local_mid, Config),
    RI  = get_conf(receive_info, Config),
    d("mg_init -> start megaco"),
    application:start(megaco),
    d("mg_init -> start megaco user"),
    megaco:start_user(Mid, []),
    d("mg_init -> update user info (user_mod)"),
    megaco:update_user_info(Mid, user_mod,  ?MODULE),
    d("mg_init -> update user info (user_args)"),
    megaco:update_user_info(Mid, user_args, [self()]),
    d("mg_init -> get user info (receive_handle)"),
    RH = megaco:user_info(Mid,receive_handle),
    d("mg_init -> parse receive info"),
    {MgcPort,RH1} = mg_parse_receive_info(RI, RH),
    d("mg_init -> start transport with"),
    ConnHandle = mg_start_transport(MgcPort, RH1),
    {Mid, ConnHandle}.

mg_loop(#mg{state = State} = S) ->
    d("mg_loop(~p) -> await request", [State]),
    receive
	{stop, Parent} when S#mg.parent == Parent ->
	    i("mg_loop(~p) -> stopping", [State]),
	    mg_close_conn(S#mg.conn_handle),
	    megaco:stop_user(S#mg.mid),
	    application:stop(megaco),
	    i("mg_loop(~p) -> stopped", [State]),
	    Parent ! {stopped, self()},
	    exit(normal);


	{reset_stats, Parent} when S#mg.parent == Parent ->
	    i("mg_loop(~p) -> got request to reset stats counters", [State]),
	    %% mg_reset_stats(S#mgc.conn_handle),
	    mg_reset_stats(S#mg.conn_handle),
	    Parent ! {reset_stats_ack, self()},
	    mg_loop(S);



	%% Give me statistics
	{statistics, 1, Parent} when S#mg.parent == Parent ->
	    i("mg_loop(~p) -> got request for statistics 1", [State]),
	    {ok, Gen} = megaco:get_stats(),
	    CH = S#mg.conn_handle,
	    Reason = {statistics, CH}, 
	    Pid = megaco:conn_info(CH, control_pid),
	    SendMod    = megaco:conn_info(CH, send_mod),
	    SendHandle = megaco:conn_info(CH, send_handle),
	    {ok, Trans} = 
		case SendMod of
		    megaco_tcp -> megaco_tcp:get_stats(SendHandle);
		    megaco_udp -> megaco_udp:get_stats(SendHandle);
		    SendMod    -> exit(Pid, Reason)
		end,
	    Parent ! {statistics, 1, [{gen, Gen}, {trans, Trans}], self()},
	    mg_loop(S); 


	%% Do a service change
	{service_change, Parent} when S#mg.parent == Parent, 
				      State == initiated ->
	    i("mg_loop(~p) -> received request to perform service change", 
	      [State]),
	    Res = mg_service_change(S#mg.conn_handle),
	    d("mg_loop(~p) -> result: ~p", [State, Res]),
	    mg_loop(S#mg{state = connecting}); 


	%% Apply some load
	{apply_load, Parent, Times} when S#mg.parent == Parent ->
	    i("mg_loop(~p) -> received apply_load request", [State]),
	    apply_load_timer(),
	    Parent ! {apply_load_ack, self()},
	    mg_loop(S#mg{load_counter = Times - 1});


	apply_load_timeout ->
	    d("mg_loop(~p) -> received apply_load timeout [~p]", 
	      [State, S#mg.load_counter]),
	    mg_apply_load(S),
	    mg_loop(S);


	%% Megaco callback messages
	{request, Request, From} ->
	    d("mg_loop(~p) -> received megaco request: ~n~p~n   From: ~p", 
	      [State, Request, From]),
	    {Reply, NewS} = mg_handle_request(Request, S),
	    d("mg_loop(~p) -> send request reply: ~n~p", 
	      [NewS#mg.state, Reply]),
	    From ! {reply, Reply, self()},
	    mg_loop(NewS);


	{'EXIT', Pid, Reason} ->
	    error_msg("MG ~p received unexpected exit signal from ~p:~n~p", 
		      [S#mg.mid, Pid, Reason]),
	    mg_loop(S);


	Invalid ->
	    i("mg_loop(~p) -> received invalid request: ~p", [State, Invalid]),
	    mg_loop(S)

    end.


mg_reset_stats(CH) ->
    megaco:reset_stats(),
    case (catch megaco:conn_info(CH, send_mod)) of
	{error, Reason} ->
	    error_msg("unexpected result when retrieving send module for "
		      "own connection ~p: ~p. "
		      "~nexiting...", [CH, Reason]),
	    exit({invalid_connection, CH, Reason});
	{'EXIT', Reason} ->
	    error_msg("exit signal when retrieving send module for "
		      "own connection ~p: ~p. "
		      "~nexiting...", [CH, Reason]),
	    exit({invalid_connection, CH, Reason});
	SendMod when atom(SendMod) ->
	    SendMod:reset_stats()
    end.


mg_close_conn(CH) ->
    Reason = {self(), ignore},
    Pid        = megaco:conn_info(CH, control_pid),
    SendMod    = megaco:conn_info(CH, send_mod),
    SendHandle = megaco:conn_info(CH, send_handle),
    megaco:disconnect(CH, Reason),
    case SendMod of
	megaco_tcp -> megaco_tcp:close(SendHandle);
	megaco_udp -> megaco_udp:close(SendHandle);
	SendMod    -> exit(Pid, Reason)
    end.

% display_tuple(T) when tuple(T), size(T) > 0 ->
%     i("size(T): ~p", [size(T)]),
%     display_tuple(T,1).

% display_tuple(T,P) when P > size(T) ->
%     ok;
% display_tuple(T,P) ->
%     i("T[~p]: ~p", [P,element(P,T)]),
%     display_tuple(T,P+1).


mg_parse_receive_info(RI, RH) ->
    d("mg_parse_receive_info -> get encoding module"),
    EM = get_encoding_module(RI),
    d("mg_parse_receive_info -> get encoding config"),
    EC = get_encoding_config(RI, EM),
    d("mg_parse_receive_info -> get transport module"),
    TM = get_transport_module(RI),
    d("mg_parse_receive_info -> get transport port"),
    TP = get_transport_port(RI),
    RH1 = RH#megaco_receive_handle{send_mod        = TM,
				   encoding_mod    = EM,
				   encoding_config = EC},
    {TP, RH1}.


mg_start_transport(MgcPort,
		   #megaco_receive_handle{send_mod = megaco_tcp} = RH) ->
    mg_start_tcp(MgcPort,RH);
mg_start_transport(MgcPort,
		   #megaco_receive_handle{send_mod = megaco_udp} = RH) ->
    mg_start_udp(MgcPort,RH);
mg_start_transport(_, #megaco_receive_handle{send_mod = Mod}) ->
    throw({error, {bad_send_mod, Mod}}).


mg_start_tcp(MgcPort, RH) ->
    d("start tcp transport"),
    case megaco_tcp:start_transport() of
	{ok, Sup} ->
	    {ok, LocalHost} = inet:gethostname(),
	    Opts = [{host, LocalHost},{port, MgcPort}, {receive_handle, RH}],
	    case megaco_tcp:connect(Sup, Opts) of
		{ok, SendHandle, ControlPid} ->
                    PrelMgcMid = preliminary_mid,
		    {ok, ConnHandle} = 
			megaco:connect(RH, PrelMgcMid, 
				       SendHandle, ControlPid),
		    ConnHandle;
		{error, Reason} ->
		    {error, {megaco_tcp_connect, Reason}}
	    end;
        {error, Reason} ->
            {error, {megaco_tcp_start_transport, Reason}}
    end.


mg_start_udp(MgcPort, RH) ->
    d("start udp transport"),
    case megaco_udp:start_transport() of
	{ok, Sup} ->
	    {ok, LocalHost} = inet:gethostname(),
	    Opts = [{port, 0}, {receive_handle, RH}],
	    case megaco_udp:open(Sup, Opts) of