megaco_test_mg.erl

OTP是开放电信平台的简称

%% ``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 via the world wide web 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.
%% 
%% The Initial Developer of the Original Code is Ericsson Utvecklings AB.
%% Portions created by Ericsson are Copyright 1999, Ericsson Utvecklings
%% AB. All Rights Reserved.''
%% 
%%     $Id$
%%
%%----------------------------------------------------------------------
%% Purpose: Implements an "MG" used by the test suite
%%----------------------------------------------------------------------
-module(megaco_test_mg).

-export([start/5, start/6, stop/1, 
	 get_stats/1, reset_stats/1, 
	 user_info/1, user_info/2, 
	 update_user_info/3, 
	 conn_info/1, conn_info/2, 
	 update_conn_info/3, 
	 service_change/1, 
	 ack_info/2, rep_info/2, 
	 group_requests/2, 
	 notify_request/1, 
	 await_notify_reply/1, 
	 notify_request_and_reply/1, 
	 cancel_request/2, 
	 apply_load/2, 
	 apply_multi_load/3, 
	 enable_test_code/4, 
	 encode_ar_first/2, 
	 verbosity/2]).

-export([mg/3, notify_request_handler_main/5]).
-export([loader_main/4]).

%% Megaco callback api
-export([
	 handle_connect/4,
	 handle_disconnect/5,
	 handle_syntax_error/5,
	 handle_message_error/5,
	 handle_trans_request/5,
	 handle_trans_long_request/5,
	 handle_trans_reply/6,
	 handle_trans_ack/6
	]).

-include("megaco_test_lib.hrl").
-include_lib("megaco/include/megaco.hrl").
-include_lib("megaco/include/megaco_message_v1.hrl").

-define(A4444, tid(255*256*256)                 ).
-define(A4445, tid(255*256*256 + 255)           ).
-define(A5555, tid(255*256*256 + 255*256)       ).
-define(A5556, tid(255*256*256 + 255*256 + 255) ).

-record(mg, {mid             = undefined,
	     state           = initiated,
	     req_handler     = undefined,
	     call_mode       = async,
	     group_size      = 1,
	     encode_ar_first = false,
	     ack_info        = undefined,
	     rep_info        = undefined,
	     load_counter    = 0,
	     reply_counter   = 0,
	     mload_info      = undefined,
	     parent          = undefined}).


%%% --------------------------------------------------------------------

start(Node, Mid, Encoding, Transport, Verbosity) ->
    %% Conf = [{megaco_trace, io}],
    %% Conf = [{megaco_trace, "megaco-mg.trace"}],
    Conf = [{megaco_trace, false}],
    start(Node, Mid, Encoding, Transport, Conf, Verbosity).

start(Node, Mid, Encoding, Transport, Conf, Verbosity) ->
    d("start mg[~p]: ~p", [Node, Mid]),
    RI1    = encoding_config(Encoding),
    RI2    = transport_config(Transport),
    RI     = {receive_info, RI1 ++ RI2},
    Config = [{local_mid, Mid}, RI] ++ Conf,
    Self   = self(),
    Fun    = 
	fun() ->
		io:format("LOADER(~p,~p) started~n", [self(),node()]),
		case (catch mg(Self, Verbosity, Config)) of
		    {'EXIT', Reason} ->
			io:format("LOADER(~p,~p) terminating with exit"
				  "~n~p"
				  "~n", [self(), node(), Reason]),
			exit(Reason);
		    Else ->
			io:format("LOADER(~p,~p) terminating with"
				  "~n~p"
				  "~n", [self(), node(), Else]),
			Else
		end
	end,
    Pid = spawn_link(Node, Fun),
    %% Pid = spawn_link(Node, ?MODULE, mg, [self(), Verbosity, Config]),
    MonRef = (catch erlang:monitor(process, Pid)),
    i("start -> "
      "~n   self():           ~p"
      "~n   node():           ~p"
      "~n   net_adm:ping(~p): ~p" 
      "~n   Loader:           ~p"
      "~n   Monitor ref:      ~p"
      "~n   Process info:     ~p", 
      [self(), node(), 
       Node, net_adm:ping(Node), 
       Pid, 
       MonRef, (catch proc_info(Pid))]),
    await_started(MonRef, Pid).

