<commit_before>%%======================================================================
%%
%% Leo RPC
%%
%% Copyright (c) 2012-2013 Rakuten, Inc.
%%
%% This file is provided to you 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.
%%
%%======================================================================
-module(leo_rpc_client_conn).

-author('<NAME>').

-behaviour(gen_server).

-include("leo_rpc.hrl").
-include_lib("eunit/include/eunit.hrl").

%% API
-export([start_link/1, stop/1]).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
         terminate/2, code_change/3]).

-record(state, {
          id     :: atom(),
          host   :: string(),
          ip     :: pos_integer(),
          port   :: integer(),
          socket :: reference()|undefined,
          reconnect_sleep :: integer(),
          buf    :: binary(),
          queue = [] :: list()
         }).


-define(SOCKET_OPTS, [binary, {active, once}, {packet, raw}, {reuseaddr, true}]).
-define(RECV_TIMEOUT, 5000).


%% ===================================================================
%% APIs
%% ===================================================================
-spec(start_link(list(any())) ->
             {ok, pid()} | {error, term()}).
start_link([Host, IP, Port, ReconnectSleepInterval]) ->
    gen_server:start_link(?MODULE, [Host, IP, Port, ReconnectSleepInterval], []).

stop(ServerRef) ->
    gen_server:call(ServerRef, stop).


%%====================================================================
%% gen_server callbacks
%%====================================================================
init([Host, IP, Port, ReconnectSleepInterval]) ->
    State = #state{host = Host,
                   ip = IP,
                   port = Port,
                   reconnect_sleep = ReconnectSleepInterval,
                   buf = <<>>,
                   queue = []},
    case connect(State) of
        {ok, NewState} ->
            {ok, NewState};
        {error, Reason} ->
            {stop, {connection_error, Reason}}
    end.

handle_call({request, Req}, From, State) ->
    exec(Req, From, State);

