megaco_test_generator.erl

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

	Error ->
	    e("send -> send failed: ~n~p",[Error]),
	    close(Sock),
	    close(Tcp#tcp.listen),
	    tcp_error(send, {send_error, Error})
    end;

do_tcp({send, Desc, Msg}, #tcp{connection = Sock,
			       encode     = Encode} = Tcp) ->
    p("send ~s message", [Desc]),
    case (catch Encode(Msg)) of
	{ok, Bin} ->
	    d("send -> message encoded [~w], now add tpkt header: ~n~s", 
	      [sz(Bin), binary_to_list(Bin)]),
	    NewBin = add_tpkt_header(Bin),
	    d("send -> tpkt header added [~w], now send", [sz(NewBin)]),
	    case (catch gen_tcp:send(Sock, NewBin)) of
		ok ->
		    d("send -> message sent"),
		    Tcp;
		Error ->
		    e("send -> send failed: ~n~p",[Error]),
		    close(Sock),
		    close(Tcp#tcp.listen),
		    tcp_error(send, {send_error, Error})
	    end;
	Error ->
	    e("send -> encode failed: ~n~p",[Error]),
	    close(Sock),
	    close(Tcp#tcp.listen),
	    tcp_error(send, {endode_error, Error})
    end;

do_tcp({expect_receive, Desc, {Verify, To}}, 
       #tcp{connection = Sock,
	    decode     = Decode} = Tcp) ->
    p("expect_receive ~s message", [Desc]),
    inet:setopts(Sock, [{active, once}]),
    receive
	{tcp, Sock, <<3:8, _X:8, Length:16, Msg/binary>>} ->
	    d("expect_receive -> received message: Length = ~p", [Length]),
	    case (catch Decode(Msg)) of
		{ok, MegaMsg} when is_tuple(MegaMsg) ->
		    d("expect_receive -> decode successfull, now verify"),
		    case (catch Verify(MegaMsg)) of
			{ok, Res} ->
			    d("expect_receive -> verify successfull"),
			    Acc = Tcp#tcp.result,
			    Tcp#tcp{result = [Res|Acc]};
			Else ->
			    e("failed to verify message: ~n~p~n~p", 
			      [Else, MegaMsg]),
			    tcp_error(expect_receive, {verify_failed, Else})
		    end;
		Error ->
		    e("failed decoding message: ~p", [Error]),
		    tcp_error(expect_receive, Error)
	    end;
	Else ->
	    d("received unknown message: ~p", [Else]),
	    tcp_error(expect_receive, {unexpected_message, Else})
    after To ->
	    tcp_error(expect_receive, timeout)
    end;

do_tcp({expect_nothing, To}, #tcp{connection = Sock} = Tcp) ->
    p("expect_nothing ~w", [To]),
    inet:setopts(Sock, [{active, once}]),
    p("expect_nothing - await anything", []),
    receive
	Any ->
	    p("expect_nothing - received: ~p", [Any]),
	    tcp_error(expect_nothing, Any)
    after To ->
	    p("expect_nothing timeout after ~w", [To]),
	    Tcp
    end;

do_tcp({trigger, Trigger}, Tcp) when is_function(Trigger) ->
    p("trigger"),
    Trigger(),
    Tcp;

do_tcp({sleep, To}, Tcp) ->
    p("sleep ~p", [To]),
    sleep(To),
    Tcp.


tcp_error(Instr, Error) ->
    throw({error, {Instr, Error}}).


tcp_cleanup(#tcp{listen = Listen, connection = Conn}) ->
    close(Listen),
    close(Conn).


parse_tcp([], RevInstrs) ->
    {ok, lists:reverse(RevInstrs)};


parse_tcp([{debug, Debug}|Instrs], RevInstrs) 
  when Debug == true; Debug == false ->
    parse_tcp(Instrs, [{debug, Debug}|RevInstrs]);

%%
%% -- All the basic events, used when bypassing the megaco stack --
%% 

parse_tcp([{encode, Encode}|Instrs], RevInstrs) 
  when is_function(Encode) ->
    parse_tcp(Instrs, [{encode, Encode}|RevInstrs]);

parse_tcp([{encode, {Mod, Func, Args}}|Instrs], RevInstrs) 
  when is_atom(Mod) and is_atom(Func) and is_list(Args) ->
    Encode = 
	fun(M) ->
		apply(Mod, Func, [M|Args])
	end,
    parse_tcp(Instrs, [{encode, Encode}|RevInstrs]);

parse_tcp([{decode, Decode}|Instrs], RevInstrs) 
  when is_function(Decode) ->
    parse_tcp(Instrs, [{decode, Decode}|RevInstrs]);

parse_tcp([{decode, {Mod, Func, Args}}|Instrs], RevInstrs) 
  when is_atom(Mod) and is_atom(Func) and is_list(Args) ->
    Decode = 
	fun(M) ->
		apply(Mod, Func, [M|Args])
	end,
    parse_tcp(Instrs, [{decode, Decode}|RevInstrs]);

parse_tcp([disconnect|Instrs], RevInstrs) ->
    parse_tcp(Instrs, [disconnect|RevInstrs]);

parse_tcp([{listen, Port}|Instrs], RevInstrs) 
  when is_integer(Port), Port > 0 ->
    parse_tcp(Instrs, [{listen, Port}|RevInstrs]);

parse_tcp([{expect_accept, any}|Instrs], RevInstrs) ->
    parse_tcp(Instrs, [{expect_accept, {any, infinity}}|RevInstrs]);

parse_tcp([{expect_accept, {any, To}}|Instrs], RevInstrs) 
  when is_integer(To), To >= 0 ->
    parse_tcp(Instrs, [{expect_accept, {any, To}}|RevInstrs]);

parse_tcp([{expect_accept, {Host, infinity}}|Instrs], RevInstrs) ->
    {ok, Addr} = inet:getaddr(Host, inet),
    parse_tcp(Instrs, [{expect_accept, {Addr, infinity}}|RevInstrs]);

parse_tcp([{expect_accept, {Host, To}}|Instrs], RevInstrs) 
  when is_integer(To), To >= 0 ->
    {ok, Addr} = inet:getaddr(Host, inet),
    parse_tcp(Instrs, [{expect_accept, {Addr, To}}|RevInstrs]);

parse_tcp([{expect_accept, Host}|Instrs], RevInstrs) ->
    {ok, Addr} = inet:getaddr(Host, inet),
    parse_tcp(Instrs, [{expect_accept, {Addr, infinity}}|RevInstrs]);

parse_tcp([{connect, Port}|Instrs], RevInstrs) 
  when is_integer(Port), Port > 0 ->
    {ok, Host} = inet:gethostname(),
    {ok, Addr} = inet:getaddr(Host, inet),
    parse_tcp(Instrs, [{connect, {Addr, Port, infinity}}|RevInstrs]);

parse_tcp([{connect, {Port, infinity}}|Instrs], RevInstrs) 
  when is_integer(Port), Port > 0 ->
    {ok, Host} = inet:gethostname(),
    {ok, Addr} = inet:getaddr(Host, inet),
    parse_tcp(Instrs, [{connect, {Addr, Port, infinity}}|RevInstrs]);

parse_tcp([{connect, {Port, To}}|Instrs], RevInstrs) 
  when is_integer(Port) and (Port > 0) and 
       is_integer(To) and (To > 0) ->
    {ok, Host} = inet:gethostname(),
    {ok, Addr} = inet:getaddr(Host, inet),
    parse_tcp(Instrs, [{connect, {Addr, Port, To}}|RevInstrs]);

parse_tcp([{connect, {Host, Port}}|Instrs], RevInstrs) 
  when is_integer(Port) and (Port > 0) ->
    {ok, Addr} = inet:getaddr(Host, inet),
    parse_tcp(Instrs, [{connect, {Addr, Port, infinity}}|RevInstrs]);

parse_tcp([{connect, {Host, Port, infinity}}|Instrs], RevInstrs) 
  when is_integer(Port) and (Port > 0) ->
    {ok, Addr} = inet:getaddr(Host, inet),
    parse_tcp(Instrs, [{connect, {Addr, Port, infinity}}|RevInstrs]);

parse_tcp([{connect, {Host, Port, To}}|Instrs], RevInstrs) 
  when is_integer(Port) and (Port > 0) and
       is_integer(To) and (To > 0) ->
    {ok, Addr} = inet:getaddr(Host, inet),
    parse_tcp(Instrs, [{connect, {Addr, Port, To}}|RevInstrs]);

parse_tcp([{sleep, To}|Instrs], RevInstrs) 
  when is_integer(To) and (To > 0) ->
    parse_tcp(Instrs, [{sleep, To}|RevInstrs]);

parse_tcp([{expect_nothing, To}|Instrs], RevInstrs) 
  when is_integer(To) and (To > 0) ->
    parse_tcp(Instrs, [{expect_nothing, To}|RevInstrs]);

parse_tcp([{send, Desc, Msg}|Instrs], RevInstrs) 
  when is_list(Desc) and (is_tuple(Msg) or is_binary(Msg)) ->
    parse_tcp(Instrs, [{send, Desc, Msg}|RevInstrs]);

parse_tcp([{expect_receive, Desc, Verify}|Instrs], RevInstrs) 
  when is_list(Desc) and is_function(Verify) ->
    ExpRecv = {expect_receive, Desc, {Verify, infinity}},
    parse_tcp(Instrs, [ExpRecv|RevInstrs]);

parse_tcp([{expect_receive, Desc, {Mod, Func, Args}}|Instrs], RevInstrs) 
  when is_list(Desc) and 
       is_atom(Mod) and is_atom(Func) and is_list(Args) ->
    Verify = 
	fun(M) ->
		apply(Mod, Func, [M|Args])
	end,
    ExpRecv = {expect_receive, Desc, {Verify, infinity}},
    parse_tcp(Instrs, [ExpRecv|RevInstrs]);

parse_tcp([{expect_receive, Desc, {Verify, To}}|Instrs], RevInstrs) 
  when is_list(Desc) and is_function(Verify) and 
       is_integer(To) and (To > 0) ->
    ExpRecv = {expect_receive, Desc, {Verify, To}},
    parse_tcp(Instrs, [ExpRecv|RevInstrs]);

parse_tcp([{expect_receive, Desc, {{Mod, Func, Args}, To}}|Instrs], RevInstrs) 
  when is_list(Desc) and 
       is_atom(Mod) and is_atom(Func) and is_list(Args) and
       is_integer(To) and (To > 0) ->
    Verify = 
	fun(M) ->
		apply(Mod, Func, [M|Args])
	end,
    ExpRecv = {expect_receive, Desc, {Verify, To}},
    parse_tcp(Instrs, [ExpRecv|RevInstrs]);

parse_tcp([{trigger, Trigger}|Instrs], RevInstrs) 
  when is_function(Trigger) ->
    ExpRecv = {trigger, Trigger},
    parse_tcp(Instrs, [ExpRecv|RevInstrs]);

parse_tcp([Instr|_], _RevInstrs) ->
    throw({error, {invalid_instruction, Instr}}).


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

start_megaco_handler(Name, Instructions) ->
    spawn_link(?MODULE, megaco_handler_main, [Name, Instructions, self()]).

megaco_handler_main(Name, Instructions, Parent) ->
    put(name, Name),
    {Megaco, Reply} = handle_megaco(Instructions),
    Parent ! {megaco_result, self(), Reply},
    receive
	{stop, Parent} ->
	    megaco_cleanup(Megaco),
	    exit(normal)
    end.


%% Instructions: [instruction()]
%% instruction() = {debug,  Debug}  |
%%                 {sleep, To} |
%%                 
handle_megaco(Instructions0) when is_list(Instructions0) ->
    case (catch parse_megaco(Instructions0, [])) of
	{ok, Instructions} ->
	    (catch handle_megaco1(Instructions));
	Error ->
	    p("parsing arguments failed: ~n~p", [Error]),
	    {invalid, {error, Error}}
    end.

handle_megaco1(Instructions) ->
    handle_megaco1(Instructions, #megaco{}).

handle_megaco1([], #megaco{result = Res} = State) ->
    Reply = {ok, lists:reverse(Res)},
    {State, Reply};
handle_megaco1([Instruction|Instructions], State0) ->
    d("handle_megaco1 -> entry with"
      "~n   Instruction: ~p", [Instruction]),
    case (catch do_megaco(Instruction, State0)) of
	State when is_record(State, megaco) ->
	    handle_megaco1(Instructions, State);
	{error, State} when is_record(State, megaco) ->
	    p("handle_megaco1 -> error"), 
	    Reply = {error, {{instruction_failed, Instruction, Instructions},
			     lists:reverse(State#megaco.result)}},
	    {State, Reply};
	Error ->
	    p("handle_megaco1 -> Error: ~n~p", [Error]),
	    Reply = {error, {Error, lists:reverse(State0#megaco.result)}},
	    {State0, Reply}
    end.

do_megaco({debug, Debug}, State) ->
    p("debug: ~p", [Debug]),
    put(debug, Debug),
    State;

do_megaco({expect_nothing, To}, State) ->
    p("expect_nothing: ~p", [To]),
    receive
	Any ->
	    megaco_error(expect_nothing, Any)
    after To ->
        State
    end;

do_megaco({megaco_trace, disable}, State) ->
    p("megaco trace: disable"),
    megaco:disable_trace(),
    State;
do_megaco({megaco_trace, Level}, State) ->
    p("megaco trace: disable"),
    megaco:enable_trace(Level, io),
    State;

do_megaco(megaco_start, State) ->
    p("megaco_start"),
    ok = megaco:start(),
    State;

do_megaco(megaco_stop, State) ->
    p("megaco_stop"),
    ok = megaco:stop(),
    State;

do_megaco({megaco_start_user, Mid, RecvInfo, Conf}, State) ->
    p("megaco_start_user: ~p", [Mid]),

    d("megaco_start_user -> start user"),
    ok = megaco:start_user(Mid, Conf),
  
    d("megaco_start_user -> update user info: user_mod"),
    ok = megaco:update_user_info(Mid, user_mod,  ?MODULE),

    d("megaco_start_user -> update user info: user_args"),
    ok = megaco:update_user_info(Mid, user_args,  [self()]),
    
    Port = get_config(port, RecvInfo),
    EM   = get_config(encoding_module, RecvInfo),
    EC   = get_config(encoding_config, RecvInfo),
    TM   = get_config(transport_module, RecvInfo),
    RH0  = megaco:user_info(Mid, receive_handle),

    RH1  = RH0#megaco_receive_handle{send_mod        = TM,
				     encoding_mod    = EM,
				     encoding_config = EC},

    State#megaco{mid = Mid, recv_handle = RH1, port = Port};

do_megaco(megaco_stop_user, #megaco{mid = Mid} = State) 
  when Mid /= undefined ->
    megaco_cleanup(State),
    ok = megaco:stop_user(Mid),
    State#megaco{mid = undefined};

do_megaco(start_transport, #megaco{recv_handle = RH} = State) ->
    p("start_transport"),
    #megaco_receive_handle{send_mod = TM} = RH,
    {ok, Sup} = TM:start_transport(),
    d("start_transport -> Sup: ~p", [Sup]),
    State#megaco{transport_sup = Sup};

do_megaco({listen, Opts0}, 
	  #megaco{recv_handle = RH, port = Port, transport_sup = Pid} = State) 
  when RH#megaco_receive_handle.send_mod == megaco_tcp ->
    p("listen(tcp)"),
    Opts = [{port, Port}, {receive_handle, RH}|Opts0],
    case megaco_tcp:listen(Pid, Opts) of
	ok ->
	    State;
	Else ->
	    megaco_error({listen, Opts0}, {failed_starting_tcp_listen, Else})