<commit_before>%%-----------------------------------------------------------------------------
%% Copyright (c) 2012-2015, <NAME> <<EMAIL>>
%% 
%% Permission is hereby granted, free of charge, to any person obtaining a copy
%% of this software and associated documentation files (the "Software"), to deal
%% in the Software without restriction, including without limitation the rights
%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
%% copies of the Software, and to permit persons to whom the Software is
%% furnished to do so, subject to the following conditions:
%% 
%% The above copyright notice and this permission notice shall be included in all
%% copies or substantial portions of the Software.
%% 
%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
%% SOFTWARE.
%%------------------------------------------------------------------------------

-module(emqtt_protocol).

-include("emqtt.hrl").

-include("emqtt_packet.hrl").

-record(proto_state, { 
        socket, 
        peer_name,
        connected = false, %received CONNECT action?
        proto_vsn, 
        proto_name,
		packet_id,
		client_id,
		clean_sess,
		will_msg,
		subscriptions,
		awaiting_ack,
		awaiting_rel
}).

-type proto_state() :: #proto_state{}.

-export([initial_state/2]).

-export([handle_packet/2, send_packet/2, client_terminated/1]).

-export([info/1]).

-define(PACKET_TYPE(Packet, Type), 
    Packet = #mqtt_packet { header = #mqtt_packet_header { type = Type }}).

initial_state(Socket, Peername) ->
	#proto_state{
		socket			= Socket,
        peer_name       = Peername,
		packet_id		= 1,
		subscriptions   = [],
		awaiting_ack	= gb_trees:empty(),
		awaiting_rel	= gb_trees:empty()
	}. 

info(#proto_state{ proto_vsn    = ProtoVsn,
                   proto_name   = ProtoName,
                   packet_id	= PacketId,
				   client_id	= ClientId,
				   clean_sess	= CleanSess,
				   will_msg		= WillMsg,
				   subscriptions= Subs }) ->
	[ {packet_id,  PacketId},
      {proto_vsn,  ProtoVsn},
      {proto_name, ProtoName},
	  {client_id,  ClientId},
	  {clean_sess, CleanSess},
	  {will_msg,   WillMsg},
	  {subscriptions, Subs} ].

-spec handle_packet(Packet, State) -> {ok, NewState} | {error, any()} when 
	Packet      :: mqtt_packet(), 
	State       :: proto_state(),
	NewState    :: proto_state().

%%CONNECT – Client requests a connection to a Server

