%%
%% %CopyrightBegin%
%%
%% Copyright Ericsson AB 2010-2022. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%
%% %CopyrightEnd%
%%
%%
%%----------------------------------------------------------------------
%% Purpose: The HDLT client module.
%%          This is the traffic generator
%%----------------------------------------------------------------------

-module(hdlt_client).

-export([
	 start/1, 
	 stop/0, 
	 start_inets/0, 
	 start_service/1, 
	 release/0, 
	 node_info/0
	]).

-export([
	 proxy/1
	]).

-include("hdlt_logger.hrl").

-define(CTRL,   hdlt_ctrl).
-define(PROXY,  hdlt_proxy).

-record(state, 
	{
	 mode = initial,
	 send_rate,
	 time, 
	 stop_time, 
	 url,
	 nof_reqs = 0,
	 nof_reps = 0,
	 last_req,
	 sizes,
	 socket_type, 
	 cert_file
	}).



start(Debug) ->
    proc_lib:start_link(?MODULE, proxy, [Debug]).

stop() ->
    (catch erlang:send(?PROXY, stop)),
    ok.

start_inets() ->
    ?PROXY ! start_inets.

start_service(Args) ->
    ?PROXY ! {start_client, Args, self()},
    receive
	client_started ->
	    %% ?LOG("client service started"),
	    ok
    end.

release() ->
    ?PROXY ! release.

node_info() ->
    ?PROXY ! {node_info, self()},
    receive 
	{node_info, NodeInfo} ->
	    NodeInfo
    end.


%% ---------------------------------------------------------------------
%% 
%% The proxy process
%% 

proxy(Debug) ->
    process_flag(trap_exit, true),
    erlang:register(?PROXY, self()),
    SName = lists:flatten(
	      io_lib:format("HDLT PROXY[~p,~p]", [self(), node()])), 
    ?SET_NAME(SName),
    ?SET_LEVEL(Debug), 
    ?LOG("starting", []),
    Ref = await_for_controller(10), 
    CtrlNode = node(Ref), 
    erlang:monitor_node(CtrlNode, true),
    proc_lib:init_ack({ok, self()}),
    ?DEBUG("started", []),
    proxy_loop(Ref, CtrlNode, undefined).

await_for_controller(N) when N > 0 ->
    case global:whereis_name(hdlt_ctrl) of
	Pid when is_pid(Pid) ->
	    erlang:monitor(process, Pid);
	_ ->
	    timer:sleep(1000),
	    await_for_controller(N-1)
    end;
await_for_controller(_) ->
    proc_lib:init_ack({error, controller_not_found, nodes()}),
    timer:sleep(500),
    init:stop().


proxy_loop(Ref, CtrlNode, Client) ->
    ?DEBUG("await command", []),
    receive
	stop ->
	    ?LOG("stop", []),
	    timer:sleep(1000),
	    halt();

	start_inets ->
	    ?LOG("start the inets service framework", []),
	    %% inets:enable_trace(max, "/tmp/inets-httpc-trace.log", all),
	    case (catch inets:start()) of
		ok ->
		    ?LOG("framework started", []),
		    proxy_loop(Ref, CtrlNode, Client);
		Error ->
		    ?LOG("failed starting inets service framework: "
			"~n   Error: ~p", [Error]),
		    timer:sleep(1000),
		    halt()
	    end;

	{start_client, Args, From} ->
	    ?LOG("start client with"
		"~n   Args: ~p", [Args]),
	    Client2 = spawn_link(fun() -> client(Args) end),
	    From ! client_started,
	    proxy_loop(Ref, CtrlNode, Client2);

	release ->
	    ?LOG("release", []),
	    Client ! go,
	    proxy_loop(Ref, CtrlNode, Client);

	{node_info, Pid} ->
	    ?LOG("received requets for node info", []),
	    NodeInfo = get_node_info(),
	    Pid ! {node_info, NodeInfo}, 
	    proxy_loop(Ref, CtrlNode, Client);

	{'EXIT', Client, normal} ->
	    ?LOG("received normal exit message from client (~p)", 
		 [Client]),
	    exit(normal);
	
	{'EXIT', Client, Reason} ->
	    ?INFO("received exit message from client (~p)"
		 "~n   Reason: ~p", [Client, Reason]),
	    %% Unexpected client termination, inform the controller and die
	    global:send(hdlt_ctrl, {client_exit, Client, node(), Reason}),
	    exit({client_exit, Reason});

	{nodedown, CtrlNode} ->
	    ?LOG("received nodedown for controller node - terminate", []), 
	    halt();

	{'DOWN', Ref, process, _, _} ->
	    ?INFO("received DOWN message for controller - terminate", []),
	    %% The controller has terminated, dont care why, time to die
	    halt()

    end.



%% ---------------------------------------------------------------------
%% 
%% The client process
%% 

client([SocketType, CertFile, URLBase, Sizes, Time, SendRate, Debug]) ->
    SName = lists:flatten(
	      io_lib:format("HDLT CLIENT[~p,~p]", [self(), node()])), 
    ?SET_NAME(SName),
    ?SET_LEVEL(Debug), 
    ?LOG("starting with"
	 "~n   SocketType: ~p"
	 "~n   Time:       ~p"
	 "~n   SendRate:   ~p", [SocketType, Time, SendRate]),
    httpc:set_options([{max_pipeline_length, 0}]),
    if
	(SocketType =:= ssl) orelse
	(SocketType =:= ossl) orelse
	(SocketType =:= essl) ->
	    %% Ensure crypto and ssl started:
	    crypto:start(),
	    ssl:start();
	true ->
	    ok
    end,
    State = #state{mode        = idle, 
		   url         = URLBase, 
		   time        = Time, 
		   send_rate   = SendRate,
		   sizes       = Sizes,
		   socket_type = SocketType,
		   cert_file   = CertFile},
    ?DEBUG("started", []),
    client_loop(State).