proc_info(Pid) ->
    rpc:call(node(Pid), erlang, process_info, [Pid]).

encoding_config({Encoding, EC}) when atom(Encoding), list(EC) ->
    {Mod, Port} = select_encoding(Encoding),
    [{encoding_module, Mod},
     {encoding_config, EC},
     {port,            Port}];
encoding_config(Encoding) when atom(Encoding) ->
    {Mod, Port} = select_encoding(Encoding),
    [{encoding_module, Mod},
     {encoding_config, []},
     {port,            Port}];
encoding_config(Encoding) ->
    throw({error, {invalid_encoding, Encoding}}).

select_encoding(text) ->
    {megaco_pretty_text_encoder, 2944};
select_encoding(pretty_text) ->
    {megaco_pretty_text_encoder, 2944};
select_encoding(compact_text) ->
    {megaco_compact_text_encoder, 2944};
select_encoding(binary) ->
    {megaco_ber_bin_encoder, 2945};
select_encoding(erl_dist) ->
    {megaco_erl_dist_encoder, 2946};
select_encoding(Encoding) ->
    throw({error, {invalid_encoding, Encoding}}).

transport_config(tcp) ->
    [{transport_module, megaco_tcp}];
transport_config(udp) ->
    [{transport_module, megaco_udp}].

await_started(MonRef, Pid) ->
    i("await_started -> entry with"
      "~n   MonRef: ~p"
      "~n   Pid:    ~p", [MonRef, Pid]),
    receive
	{started, Pid} ->
	    d("await_started ~p: "
	      "~n   Process info: ~p", [Pid, (catch proc_info(Pid))]),
	    erlang:demonitor(MonRef),
	    {ok, Pid};

	{'DOWN', MonRef, process, Pid, Info} ->
	    i("await_started ~p: received down signal: ~p", 
	      [Pid, Info]),
	    exit({failed_starting, Pid, Info});

	{'EXIT', Pid, Reason} ->
	    i("await_started ~p: received exit signal: ~p", [Pid, Reason]),
	    exit({failed_starting, Pid, Reason})

    after 10000 ->
	    i("await_started ~p: timeout: "
	      "~n~p", [Pid, (catch process_info(Pid))]),
	    exit({error, timeout})
    end.


verbosity(Pid, V) ->
    Pid ! {verbosity, V, self()}.


stop(Pid) ->
    server_request(Pid, stop, stopped).


get_stats(Pid) ->
    server_request(Pid, statistics, statistics_reply).

reset_stats(Pid) ->
    server_request(Pid, reset_stats, reset_stats_ack).


user_info(Pid) ->
    server_request(Pid, {user_info, all}, user_info_ack).

user_info(Pid, Tag) when atom(Tag) ->
    server_request(Pid, {user_info, Tag}, user_info_ack).


update_user_info(Pid, Tag, Val) ->
    server_request(Pid, {update_user_info, Tag, Val}, update_user_info_ack).


conn_info(Pid) ->
    server_request(Pid, {conn_info, all}, conn_info_ack).

conn_info(Pid, Tag) when atom(Tag) ->
    server_request(Pid, {conn_info, Tag}, conn_info_ack).


update_conn_info(Pid, Tag, Val) ->
    server_request(Pid, {update_conn_info, Tag, Val}, update_conn_info_ack).


enable_test_code(Pid, Module, Where, Fun) 
  when atom(Module), atom(Where), function(Fun) ->
    Tag = {Module, Where},
    server_request(Pid, {enable_test_code, Tag, Fun}, enable_test_code_reply).

encode_ar_first(Pid, New) when atom(New) ->
    server_request(Pid, {encode_ar_first, New}, encode_ar_first_reply).

service_change(Pid) ->
    server_request(Pid, service_change, service_change_reply).


group_requests(Pid, N) ->
    server_request(Pid, {group_requests, N}, group_requests_reply).


ack_info(Pid, InfoPid) ->
    Pid ! {{ack_info, InfoPid}, self()}.

rep_info(Pid, InfoPid) ->
    Pid ! {{rep_info, InfoPid}, self()}.


notify_request(Pid) ->
    Pid ! {notify_request, self()}.


await_notify_reply(Pid) ->
    await_reply(Pid, notify_request_reply).


