<commit_before>%% Copyright 2014 Erlio GmbH Basel Switzerland (http://erl.io)
%%
%% 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.

-module(vmq_tcp).
-behaviour(ranch_protocol).
-include_lib("public_key/include/public_key.hrl").
-include("vmq_server.hrl").

-export([start_link/4]).
-export([init/4]).

-record(st, {socket,
             transport, buffer= <<>>,
             parser_state=emqtt_frame:initial_state(),
             session, session_mon,
             proto_tag, bytes_recv={os:timestamp(), 0}}).

-spec start_link(_, _, _, _) -> {'ok', pid()}.
start_link(Ref, Socket, Transport, Opts) ->
    Pid = proc_lib:spawn_link(?MODULE, init, [Ref, Socket, Transport, Opts]),
    {ok, Pid}.

-spec init(_, _, atom() | tuple(), maybe_improper_list()) -> any().
init(Ref, Socket, Transport, Opts) ->
    ok = ranch:accept_ack(Ref),
    {ok, Peer} = apply(Transport, peername, [Socket]),
    {ok, WriterPid} = vmq_writer:start_link(Transport, Socket),
    NewOpts =
    case Transport of
        ranch_ssl ->
            case proplists:get_value(use_identity_as_username, Opts, false) of
                true ->
                    [{preauth, socket_to_common_name(Socket)}|Opts];
                false ->
                    Opts
            end;
        _ ->
            Opts
    end,
    {ok, SessionPid} =
    vmq_session:start_link(Peer, fun(Frame) ->
                                         vmq_writer:send(WriterPid, Frame),
                                         ok
                                 end, NewOpts),
    ok = maybe_tune_buffer_size(Transport, Socket),
    ok = apply(Transport, setopts, [Socket, [{nodelay, true},
                                    {packet, raw},
                                    {active, once}]]),
    process_flag(trap_exit, true),
    vmq_systree:incr_socket_count(),
    loop(#st{socket=Socket,
             transport=Transport,
             session=SessionPid,
             proto_tag=proto_tag(Transport)}).

loop(#st{buffer=_Buffer, socket=Socket, transport=Transport,
            session=SessionPid, parser_state=ParserState,
            proto_tag={Proto, ProtoClosed, ProtoError},
            bytes_recv={TS, V}
        } = State) ->
    apply(Transport, setopts, [Socket, [{active, once}]]),
    receive
        {inet_reply, _, ok} ->
            loop(State);
        {inet_reply, _, Status} ->
            teardown(State, Status);
        {Proto, Socket, Data} ->
            NewParserState = process_bytes(SessionPid, Data, ParserState),
            {M, S, _} = TS,
            NrOfBytes = byte_size(Data),
            NewBytesRecv =
            case os:timestamp() of
                {M, S, _} = NewTS ->
                    {NewTS, V + NrOfBytes};
                NewTS ->
                    vmq_systree:incr_bytes_received(V + NrOfBytes),
                    {NewTS, 0}
            end,
            loop(State#st{parser_state=NewParserState,
                          bytes_recv=NewBytesRecv});
        {ProtoClosed, Socket} ->
            teardown(State, tcp_closed);
        {ProtoError, Socket, Reason} ->
            teardown(State, Reason);
        {'EXIT', _Parent, Reason} ->
            teardown(State, Reason);
        M ->
            exit({received_unknown_msg, M})
    end.