%% The point is to first start all client nodes and then this
%% process. Then, when they are all started, the go-ahead, go, 
%% message is sent to let them lose at the same time.
client_loop(#state{mode      = idle, 
		   time      = Time, 
		   send_rate = SendRate} = State) ->
    ?DEBUG("[idle] awaiting the go command", []),
    receive 
	go ->
	    ?LOG("[idle] received go", []),
	    erlang:send_after(Time, self(), stop),
	    NewState = send_requests(State, SendRate),	    
	    client_loop(NewState#state{mode     = generating, 
				       nof_reqs = SendRate})
    end;

%% In this mode the client is generating traffic.
%% It will continue to do so until the stop message
%% is received. 
client_loop(#state{mode = generating} = State) -> 
    receive 
	stop ->
	    ?LOG("[generating] received stop", []),
	    StopTime = timestamp(), 
	    req_reply(State),
	    client_loop(State#state{mode = stopping, stop_time = StopTime});

	{http, {_, {{_, 200, _}, _, _}}} ->
	    %% ?DEBUG("[generating] received reply - send another request", []),
	    NewState = send_requests(State, 1),
	    client_loop(NewState#state{nof_reps = NewState#state.nof_reps + 1,
				       nof_reqs = NewState#state.nof_reqs + 1});

	{http, {ReqId, {error, Reason}}} ->
	    ?INFO("[generating] request ~p failed: "
		  "~n   Reason:  ~p"
		  "~n   NofReqs: ~p"
		  "~n   NofReps: ~p", 
		  [ReqId, Reason, State#state.nof_reqs, State#state.nof_reps]),
	    exit({Reason, generating, State#state.nof_reqs, State#state.nof_reps});

	Else ->
	    ?LOG("[generating] received unexpected message: "
		 "~n~p", [Else]),
	    unexpected_data(Else), 
	    client_loop(State)
    end;

%% The client no longer issues any new requests, instead it 
%% waits for replies for all the outstanding requests to 
%% arrive.
client_loop(#state{mode     = stopping, 
		   time     = Time, 
		   last_req = LastReqId} = State) ->
    receive 
	{http, {LastReqId, {{_, 200, _}, _, _}}} ->
	    ?DEBUG("[stopping] received reply for last request (~p)", [LastReqId]),
	    time_to_complete(State),
	    ok;

	{http, {ReqId, {{_, 200, _}, _, _}}} ->
	    ?DEBUG("[stopping] received reply ~p", [ReqId]),
	    client_loop(State);

	{http, {ReqId, {error, Reason}}} ->
	    ?INFO("[stopping] request ~p failed: "
		  "~n   Reason:  ~p"
		  "~n   NofReqs: ~p"
		  "~n   NofReps: ~p", 
		  [ReqId, Reason, State#state.nof_reqs, State#state.nof_reps]),
	    exit({Reason, stopping, State#state.nof_reqs, State#state.nof_reps});

	Else ->
	    ?LOG("[stopping] received unexpected message: "
		 "~n~p", [Else]),
	    unexpected_data(Else), 
	    client_loop(State)

    after Time ->
	    ?INFO("timeout when"
		  "~n   Number of requests: ~p"
		  "~n   Number of replies:  ~p", 
		  [State#state.nof_reqs, State#state.nof_reps]),
	    exit({timeout, State#state.nof_reqs, State#state.nof_reps})
    end.

req_reply(#state{nof_reqs = NofReqs, nof_reps = NofReps}) ->
    load_data({req_reply, node(), NofReqs, NofReps}).

time_to_complete(#state{stop_time = StopTime}) ->
    StoppedTime = os:timestamp(),
    load_data({time_to_complete, node(), StopTime, StoppedTime}).

load_data(Data) ->
    global:send(?CTRL, {load_data, Data}).

unexpected_data(Else) ->
    global:send(?CTRL, {unexpected_data, Else}).


send_requests(#state{sizes = Sizes} = State, N) ->
    send_requests(State, N, Sizes).

send_requests(State, 0, Sizes) ->
    State#state{sizes = Sizes};
send_requests(#state{socket_type = SocketType, 
		     cert_file   = CertFile} = State, N, [Sz | Sizes]) ->
    URL = lists:flatten(io_lib:format("~s~w", [State#state.url, Sz])),
    Method      = get,
    Request     = {URL, []},
    HTTPOptions = 
	case SocketType of
	    ip_comm ->
		[];
	    _ ->
		SslOpts = [{verify, 0},
			   {certfile, CertFile},
			   {keyfile,  CertFile}],
		case SocketType of
		    ssl ->
			[{ssl, SslOpts}];
		    ossl ->
			[{ssl, {ossl, SslOpts}}];
		    essl ->
			[{ssl, {essl, SslOpts}}]
		end
	end,
    Options = [{sync, false}], 
    {ok, Ref} = httpc:request(Method, Request, HTTPOptions, Options), 
    send_requests(State#state{last_req = Ref}, N-1, lists:append(Sizes, [Sz])).


timestamp() ->
   os:timestamp().
 

get_node_info() ->
    [{cpu_topology,        erlang:system_info(cpu_topology)},
     {heap_type,           erlang:system_info(heap_type)},
     {nof_schedulers,      erlang:system_info(schedulers)},
     {otp_release,         erlang:system_info(otp_release)}, 
     {version,             erlang:system_info(version)}, 
     {system_version,      erlang:system_info(system_version)},
     {system_architecture, erlang:system_info(system_architecture)}].