megaco_test_mg.erl

OTP是开放电信平台的简称

	    i("loop -> received notify request complete from "
	      "~n   ~p with"
	      "~n   NotifyReply: ~p", 
	      [Pid, NotifyReply]),
	    server_reply(Parent, notify_request_reply, NotifyReply),
	    loop(S#mg{req_handler = undefined});


	%% cancel requests
	{cancel_request, Reason, Parent} ->
	    i("loop -> received request to cancel (all) megaco requests ", []),
	    Res = do_cancel_requests(Mid, Reason),
	    server_reply(Parent, cancel_request_reply, Res),
	    loop(S);


	%% Apply multi-load
	{apply_multi_load, {NL, NR}, Parent} -> 
	    i("loop -> received apply_multi_load request: ~w, ~w", [NL, NR]),
	    S1 = start_loaders(S, NL, NR),
	    loop(S1);


	%% Apply some load
	{apply_load, Times, Parent} ->
	    i("loop -> received apply_load request: ~w", [Times]),
	    S1 = 
		case update_load_times(S, Times) of
		    {ok, MG} ->
			apply_load_timer(),
			server_reply(Parent, apply_load_ack, ok),
			MG;
		    Error ->
			server_reply(Parent, apply_load_ack, Error),
			S
		end,
	    loop(S1);

	apply_load_timeout ->
	    d("loop -> received apply_load timeout", []),
	    S1 = do_apply_load(S),
	    loop(S1);


	%% Megaco callback messages
	{request, Request, Mid, From} ->
	    d("loop -> received megaco request: ~n   ~p"
	      "~n   Mid:  ~p"
	      "~n   From: ~p", 
	      [Request, Mid, From]),
	    {Reply, S1} = handle_megaco_request(S, Request),
	    d("loop -> send (megaco callback) request reply: ~n~p", [Reply]),
	    From ! {reply, Reply, self()},
	    loop(S1);


	{'EXIT', Pid, Reason} -> 
	    i("loop -> received exit signal from ~p: ~n~p", [Pid, Reason]),
	    S1 = handle_exit(S, Pid, Reason),
	    loop(S1);


	Invalid ->
	    error_msg("received invalid request: ~n~p", [Invalid]),
	    loop(S)

    end.


handle_encode_ar_first(#mg{encode_ar_first = Old} = MG, New) 
  when New == true; New == false ->
    {{ok, Old}, MG#mg{encode_ar_first = New}};
handle_encode_ar_first(MG, New) ->
    {{error, {invalid_value, New}}, MG}.
    

%%
%% Stop user
%%
do_stop(Mid) ->
    d("do_stop -> stopping user ~p", [Mid]),
    Disco = fun close_conn/1, 
    lists:map(Disco, megaco:user_info(Mid, connections)),
    megaco:stop_user(Mid).

close_conn(CH) ->
    d("do_stop -> closing connection ~p", [CH]),
    Reason     = {stopped_by_user,self()},
    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),
    megaco:cancel(CH, Reason),
    case SendMod of
	megaco_tcp -> megaco_tcp:close(SendHandle);
	megaco_udp -> megaco_udp:close(SendHandle);
	SendMod    -> exit(Pid, Reason)
    end.



%% 
%% Get statistics 
%% 
do_get_statistics(Mid) ->
    case megaco:user_info(Mid, connections) of
	[CH] ->
	    do_get_conn_statistics(CH);
	[] ->
	    []
    end.

do_get_conn_statistics(CH) ->
    {ok, Gen}  = megaco:get_stats(),
    %% 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)
	end,
    [{gen, Gen}, {trans, Trans}].


%%
%% reset user stats
%%
do_reset_stats(Mid) ->
    %% We only have one connection
    [CH] = megaco:user_info(Mid, connections),
    do_reset_stats1(CH).

do_reset_stats1(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.


%%
%% Get user info for user
%%
do_get_user_info(Mid, all = Tag) ->
    case (catch megaco:user_info(Mid, Tag)) of
	L when is_list(L) ->
	    lists:sort(L);
	Else ->
	    Else
    end;
do_get_user_info(Mid, Tag) ->
    (catch megaco:user_info(Mid, Tag)).


%%
%% Update user info for user
%%
do_update_user_info(Mid, Tag, Val) ->
    (catch megaco:update_user_info(Mid, Tag, Val)).


%%
%% Get conn info 
%%
do_get_conn_info(CH, all = Tag) when record(CH, megaco_conn_handle) ->
    case (catch megaco:conn_info(CH, Tag)) of
	L when is_list(L) ->
	    lists:sort(L);
	Else ->
	    Else
    end;
do_get_conn_info(CH, Tag) when record(CH, megaco_conn_handle) ->
    (catch megaco:conn_info(CH, Tag));
do_get_conn_info(Mid, Tag) ->
    case megaco:user_info(Mid, connections) of
	[CH|_] ->
	    do_get_conn_info(CH, Tag);
	[] ->
	    []
    end.

%%
%% Update conn info for user
%%
do_update_conn_info(Mid, Tag, Val) ->
    %% We only have one connection
    [CH] = megaco:user_info(Mid, connections),
    (catch megaco:update_conn_info(CH, Tag, Val)).




%%
%% Perform service change 
%%
do_service_change(#mg{mid             = Mid, 
		      state           = initiated, 
		      encode_ar_first = EAF} = MG) ->
    %% We only have one connection
    d("do service change for ~p", [Mid]),
    [CH]   = megaco:user_info(Mid, connections),
    Method = restart,
    Reason = ?megaco_cold_boot,
    case do_service_change(CH, Method, EAF, Reason) of
	ok ->
	    {ok, MG#mg{state = connecting}};
	Error ->
	    d("service change for ~p failed: ~n~p", [Mid, Error]),
	    Error
    end;
do_service_change(#mg{state = State} = MG) ->
    {{error, {invalid_state, State}}, MG}.

do_service_change(ConnHandle, Method, EAF, Reason) ->
    d("sending service change using:"
      "~n   ConnHandle: ~p"
      "~n   Method:     ~p"
      "~n   EAF:        ~p"
      "~n   Reason:     ~p", [ConnHandle, Method, EAF, Reason]),
    SCP    = cre_serviceChangeParm(Method, [Reason]),
    TermId = [?megaco_root_termination_id],
    SCR    = cre_serviceChangeReq(TermId, SCP),
    CR     = cre_commandReq({serviceChangeReq, SCR}),
    AR     = cre_actionReq(?megaco_null_context_id,[CR]),
    send_async(EAF, ConnHandle, [AR], []).


%% Make a sync call 
do_handle_notify_request(#mg{mid             = Mid, 
			     group_size      = N, 
			     encode_ar_first = EAF, 
			     state           = connected} = MG) ->
    d("do_handle_notify_request -> entry"),
    [CH] = megaco:user_info(Mid, connections),
    Pid = start_notify_request_handler(EAF, CH, N),
    {ok, MG#mg{req_handler = Pid}};
do_handle_notify_request(#mg{state = State} = MG) ->
    d("do_handle_notify_request -> entry with"
      "~n   State: ~p", [State]),
    {{error, {invalid_state, State}}, MG}.



%%
%% Cancel requests 
%%
do_cancel_requests(Mid, Reason) ->
    [CH] = megaco:user_info(Mid, connections),
    megaco:cancel(CH, Reason).    


%% 
%% Apply multi load 
%% 
start_loaders(#mg{mid = Mid, encode_ar_first = EAF} = MG, NumLoaders, Times) ->
    [CH] = megaco:user_info(Mid, connections),
    Env  = get(),
    Loaders = start_loaders1(NumLoaders, [], [Env, EAF, Times, CH]),
    d("start_loaders -> Loaders: ~n~w", [Loaders]),
    MG#mg{mload_info = {Loaders, 0, 0}}.

start_loaders1(0, Acc, _) ->
    Acc;
start_loaders1(N, Acc, Args) ->
    Pid = spawn_link(?MODULE, loader_main, Args),
    start_loaders1(N-1, [Pid|Acc], Args).

loader_main(Env, EAF, N, CH) ->
    lists:foreach(fun({Tag,Val}) -> put(Tag,Val) end, Env),
    loader_main(EAF, N, CH).
			  
loader_main(_EAF, 0, _) ->
    d("loader_main -> done"),
    exit(loader_done);
loader_main(EAF, N, CH) ->
    d("loader_main -> entry with: ~w", [N]),
    {Act, _} = make_notify_request(),
    _Res     = send_sync(EAF, CH, Act, []),
    loader_main(EAF, N-1, CH).



handle_exit(#mg{parent = Pid}, Pid, Reason) ->
    error_msg("received exit from the parent:"
	      "~n   ~p", [Reason]),
    exit({parent_terminated, Reason});

handle_exit(#mg{parent = Parent, req_handler = Pid} = MG, Pid, Reason) ->
    error_msg("received unexpected exit from the request handler:"
	      "~n   ~p", [Reason]),
    server_reply(Parent, notify_request_reply, 
		 {error, {request_handler_exit, Reason}}),
    MG#mg{req_handler = undefined};

handle_exit(#mg{parent = Parent, mload_info = {Loaders0, Ok, Err}} = MG, 
	    Pid, loader_done) ->
    d("handle_exit(loader_done) -> entry when"
      "~n   Loaders0: ~p"
      "~n   Ok:       ~p"
      "~n   Err:      ~p", [Loaders0, Ok, Err]),
    Loaders = lists:delete(Pid, Loaders0),
    LoadInfo =
	case Loaders of
	    [] ->
		d("handle_exit -> multi load done", []),
		server_reply(Parent, apply_multi_load_ack, {ok, Ok+1, Err}),
		undefined;
	    _ ->
		{Loaders, Ok+1, Err}
	end,
    MG#mg{mload_info = LoadInfo};


handle_exit(#mg{parent = Parent, mload_info = {Loaders, Ok, Err}} = MG, 
	    Pid, Reason) 
  when length(Loaders) > 0 ->
    d("handle_exit -> entry when"
      "~n   Reason:  ~p"
      "~n   Loaders: ~p"
      "~n   Ok:      ~p"
      "~n   Err:     ~p", [Reason, Loaders, Ok, Err]),
    case lists:delete(Pid, Loaders) of
	[] -> 
	    %% since we cannot be empty prior the delete, 
	    %% the last one exited...
	    server_reply(Parent, apply_multi_load, {ok, Ok, Err+1}),
	    MG#mg{mload_info = undefined};
	Loaders ->
	    %% Could not be this MG, so go on to the next
	    error_msg("received unexpected exit signal from ~p:~n~p", 
		      [Pid, Reason]);
	Loaders1 ->
	    %% Not empty, but we removed one
	    MG#mg{mload_info = {Loaders1,Ok,Err+1}}
    end;
handle_exit(_MG, Pid, Reason) ->
    error_msg("received unexpected exit signal from ~p:~n~p", 
	      [Pid, Reason]).

		      
parse_receive_info(RI, RH) ->
    d("parse_receive_info -> get encoding module"),
    EM = get_encoding_module(RI),
    d("parse_receive_info -> get encoding config"),
    EC = get_encoding_config(RI, EM),
    d("parse_receive_info -> get transport module"),
    TM = get_transport_module(RI),
    d("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}.


start_transport(MgcPort,
		   #megaco_receive_handle{send_mod = megaco_tcp} = RH) ->
    start_tcp(MgcPort,RH);
start_transport(MgcPort,
		   #megaco_receive_handle{send_mod = megaco_udp} = RH) ->
    start_udp(MgcPort,RH);
start_transport(_, #megaco_receive_handle{send_mod = Mod}) ->
    throw({error, {bad_send_mod, Mod}}).


start_tcp(MgcPort, RH) ->
    d("start tcp transport (~p)", [MgcPort]),
    case megaco_tcp:start_transport() of
	{ok, Sup} ->
	    d("tcp transport started: ~p", [Sup]),
	    {ok, LocalHost} = inet:gethostname(),
	    Opts = [{host, LocalHost},{port, MgcPort}, {receive_handle, RH}],
	    d("tcp connect", []),
	    case megaco_tcp:connect(Sup, Opts) of
		{ok, SendHandle, ControlPid} ->
		    d("tcp connected: ~p, ~p", [SendHandle, ControlPid]),
		    megaco_tcp_connect(RH, LocalHost, SendHandle, ControlPid);
		{error, Reason} ->
		    throw({error, {megaco_tcp_connect, Reason}})
	    end;
        {error, Reason} ->
            throw({error, {megaco_tcp_start_transport, Reason}})
    end.

megaco_tcp_connect(RH, _LocalHost, SendHandle, ControlPid) ->
    PrelMgcMid = preliminary_mid,
    d("megaco connect", []),    
    {ok, CH} = megaco:connect(RH, PrelMgcMid, SendHandle, ControlPid),
    d("megaco connected: ~p", [CH]),    
    {ok, CH}.

start_udp(MgcPort, RH) ->
    d("start udp transport (~p)", [MgcPort]),
    case megaco_udp:start_transport() of
	{ok, Sup} ->
	    d("udp transport started: ~p", [Sup]),
	    Opts = [{port, 0}, {receive_handle, RH}],
	    d("udp open", []),
	    case megaco_udp:open(Sup, Opts) of
		{ok, Handle, ControlPid} ->
		    d("udp opened: ~p, ~p", [Handle, ControlPid]),
		    megaco_udp_connect(MgcPort, RH, Handle, ControlPid);
		{error, Reason} ->
                    throw({error, {megaco_udp_open, Reason}})
	    end;
        {error, Reason} ->
            throw({error, {megaco_udp_start_transport, Reason}})
    end.

megaco_udp_connect(MgcPort, RH, Handle, ControlPid) ->
    {ok, LocalHost} = inet:gethostname(),
    MgcMid     = preliminary_mid,
    SendHandle = megaco_udp:create_send_handle(Handle, LocalHost, MgcPort),
    d("megaco connect", []),    
    {ok, CH}   = megaco:connect(RH, MgcMid, SendHandle, ControlPid),
    d("megaco connected: ~p", [CH]),    
    {ok, CH}.



update_load_times(#mg{load_counter = 0} = MG, Times) ->
    d("update_load_times(0) -> entry with"
      "~n   Times: ~p", [Times]),
    {ok, MG#mg{load_counter = Times}};
update_load_times(#mg{load_counter = N}, Times) ->
    d("update_load_times(~p) -> entry with"
      "~n   Times: ~p", [N, Times]),
    {error, {already_counting, N}}.


do_apply_load(#mg{mid = Mid} = MG) ->
    d("do_apply_load -> entry"),
    case megaco:user_info(Mid, connections) of
	[CH] ->
	    do_apply_load(MG, CH);
	[] ->
	    i("failed to apply load: no connections for ~p", [Mid]),
	    MG
    end.

do_apply_load(#mg{parent          = Parent,
		  encode_ar_first = EAF,
		  call_mode       = Mode, 
		  group_size      = Sz, 
		  load_counter    = N0} = MG, CH) ->
    d("do_apply_load -> entry with"
      "~n   Mode: ~p"
      "~n   Sz:   ~p"
      "~n   N0:   ~p", [Mode, Sz, N0]),
    {NofSent, Actions, ReplyData} = make_notify_request(N0, Sz),
    d("do_apply_load -> notifications constructed:"
      "~n   NofSent:   ~p"
      "~n   Actions:   ~p"
      "~n   ReplyData: ~p", [NofSent, Actions, ReplyData]),
    N = N0 - NofSent,
    case Mode of
	sync ->
	    Result = send_sync(EAF, CH, Actions, []),
	    d("do_apply_load -> call result when N = ~p: ~n~p", [N,Result]),
	    case N of
		0 ->
		    d("do_apply_load -> load complete"),
		    Parent ! {load_complete, self()},
		    MG#mg{call_mode = async, load_counter = 0};
		_ ->
		    d("do_apply_load -> make another round"),
		    apply_load_timer(),
		    MG#mg{call_mode = async, load_counter = N}
	    end;
	async ->
	    Result = send_async(EAF, CH, Actions, [{reply_data, ReplyData}]),
	    d("do_apply_load -> cast result:~n   ~p", [Result]),
	    MG#mg{call_mode     = sync, 
		  load_counter  = N, 
		  reply_counter = NofSent} % Outstanding replies