handle_call(status,_From, #state{socket = Socket} = State) ->
    Ret = (Socket /= undefined),
    {reply, {ok, Ret}, State};

handle_call(stop, _From, State) ->
    {stop, normal, ok, State};

handle_call(_Request, _From, State) ->
    {reply, unknown_request, State}.

handle_cast(_Msg, State) ->
    {noreply, State}.


handle_info({tcp, Socket, Bs}, #state{buf = Buf} = State) ->
    Res = recv(Socket, <<Buf/binary, Bs/binary>>),
    case Res of
        {error, Cause} ->
            {stop, Cause, State};
        {value, Value, Rest} ->
            inet:setopts(Socket, [{active, once}]),
            NewState = State#state{buf = Rest},
            {noreply, handle_response(Value, NewState)}
    end;
handle_info({tcp_error, _Socket, _Reason}, State) ->
    {noreply, State};

handle_info({tcp_closed, _Socket}, State) ->
    case State#state.reconnect_sleep of
        0 ->
            void;
        _ ->
            Self = self(),
            spawn(fun() -> reconnect_loop(Self, State) end)
    end,
    {noreply, State#state{socket = undefined, queue = []}};

handle_info({connection_ready, Socket}, #state{socket = undefined} = State) ->
    {noreply, State#state{socket = Socket}};

handle_info(stop, State) ->
    {stop, shutdown, State};

handle_info(_Info, State) ->
    {stop, {unhandled_message, _Info}, State}.

terminate(_Reason, #state{socket = Socket}) ->
    case Socket of
        undefined ->
            ok;
        Socket ->
            catch gen_tcp:close(Socket),
            ok
    end.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%% ===================================================================
%% Inner Functions
%% ===================================================================
%% @doc: Send the given request to the rpc-server
%% @ptivate
-spec(exec(iolist(), pid(), #state{}) ->
             {noreply, #state{}} | {reply, Reply::any(), #state{}}).
exec(Req, From, #state{socket = undefined} = State) ->
    case connect(State) of
        {ok, #state{socket = Socket} = State1} ->
            case gen_tcp:send(Socket, Req) of
                ok ->
                    NewQueue = [From|State1#state.queue],
                    {noreply, State1#state{queue = NewQueue}};
                {error, Reason} ->
                    {reply, {error, Reason}, State1}
            end;
        {error, Reason} ->
            {reply, {error, Reason}, State}
    end;

exec(Req, From, #state{socket = Socket} = State) ->
    case gen_tcp:send(Socket, Req) of
        ok ->
            NewQueue = [From|State#state.queue],
            {noreply, State#state{queue = NewQueue}};
        {error, Reason} ->
            {reply, {error, Reason}, State}
    end.


%% @doc: Handle the response coming from Server
%% @private
-spec(handle_response(binary(), #state{}) ->
             #state{}).
handle_response(Data, #state{queue = Queue} = State) ->
    NewQueue = reply(Data, Queue),
    State#state{queue = NewQueue}.


%% @doc: Send data to the 1st-client in the queue
%% @private
reply(_, []) ->
    error_logger:warning_msg(
      "~p,~p,~p,~p~n",
      [{module, ?MODULE_STRING}, {function, "reply/2"},
       {line, ?LINE}, {body, "Nothing in queue"}]),
    throw(empty_queue);

reply(Value, Queue) ->
    [From|NewQueue] = lists:reverse(Queue),
    gen_server:reply(From, {ok, Value}),
    NewQueue.


%% @doc: Connect to server
%% @private
connect(State) ->
    case gen_tcp:connect(State#state.ip, State#state.port, ?SOCKET_OPTS) of
        {ok, Socket} -><commit_msg>impl max request per one session<commit_after>%%======================================================================
%%
%% Leo RPC
%%
%% Copyright (c) 2012-2013 Rakuten, Inc.
%%
%% This file is provided to you 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.
%%
%%======================================================================
-module(leo_rpc_client_conn).

-author('<NAME>').

-behaviour(gen_server).

-include("leo_rpc.hrl").
-include_lib("eunit/include/eunit.hrl").

%% API
-export([start_link/1, stop/1]).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
         terminate/2, code_change/3]).

-record(state, {
          id     :: atom(),
          host   :: string(),
          ip     :: pos_integer(),
          port   :: integer(),
          socket :: reference()|undefined,
          reconnect_sleep :: integer(),
          buf    :: binary(),
          nreq   :: pos_integer(),
          queue = [] :: list()
         }).


-define(SOCKET_OPTS, [binary, {active, once}, {packet, raw}, {reuseaddr, true}]).
-define(RECV_TIMEOUT, 5000).
-define(MAX_REQ_PER_CON, 100).

%% ===================================================================
%% APIs
%% ===================================================================
-spec(start_link(list(any())) ->
             {ok, pid()} | {error, term()}).
start_link([Host, IP, Port, ReconnectSleepInterval]) ->
    gen_server:start_link(?MODULE, [Host, IP, Port, ReconnectSleepInterval], []).

stop(ServerRef) ->
    gen_server:call(ServerRef, stop).


%%====================================================================
%% gen_server callbacks
%%====================================================================
init([Host, IP, Port, ReconnectSleepInterval]) ->
    State = #state{host = Host,
                   ip = IP,
                   port = Port,
                   reconnect_sleep = ReconnectSleepInterval,
                   nreq = 0,
                   buf = <<>>,
                   queue = []},
    case connect(State) of
        {ok, NewState} ->
            {ok, NewState};
        {error, Reason} ->
            {stop, {connection_error, Reason}}
    end.

handle_call({request, Req}, From, State) ->
    exec(Req, From, State);

handle_call(status,_From, #state{socket = Socket} = State) ->
    Ret = (Socket /= undefined),
    {reply, {ok, Ret}, State};

handle_call(stop, _From, State) ->
    {stop, normal, ok, State};

handle_call(_Request, _From, State) ->
    {reply, unknown_request, State}.

handle_cast(_Msg, State) ->
    {noreply, State}.


handle_info({tcp, Socket, Bs}, #state{buf = Buf} = State) ->
    Res = recv(Socket, <<Buf/binary, Bs/binary>>),
    case Res of
        {error, Cause} ->
            {stop, Cause, State};
        {value, Value, Rest} ->
            inet:setopts(Socket, [{active, once}]),
            NewState = State#state{buf = Rest},
            {noreply, handle_response(Value, NewState)}
    end;
handle_info({tcp_error, _Socket, _Reason}, State) ->
    {noreply, State};

handle_info({tcp_closed, _Socket}, State) ->
    case State#state.reconnect_sleep of
        0 ->
            void;
        _ ->
            Self = self(),
            spawn(fun() -> reconnect_loop(Self, State) end)
    end,
    {noreply, State#state{socket = undefined, queue = []}};

handle_info({connection_ready, Socket}, #state{socket = undefined} = State) ->
    {noreply, State#state{socket = Socket}};

handle_info(stop, State) ->
    {stop, shutdown, State};

handle_info(_Info, State) ->
    {stop, {unhandled_message, _Info}, State}.

terminate(_Reason, #state{socket = Socket}) ->
    case Socket of
        undefined ->
            ok;
        Socket ->
            catch gen_tcp:close(Socket),
            ok
    end.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%% ===================================================================
%% Inner Functions
%% ===================================================================
%% @doc: Send the given request to the rpc-server
%% @ptivate
-spec(exec(iolist(), pid(), #state{}) ->
             {noreply, #state{}} | {reply, Reply::any(), #state{}}).
exec(Req, From, #state{socket = undefined} = State) ->
    case connect(State) of
        {ok, #state{socket = Socket} = State1} ->
            case gen_tcp:send(Socket, Req) of
                ok ->
                    NewQueue = [From|State1#state.queue],
                    {noreply, State1#state{queue = NewQueue}};
                {error, Reason} ->
                    {reply, {error, Reason}, State1}
            end;
        {error, Reason} ->
            {reply, {error, Reason}, State}
    end;

exec(Req, From, #state{socket = Socket} = State) ->
    case gen_tcp:send(Socket, Req) of
        ok ->
            NewQueue = [From|State#state.queue],
            {noreply, State#state{queue = NewQueue}};
        {error, Reason} ->
            {reply, {error, Reason}, State}
    end.


%% @doc: Handle the response coming from Server
%% @private
-spec(handle_response(binary(), #state{}) ->
             #state{}).
handle_response(Data, #state{queue  = Queue, 
                             socket = Socket,
                             nreq   = NumReq} = State) ->
    NewQueue = reply(Data, Queue),
    case NumReq of
        ?MAX_REQ_PER_CON ->
            catch gen_tcp:close(Socket),
            State#state{queue = NewQueue, socket = undefined, nreq = 0};
        _ ->
            State#state{queue = NewQueue, nreq = NumReq + 1}
    end.


%% @doc: Send data to the 1st-client in the queue
%% @private
reply(_, []) ->
    error_logger:warning_msg(
      "~p,~p,~p,~p~n",
      [{module, ?MODULE_STRING}, {function, "reply/2"},
       {line, ?LINE}, {body, "Nothing in queue"}]),
    throw(empty_queue);

reply(Value, Queue) ->
    [From|NewQueue] = lists:reverse(Queue),
    gen_server:reply(From, {ok, Value}),
    NewQueue.


%% @doc: Connect to server
%% @private
connect(State) ->
    case gen_tcp:connect(State#state.ip, State#state.port, ?SOCKET_OPTS) of
        {ok, Socket} ->
{'fraction_non_alphanumeric': 0.23145377224759806, 'fraction_numerical': 0.005355174043156403, 'mean_word_length': 2.8089382123575284, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 97, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '213714', 'n_tokens_mistral': 3881, 'n_tokens_neox': 3671, 'n_words': 935}