notify_request_and_reply(Pid) ->
    notify_request(Pid),
    await_notify_reply(Pid).


cancel_request(Pid, Reason) ->
    server_request(Pid, cancel_request, Reason, cancel_request_reply).


apply_load(Pid, CounterStart) ->
    server_request(Pid, apply_load, CounterStart, apply_load_ack).


apply_multi_load(Pid, NumLoaders, NumReq) ->
    server_request(Pid, apply_multi_load, {NumLoaders, NumReq}, apply_multi_load_ack).


server_request(Pid, Req, ReplyTag) ->
    Pid ! {Req, self()},
    await_reply(Pid, ReplyTag).

server_request(Pid, Req, ReqData, ReplyTag) ->
    Pid ! {Req, ReqData, self()},
    await_reply(Pid, ReplyTag).

await_reply(Pid, ReplyTag) ->
    await_reply(Pid, ReplyTag, infinity).

await_reply(Pid, ReplyTag, Timeout) ->
    receive
	{ReplyTag, Reply, Pid} ->
	    Reply;
	{'EXIT', Pid, Reason} ->
	    exit({failed, ReplyTag, Pid, Reason})
    after Timeout ->
	    exit({timeout, ReplyTag, Pid})
    end.


server_reply(Pid, ReplyTag, Reply) ->
    Pid ! {ReplyTag, Reply, self()}.


%%% --------------------------------------------------------------------


mg(Parent, Verbosity, Config) ->
    process_flag(trap_exit, true),
    put(verbosity, Verbosity),
    %% put(verbosity, debug),
    put(sname,   "MG"),
    i("mg -> starting"),
    case (catch init(Config)) of
	{error, _} = Error ->
	    exit(Error);
	Mid ->
	    notify_started(Parent),
	    MG = #mg{parent = Parent, mid = Mid},
	    i("mg -> started"),
	    case (catch loop(MG)) of
		{'EXIT', normal} ->
		    exit(normal);
		{'EXIT', Reason} ->
		    i("mg failed with reason:~n   ~p", [Reason]),
		    exit(Reason);
		Else ->
		    i("mg terminated: ~n   ~p", [Else]),
		    exit({unexpected, Else})
	    end
    end.

init(Config) ->
    d("init -> entry with"
      "~n   Config: ~p", [Config]),
    random_init(),
    Mid = get_conf(local_mid, Config),
    RI  = get_conf(receive_info, Config),
    d("init -> start megaco"),
    application:start(megaco),


    d("init -> possibly enable megaco trace"),
    case lists:keysearch(megaco_trace, 1, Config) of
	{value, {megaco_trace, true}} ->
	    megaco:enable_trace(max, io);
	{value, {megaco_trace, io}} ->
	    megaco:enable_trace(max, io);
	{value, {megaco_trace, File}} when list(File) ->
	    megaco:enable_trace(max, File);
	_ ->
	    ok
    end,
    Conf0 = lists:keydelete(megaco_trace, 1, Config),
    
    d("init -> start megaco user"),
    Conf1 = lists:keydelete(local_mid,    1, Conf0),
    Conf2 = lists:keydelete(receive_info, 1, Conf1),
    ok = megaco:start_user(Mid, Conf2),
    d("init -> update user info (user_mod)"),
    ok = megaco:update_user_info(Mid, user_mod,  ?MODULE),
    d("init -> update user info (user_args)"),
    ok = megaco:update_user_info(Mid, user_args, [self(), Mid]),

%     d("init -> start megaco user"),
%     megaco:start_user(Mid, []),
%     d("init -> update user info (user_mod)"),
%     megaco:update_user_info(Mid, user_mod,  ?MODULE),
%     d("init -> update user info (user_args)"),
%     megaco:update_user_info(Mid, user_args, [self()]),

    d("init -> get user info (receive_handle)"),
    RH = megaco:user_info(Mid,receive_handle),
    d("init -> parse receive info"),
    {MgcPort,RH1} = parse_receive_info(RI, RH),
    d("init -> start transport"),
    {ok, _CH} = start_transport(MgcPort, RH1),
    Mid.


