<commit_before>%%%===================================================================
%%% @copyright (C) 2011-2012, Erlang Solutions Ltd.
%%% @doc Module abstracting TCP connection to XMPP server
%%% @end
%%%===================================================================

-module(escalus_bosh).
-behaviour(gen_server).

-include_lib("exml/include/exml_stream.hrl").
-include("include/escalus.hrl").

%% API exports
-export([connect/1,
         send/2,
         is_connected/1,
         upgrade_to_tls/2,
         use_zlib/2,
         get_transport/1,
         reset_parser/1,
         stop/1]).

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

-define(WAIT_FOR_SOCKET_CLOSE_TIMEOUT, 200).
-define(SERVER, ?MODULE).

-record(state, {owner, url, parser, sid = nil, rid = nil}).

%%%===================================================================
%%% API
%%%===================================================================

-spec connect({binary(), integer()}) -> {ok, #transport{}}.
connect(Args) ->
    ssl:start(),
    lhttpc:start(),
    {ok, Pid} = gen_server:start_link(?MODULE, [Args, self()], []),
    Transport = gen_server:call(Pid, get_transport),
    {ok, Transport}.

send(#transport{rcv_pid = Pid} = Socket, Elem) ->
    gen_server:cast(Pid, {send, Socket, Elem}).

is_connected(#transport{rcv_pid = Pid}) ->
    erlang:is_process_alive(Pid).

reset_parser(#transport{rcv_pid = Pid}) ->
    gen_server:cast(Pid, reset_parser).

stop(#transport{rcv_pid = Pid}) ->
    try
        gen_server:call(Pid, stop)
    catch
        exit:{noproc, {gen_server, call, _}} ->
            already_stopped
    end.

upgrade_to_tls(#transport{} = _Conn, _Props) ->
    not_supported.

use_zlib(#transport{} = _Conn, _Props) ->
    not_supported.

get_transport(#transport{rcv_pid = Pid}) ->
    gen_server:call(Pid, get_transport).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

init([Args, Owner]) ->
    Host = proplists:get_value(host, Args, <<"localhost">>),
    Port = proplists:get_value(port, Args, 5280),
    Path = proplists:get_value(path, Args, <<"/http-bind">>),
    HostStr = binary_to_list(Host),
    {MS, S, MMS} = now(),
    InitRid = MS * 1000000 * 1000000 + S * 1000000 + MMS,
    {ok, Parser} = exml_stream:new_parser(),
    {ok, #state{owner = Owner,
                url = {HostStr, Port, binary_to_list(Path)},
                parser = Parser, rid=InitRid}}.

handle_call(get_transport, _From, State) ->
    {reply, transport(State), State};
handle_call(stop, _From, #state{} = State) ->
    StreamEnd = escalus_stanza:stream_end(),
    send0(transport(State), exml:to_iolist(StreamEnd), State),
    {stop, normal, ok, State}.
handle_cast(stop, State) ->
    {stop, normal, State};
handle_cast({send, Transport, #xmlstreamstart{} = Elem}, State) ->
    send0(Transport, Elem, State),
    {noreply, State};
handle_cast({send, Transport, Elem}, State) ->
    send0(Transport, Elem, State),
    {noreply, State#state{rid=State#state.rid+1}};
handle_cast(reset_parser, #state{parser = Parser} = State) ->
    {ok, NewParser} = exml_stream:reset_parser(Parser),
    {noreply, State#state{parser = NewParser}}.

%% Handle async HTTP request replies.
handle_info({http_reply, {{_StatusCode, _Reason}, _Hdrs, RespBody}}, State) ->
    NewState = handle_data(RespBody, State),
    {noreply, NewState};
handle_info(_, State) ->
    {noreply, State}.

terminate(_Reason, #state{parser = Parser}) ->
    exml_stream:free_parser(Parser).

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

%%%===================================================================
%%% Helpers
%%%===================================================================
send0(#transport{socket = {Host, Port, Path}}, Elem0, State) ->
    Headers = [{"Content-Type", "text/xml; charset=utf-8"}],
    Elem = wrap_elem(Elem0, State),
    Self = self(),
    AsyncReq = fun() ->
        {ok, Reply} =
            lhttpc:request(Host, Port, false, Path, 'POST',
                Headers, exml:to_iolist(Elem), infinity, []),
        Self ! {http_reply, Reply}
    end,
    spawn_link(AsyncReq).

handle_data(Data, #state{owner = Owner, rid = Rid} = State) ->
    {ok, Body} = exml:parse(Data),
    NewState = case State#state.sid of
        %% First reply for this transport, set sid
        nil ->
            State#state{rid = Rid + 1,
                        sid = exml_query:attr(Body, <<"sid">>)};
        _ ->
            State<commit_msg>Properly update rid when issuing async requests<commit_after>%%%===================================================================
%%% @copyright (C) 2011-2012, Erlang Solutions Ltd.
%%% @doc Module abstracting TCP connection to XMPP server
%%% @end
%%%===================================================================

-module(escalus_bosh).
-behaviour(gen_server).

-include_lib("exml/include/exml_stream.hrl").
-include("include/escalus.hrl").

%% API exports
-export([connect/1,
         send/2,
         is_connected/1,
         upgrade_to_tls/2,
         use_zlib/2,
         get_transport/1,
         reset_parser/1,
         stop/1]).

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

-define(WAIT_FOR_SOCKET_CLOSE_TIMEOUT, 200).
-define(SERVER, ?MODULE).

-record(state, {owner, url, parser, sid = nil, rid = nil}).

%%%===================================================================
%%% API
%%%===================================================================

-spec connect({binary(), integer()}) -> {ok, #transport{}}.
connect(Args) ->
    ssl:start(),
    lhttpc:start(),
    {ok, Pid} = gen_server:start_link(?MODULE, [Args, self()], []),
    Transport = gen_server:call(Pid, get_transport),
    {ok, Transport}.

send(#transport{rcv_pid = Pid} = Socket, Elem) ->
    gen_server:cast(Pid, {send, Socket, Elem}).

is_connected(#transport{rcv_pid = Pid}) ->
    erlang:is_process_alive(Pid).

reset_parser(#transport{rcv_pid = Pid}) ->
    gen_server:cast(Pid, reset_parser).

stop(#transport{rcv_pid = Pid}) ->
    try
        gen_server:call(Pid, stop)
    catch
        exit:{noproc, {gen_server, call, _}} ->
            already_stopped
    end.

upgrade_to_tls(#transport{} = _Conn, _Props) ->
    not_supported.

use_zlib(#transport{} = _Conn, _Props) ->
    not_supported.

get_transport(#transport{rcv_pid = Pid}) ->
    gen_server:call(Pid, get_transport).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

init([Args, Owner]) ->
    Host = proplists:get_value(host, Args, <<"localhost">>),
    Port = proplists:get_value(port, Args, 5280),
    Path = proplists:get_value(path, Args, <<"/http-bind">>),
    HostStr = binary_to_list(Host),
    {MS, S, MMS} = now(),
    InitRid = MS * 1000000 * 1000000 + S * 1000000 + MMS,
    {ok, Parser} = exml_stream:new_parser(),
    {ok, #state{owner = Owner,
                url = {HostStr, Port, binary_to_list(Path)},
                parser = Parser, rid=InitRid}}.

handle_call(get_transport, _From, State) ->
    {reply, transport(State), State};
handle_call(stop, _From, #state{} = State) ->
    StreamEnd = escalus_stanza:stream_end(),
    NewState = send0(transport(State), exml:to_iolist(StreamEnd), State),
    {stop, normal, ok, NewState}.
handle_cast(stop, State) ->
    {stop, normal, State};
handle_cast({send, Transport, #xmlstreamstart{} = Elem}, State) ->
    NewState = send0(Transport, Elem, State),
    {noreply, NewState};
handle_cast({send, Transport, Elem}, State) ->
    NewState = send0(Transport, Elem, State),
    {noreply, NewState};
handle_cast(reset_parser, #state{parser = Parser} = State) ->
    {ok, NewParser} = exml_stream:reset_parser(Parser),
    {noreply, State#state{parser = NewParser}}.

%% Handle async HTTP request replies.
handle_info({http_reply, {{_StatusCode, _Reason}, _Hdrs, RespBody}}, State) ->
    NewState = handle_data(RespBody, State),
    {noreply, NewState};
handle_info(_, State) ->
    {noreply, State}.

terminate(_Reason, #state{parser = Parser}) ->
    exml_stream:free_parser(Parser).

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

%%%===================================================================
%%% Helpers
%%%===================================================================
send0(#transport{socket = {Host, Port, Path}}, Elem0, State) ->
    Headers = [{"Content-Type", "text/xml; charset=utf-8"}],
    Elem = wrap_elem(Elem0, State),
    Self = self(),
    AsyncReq = fun() ->
        {ok, Reply} =
            lhttpc:request(Host, Port, false, Path, 'POST',
                Headers, exml:to_iolist(Elem), infinity, []),
        Self ! {http_reply, Reply}
    end,
    spawn_link(AsyncReq),
    State#state{rid=State#state.rid+1}.

handle_data(Data, #state{owner = Owner} = State) ->
    {ok, Body} = exml:parse(Data),
    NewState = case State#state.sid of
        %% First reply for this transport, set sid
        nil ->
            State#state{sid = exml_query:attr(Body, <<"sid">>)};
        _ ->
            State
{'fraction_non_alphanumeric': 0.2637304041804415, 'fraction_numerical': 0.012477338167857524, 'mean_word_length': 3.869158878504673, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 67, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 22}, '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': '1417808', 'n_tokens_mistral': 3163, 'n_tokens_neox': 2963, 'n_words': 588}