%%A Client can only send the CONNECT Packet once over a Network Connection. 369
handle_packet(?PACKET_TYPE(Packet, ?CONNECT), State = #proto_state{connected = false}) ->
    handle_packet(?CONNECT, Packet, State#proto_state{connected = true});

handle_packet(?PACKET_TYPE(_Packet, ?CONNECT), State = #proto_state{connected = true}) ->
    {error, protocol_bad_connect, State};

%%Received other packets when CONNECT not arrived.
handle_packet(_Packet, State = #proto_state{connected = false}) ->
    {error, protocol_not_connected, State};

handle_packet(?PACKET_TYPE(Packet, Type),
				State = #proto_state { peer_name = PeerName, client_id = ClientId }) ->
	lager:info("RECV from ~s@~s: ~s", [ClientId, PeerName, emqtt_packet:dump(Packet)]),
	case validate_packet(Type, Packet) of	
	ok ->
		handle_packet(Type, Packet, State);
	{error, Reason} ->
		{error, Reason, State}
	end.

handle_packet(?CONNECT, Packet = #mqtt_packet { 
                                    variable = #mqtt_packet_connect { 
                                         username   = Username, 
                                         password   = Password, 
                                         proto_ver  = ProtoVersion, 
                                         clean_sess = CleanSess, 
                                         keep_alive = KeepAlive, 
                                         client_id  = ClientId } = Var }, 
              State = #proto_state{ peer_name = PeerName} ) ->
    lager:info("RECV from ~s@~s: ~s", [ClientId, PeerName, emqtt_packet:dump(Packet)]),
    {ReturnCode, State1} =
        case {lists:member(ProtoVersion, proplists:get_keys(?PROTOCOL_NAMES)),
              valid_client_id(ClientId)} of
            {false, _} ->
                {?CONNACK_PROTO_VER, State#proto_state{client_id = ClientId}};
            {_, false} ->
                {?CONNACK_INVALID_ID, State#proto_state{client_id = ClientId}};
            _ ->
                case emqtt_auth:check(Username, Password) of
                    false ->
                        lager:error("MQTT login failed - no credentials"),
                        {?CONNACK_CREDENTIALS, State#proto_state{client_id = ClientId}};
                    true ->
                        start_keepalive(KeepAlive),
						emqtt_cm:register(ClientId, self()),
						{?CONNACK_ACCEPT,
						 State #proto_state{ will_msg   = make_will_msg(Var),
											 client_id  = ClientId }}
                end
        end,
		send_packet( #mqtt_packet { 
                        header = #mqtt_packet_header { type = ?CONNACK }, 
                        variable = #mqtt_packet_connack{ return_code = ReturnCode }}, State ),
    {ok, State1};

handle_packet(?PUBLISH, Packet = #mqtt_packet {
                                     header = #mqtt_packet_header {qos = ?QOS_0}}, State) ->
	emqtt_router:route(make_message(Packet)),
	{ok, State};

handle_packet(?PUBLISH, Packet = #mqtt_packet { 
                                     header = #mqtt_packet_header { qos    = ?QOS_1 }, 
                                     variable = #mqtt_packet_publish{packet_id = PacketId}}, 
              State) ->
	emqtt_router:route(make_message(Packet)),
	send_packet( make_packet(?PUBACK,  PacketId), State ),
	{ok, State};
<commit_msg>send connack with State1<commit_after>%%-----------------------------------------------------------------------------
%% Copyright (c) 2012-2015, <NAME> <<EMAIL>>
%% 
%% Permission is hereby granted, free of charge, to any person obtaining a copy
%% of this software and associated documentation files (the "Software"), to deal
%% in the Software without restriction, including without limitation the rights
%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
%% copies of the Software, and to permit persons to whom the Software is
%% furnished to do so, subject to the following conditions:
%% 
%% The above copyright notice and this permission notice shall be included in all
%% copies or substantial portions of the Software.
%% 
%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
%% SOFTWARE.
%%------------------------------------------------------------------------------

-module(emqtt_protocol).

-include("emqtt.hrl").

-include("emqtt_packet.hrl").

-record(proto_state, { 
        socket, 
        peer_name,
        connected = false, %received CONNECT action?
        proto_vsn, 
        proto_name,
		packet_id,
		client_id,
		clean_sess,
		will_msg,
		subscriptions,
		awaiting_ack,
		awaiting_rel
}).

-type proto_state() :: #proto_state{}.

-export([initial_state/2]).

-export([handle_packet/2, send_packet/2, client_terminated/1]).

-export([info/1]).

-define(PACKET_TYPE(Packet, Type), 
    Packet = #mqtt_packet { header = #mqtt_packet_header { type = Type }}).

initial_state(Socket, Peername) ->
	#proto_state{
		socket			= Socket,
        peer_name       = Peername,
		packet_id		= 1,
		subscriptions   = [],
		awaiting_ack	= gb_trees:empty(),
		awaiting_rel	= gb_trees:empty()
	}. 

info(#proto_state{ proto_vsn    = ProtoVsn,
                   proto_name   = ProtoName,
                   packet_id	= PacketId,
				   client_id	= ClientId,
				   clean_sess	= CleanSess,
				   will_msg		= WillMsg,
				   subscriptions= Subs }) ->
	[ {packet_id,  PacketId},
      {proto_vsn,  ProtoVsn},
      {proto_name, ProtoName},
	  {client_id,  ClientId},
	  {clean_sess, CleanSess},
	  {will_msg,   WillMsg},
	  {subscriptions, Subs} ].

-spec handle_packet(Packet, State) -> {ok, NewState} | {error, any()} when 
	Packet      :: mqtt_packet(), 
	State       :: proto_state(),
	NewState    :: proto_state().

%%CONNECT – Client requests a connection to a Server

%%A Client can only send the CONNECT Packet once over a Network Connection. 369
handle_packet(?PACKET_TYPE(Packet, ?CONNECT), State = #proto_state{connected = false}) ->
    handle_packet(?CONNECT, Packet, State#proto_state{connected = true});

handle_packet(?PACKET_TYPE(_Packet, ?CONNECT), State = #proto_state{connected = true}) ->
    {error, protocol_bad_connect, State};

%%Received other packets when CONNECT not arrived.
handle_packet(_Packet, State = #proto_state{connected = false}) ->
    {error, protocol_not_connected, State};

handle_packet(?PACKET_TYPE(Packet, Type),
				State = #proto_state { peer_name = PeerName, client_id = ClientId }) ->
	lager:info("RECV from ~s@~s: ~s", [ClientId, PeerName, emqtt_packet:dump(Packet)]),
	case validate_packet(Type, Packet) of	
	ok ->
		handle_packet(Type, Packet, State);
	{error, Reason} ->
		{error, Reason, State}
	end.

handle_packet(?CONNECT, Packet = #mqtt_packet { 
                                    variable = #mqtt_packet_connect { 
                                         username   = Username, 
                                         password   = Password, 
                                         proto_ver  = ProtoVersion, 
                                         clean_sess = CleanSess, 
                                         keep_alive = KeepAlive, 
                                         client_id  = ClientId } = Var }, 
              State = #proto_state{ peer_name = PeerName} ) ->
    lager:info("RECV from ~s@~s: ~s", [ClientId, PeerName, emqtt_packet:dump(Packet)]),
    {ReturnCode, State1} =
        case {lists:member(ProtoVersion, proplists:get_keys(?PROTOCOL_NAMES)),
              valid_client_id(ClientId)} of
            {false, _} ->
                {?CONNACK_PROTO_VER, State#proto_state{client_id = ClientId}};
            {_, false} ->
                {?CONNACK_INVALID_ID, State#proto_state{client_id = ClientId}};
            _ ->
                case emqtt_auth:check(Username, Password) of
                    false ->
                        lager:error("MQTT login failed - no credentials"),
                        {?CONNACK_CREDENTIALS, State#proto_state{client_id = ClientId}};
                    true ->
                        start_keepalive(KeepAlive),
						emqtt_cm:register(ClientId, self()),
						{?CONNACK_ACCEPT,
						 State #proto_state{ will_msg   = make_will_msg(Var),
											 client_id  = ClientId }}
                end
        end,
		send_packet( #mqtt_packet { 
                        header = #mqtt_packet_header { type = ?CONNACK }, 
                        variable = #mqtt_packet_connack{ return_code = ReturnCode }}, State1 ),
    {ok, State1};

handle_packet(?PUBLISH, Packet = #mqtt_packet {
                                     header = #mqtt_packet_header {qos = ?QOS_0}}, State) ->
	emqtt_router:route(make_message(Packet)),
	{ok, State};

handle_packet(?PUBLISH, Packet = #mqtt_packet { 
                                     header = #mqtt_packet_header { qos    = ?QOS_1 }, 
                                     variable = #mqtt_packet_publish{packet_id = PacketId}}, 
              State) ->
	emqtt_router:route(make_message(Packet)),
	send_packet( make_packet(?PUBACK,  PacketId), State ),
	{ok, State};

{'fraction_non_alphanumeric': 0.13183015141540488, 'fraction_numerical': 0.003620803159973667, 'mean_word_length': 2.2881493506493507, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 32, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2955329', 'n_tokens_mistral': 3803, 'n_tokens_neox': 3476, 'n_words': 915}