loop(#mg{parent = Parent, mid = Mid} = S) ->
    d("loop -> await request", []),
    receive
	{verbosity, V, Parent} ->
	    i("loop -> received new verbosity: ~p", [V]),
	    put(verbosity,V),
	    loop(S);


	{stop, Parent} ->
	    i("loop -> stopping", []),
	    Res = do_stop(Mid),
	    d("loop -> stop result: ~p", [Res]),
	    server_reply(Parent, stopped, {ok, Res}),
	    exit(normal);

	{{enable_test_code, Tag, Fun}, Parent} ->
	    i("loop -> enable_test_code: ~p, ~p", [Tag, Fun]),
	    Reply = (catch ets:insert(megaco_test_data, {Tag, Fun})),
	    d("loop -> enable_test_code -> "
	      "~n   Reply:                          ~p"
	      "~n   ets:tab2list(megaco_test_data): ~p", 
	      [Reply,ets:tab2list(megaco_test_data)]),
	    server_reply(Parent, enable_test_code_reply, Reply),
	    loop(S);

	{{encode_ar_first, EAF}, Parent} ->
	    i("loop -> encode_ar_first: ~p", [EAF]),
	    {Reply, S1} = handle_encode_ar_first(S, EAF),
	    server_reply(Parent, encode_ar_first_reply, Reply),
	    loop(S1#mg{encode_ar_first = EAF});

	%% Give me statistics
	{statistics, Parent} ->
	    i("loop -> got request for statistics", []),
	    Stats = do_get_statistics(Mid),
	    server_reply(Parent, statistics_reply, {ok, Stats}),
	    loop(S); 

	{reset_stats, Parent} ->
	    i("loop -> got request to reset stats counters", []),
	    do_reset_stats(Mid),
	    server_reply(Parent, reset_stats_ack, ok),
	    loop(S);

	{{user_info, Tag}, Parent} ->
	    i("loop -> got user_info request for ~w", [Tag]),
	    Res = do_get_user_info(Mid, Tag),
	    d("loop -> Res: ~p", [Res]),
	    server_reply(Parent, user_info_ack, Res),
	    loop(S);

	{{update_user_info, Tag, Val}, Parent} ->
	    i("loop -> got update_user_info: ~w -> ~p", [Tag, Val]),
	    Res = do_update_user_info(Mid, Tag, Val),
	    d("loop -> Res: ~p", [Res]),
	    server_reply(Parent, update_user_info_ack, Res),
	    loop(S);

	{{conn_info, Tag}, Parent} ->
	    i("loop -> got conn_info request for ~w", [Tag]),
	    Res = do_get_conn_info(Mid, Tag),
	    server_reply(Parent, conn_info_ack, Res),
	    loop(S);

	{{update_conn_info, Tag, Val}, Parent} ->
	    i("loop -> got update_conn_info: ~w -> ~p", [Tag, Val]),
	    Res = do_update_conn_info(Mid, Tag, Val),
	    server_reply(Parent, update_conn_info_ack, Res),
	    loop(S);


	%% Do a service change
	%% No server-reply here. Since the service change is 
	%% async, the reply (from the MGC) will come later.
	{service_change, Parent} ->
	    i("loop -> received request to perform service change", []),
	    S1 = 
		case (catch do_service_change(S)) of
		    {ok, MG} ->
			d("loop -> service change initiated", []),
			MG;
		    Error ->
			d("loop -> service change failed: ~p", [Error]),
			server_reply(Parent, service_change_reply, Error),
			S
		end,
	    loop(S1); 

	{{group_requests, N}, Parent} when N > 0 ->
	    i("loop -> received group_requests ~p", [N]),
	    Reply = {ok, S#mg.group_size}, 
	    server_reply(Parent, group_requests_reply, Reply),
	    loop(S#mg{group_size = N}); 

	{{ack_info, To}, Parent} ->
	    i("loop -> received request to inform about received ack's ", []),
	    loop(S#mg{ack_info = To});

	{{rep_info, To}, Parent} ->
	    i("loop -> received request to inform about received rep's ", []),
	    loop(S#mg{rep_info = To});

	%% Make a sync-call
	{notify_request, Parent} ->
	    i("loop -> received request to send notify request ", []),
	    {Res, S1} = do_handle_notify_request(S),
	    d("loop -> notify request result: ~p", [Res]),
	    loop(S1);

	%% sync-call complete
	{notify_request_complete, NotifyReply, Pid} ->