teardown(#st{session=SessionPid, transport=Transport, socket=Socket}, Reason) ->
    case Reason of
        normal ->
            lager:debug("[~p] session stopped", [SessionPid]);
        _ ->
            lager:info("[~p] session stopped
                       abnormally due to ~p", [SessionPid, Reason])
    end,
    fast_close(Transport, Socket).


process_bytes(SessionPid, Bytes, ParserState) ->
    case emqtt_frame:parse(Bytes, ParserState) of
        {more, NewParserState} ->
            NewParserState;
        {ok, #mqtt_frame{} = Frame, Rest} ->
            vmq_session:in(SessionPid, Frame),
            process_bytes(SessionPid, Rest, emqtt_frame:initial_state());
        {error, Reason} ->
            io:format("parse error ~p~n", [Reason]),
            emqtt_frame:initial_state()
    end.

maybe_tune_buffer_size(Transport, Socket) ->
    case vmq_config:get_env(tune_tcp_buffer_size, false) of
        true -> tune_buffer_size(Transport, Socket);
        false -> ok
    end.

tune_buffer_size(ranch_tcp, Sock) ->
    tune_buffer_size_(inet, Sock);<commit_msg>use warning instead of error log<commit_after>%% Copyright 2014 Erlio GmbH Basel Switzerland (http://erl.io)
%%
%% 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.

-module(vmq_tcp).
-behaviour(ranch_protocol).
-include_lib("public_key/include/public_key.hrl").
-include("vmq_server.hrl").

-export([start_link/4]).
-export([init/4]).

-record(st, {socket,
             transport, buffer= <<>>,
             parser_state=emqtt_frame:initial_state(),
             session, session_mon,
             proto_tag, bytes_recv={os:timestamp(), 0}}).

-spec start_link(_, _, _, _) -> {'ok', pid()}.
start_link(Ref, Socket, Transport, Opts) ->
    Pid = proc_lib:spawn_link(?MODULE, init, [Ref, Socket, Transport, Opts]),
    {ok, Pid}.

-spec init(_, _, atom() | tuple(), maybe_improper_list()) -> any().
init(Ref, Socket, Transport, Opts) ->
    ok = ranch:accept_ack(Ref),
    {ok, Peer} = apply(Transport, peername, [Socket]),
    {ok, WriterPid} = vmq_writer:start_link(Transport, Socket),
    NewOpts =
    case Transport of
        ranch_ssl ->
            case proplists:get_value(use_identity_as_username, Opts, false) of
                true ->
                    [{preauth, socket_to_common_name(Socket)}|Opts];
                false ->
                    Opts
            end;
        _ ->
            Opts
    end,
    {ok, SessionPid} =
    vmq_session:start_link(Peer, fun(Frame) ->
                                         vmq_writer:send(WriterPid, Frame),
                                         ok
                                 end, NewOpts),
    ok = maybe_tune_buffer_size(Transport, Socket),
    ok = apply(Transport, setopts, [Socket, [{nodelay, true},
                                    {packet, raw},
                                    {active, once}]]),
    process_flag(trap_exit, true),
    vmq_systree:incr_socket_count(),
    loop(#st{socket=Socket,
             transport=Transport,
             session=SessionPid,
             proto_tag=proto_tag(Transport)}).

loop(#st{buffer=_Buffer, socket=Socket, transport=Transport,
            session=SessionPid, parser_state=ParserState,
            proto_tag={Proto, ProtoClosed, ProtoError},
            bytes_recv={TS, V}
        } = State) ->
    apply(Transport, setopts, [Socket, [{active, once}]]),
    receive
        {inet_reply, _, ok} ->
            loop(State);
        {inet_reply, _, Status} ->
            teardown(State, Status);
        {Proto, Socket, Data} ->
            NewParserState = process_bytes(SessionPid, Data, ParserState),
            {M, S, _} = TS,
            NrOfBytes = byte_size(Data),
            NewBytesRecv =
            case os:timestamp() of
                {M, S, _} = NewTS ->
                    {NewTS, V + NrOfBytes};
                NewTS ->
                    vmq_systree:incr_bytes_received(V + NrOfBytes),
                    {NewTS, 0}
            end,
            loop(State#st{parser_state=NewParserState,
                          bytes_recv=NewBytesRecv});
        {ProtoClosed, Socket} ->
            teardown(State, tcp_closed);
        {ProtoError, Socket, Reason} ->
            teardown(State, Reason);
        {'EXIT', _Parent, Reason} ->
            teardown(State, Reason);
        M ->
            exit({received_unknown_msg, M})
    end.

teardown(#st{session=SessionPid, transport=Transport, socket=Socket}, Reason) ->
    case Reason of
        normal ->
            lager:debug("[~p] session stopped", [SessionPid]);
        _ ->
            lager:warning("[~p] session stopped
                       abnormally due to ~p", [SessionPid, Reason])
    end,
    fast_close(Transport, Socket).


process_bytes(SessionPid, Bytes, ParserState) ->
    case emqtt_frame:parse(Bytes, ParserState) of
        {more, NewParserState} ->
            NewParserState;
        {ok, #mqtt_frame{} = Frame, Rest} ->
            vmq_session:in(SessionPid, Frame),
            process_bytes(SessionPid, Rest, emqtt_frame:initial_state());
        {error, Reason} ->
            io:format("parse error ~p~n", [Reason]),
            emqtt_frame:initial_state()
    end.

maybe_tune_buffer_size(Transport, Socket) ->
    case vmq_config:get_env(tune_tcp_buffer_size, false) of
        true -> tune_buffer_size(Transport, Socket);
        false -> ok
    end.

tune_buffer_size(ranch_tcp, Sock) ->
    tune_buffer_size_(inet, Sock);
{'fraction_non_alphanumeric': 0.12401981015270326, 'fraction_numerical': 0.0024762690879075525, 'mean_word_length': 2.1614481409001955, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 67, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '180000', 'n_tokens_mistral': 2946, 'n_tokens_neox': 2879, 'n_words': 656}