emqx_protocol.erl

%% Copyright (c) 2018 EMQ Technologies Co., Ltd. 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.

-module(emqx_protocol).

-define(LOG_HEADER, "[MQTT]").

-include("emqx.hrl").
-include("emqx_mqtt.hrl").
-include("logger.hrl").

-export([init/2]).
-export([info/1]).
-export([attrs/1]).
-export([caps/1]).
-export([stats/1]).
-export([client_id/1]).
-export([credentials/1]).
-export([parser/1]).
-export([session/1]).
-export([received/2]).
-export([process_packet/2]).
-export([deliver/2]).
-export([send/2]).
-export([shutdown/2]).

-record(pstate, {
          zone,
          sendfun,
          peername,
          peercert,
          proto_ver,
          proto_name,
          client_id,
          is_assigned,
          conn_pid,
          conn_props,
          ack_props,
          username,
          session,
          clean_start,
          topic_aliases,
          packet_size,
          will_topic,
          will_msg,
          keepalive,
          mountpoint,
          is_super,
          is_bridge,
          enable_ban,
          enable_acl,
          recv_stats,
          send_stats,
          connected,
          connected_at,
          ignore_loop,
          topic_alias_maximum
        }).

-type(state() :: #pstate{}).
-export_type([state/0]).

-ifdef(TEST).
-compile(export_all).
-compile(nowarn_export_all).
-endif.

-define(NO_PROPS, undefined).

%%------------------------------------------------------------------------------
%% Init
%%------------------------------------------------------------------------------

-spec(init(map(), list()) -> state()).
init(#{peername := Peername, peercert := Peercert, sendfun := SendFun}, Options) ->
    Zone = proplists:get_value(zone, Options),
    #pstate{zone                = Zone,
            sendfun             = SendFun,
            peername            = Peername,
            peercert            = Peercert,
            proto_ver           = ?MQTT_PROTO_V4,
            proto_name          = <<"MQTT">>,
            client_id           = <<>>,
            is_assigned         = false,
            conn_pid            = self(),
            username            = init_username(Peercert, Options),
            is_super            = false,
            clean_start         = false,
            topic_aliases       = #{},
            packet_size         = emqx_zone:get_env(Zone, max_packet_size),
            mountpoint          = emqx_zone:get_env(Zone, mountpoint),
            is_bridge           = false,
            enable_ban          = emqx_zone:get_env(Zone, enable_ban, false),
            enable_acl          = emqx_zone:get_env(Zone, enable_acl),
            recv_stats          = #{msg => 0, pkt => 0},
            send_stats          = #{msg => 0, pkt => 0},
            connected           = false,
            ignore_loop         = emqx_config:get_env(mqtt_ignore_loop_deliver, false),
            topic_alias_maximum = #{to_client => 0, from_client => 0}}.

init_username(Peercert, Options) ->
    case proplists:get_value(peer_cert_as_username, Options) of
        cn  -> esockd_peercert:common_name(Peercert);
        dn  -> esockd_peercert:subject(Peercert);
        crt -> Peercert;
        _   -> undefined
    end.

set_username(Username, PState = #pstate{username = undefined}) ->
    PState#pstate{username = Username};
set_username(_Username, PState) ->
    PState.

%%------------------------------------------------------------------------------
%% API
%%------------------------------------------------------------------------------

info(PState = #pstate{conn_props    = ConnProps,
                      ack_props     = AckProps,
                      session       = Session,
                      topic_aliases = Aliases,
                      will_msg      = WillMsg,
                      enable_acl    = EnableAcl}) ->
    attrs(PState) ++ [{conn_props, ConnProps},
                      {ack_props, AckProps},
                      {session, Session},
                      {topic_aliases, Aliases},
                      {will_msg, WillMsg},
                      {enable_acl, EnableAcl}].

attrs(#pstate{zone         = Zone,
              client_id    = ClientId,
              username     = Username,
              peername     = Peername,
              peercert     = Peercert,
              clean_start  = CleanStart,
              proto_ver    = ProtoVer,
              proto_name   = ProtoName,
              keepalive    = Keepalive,
              mountpoint   = Mountpoint,
              is_super     = IsSuper,
              is_bridge    = IsBridge,
              connected_at = ConnectedAt}) ->
    [{zone, Zone},
     {client_id, ClientId},
     {username, Username},
     {peername, Peername},
     {peercert, Peercert},
     {proto_ver, ProtoVer},
     {proto_name, ProtoName},
     {clean_start, CleanStart},
     {keepalive, Keepalive},
     {mountpoint, Mountpoint},
     {is_super, IsSuper},
     {is_bridge, IsBridge},
     {connected_at, ConnectedAt}].

caps(#pstate{zone = Zone}) ->
    emqx_mqtt_caps:get_caps(Zone).

client_id(#pstate{client_id = ClientId}) ->
    ClientId.

credentials(#pstate{zone       = Zone,
                    client_id  = ClientId,
                    username   = Username,
                    peername   = Peername}) ->
    #{zone      => Zone,
      client_id => ClientId,
      username  => Username,
      peername  => Peername}.

stats(#pstate{recv_stats = #{pkt := RecvPkt, msg := RecvMsg},
              send_stats = #{pkt := SendPkt, msg := SendMsg}}) ->
    [{recv_pkt, RecvPkt},
     {recv_msg, RecvMsg},
     {send_pkt, SendPkt},
     {send_msg, SendMsg}].

session(#pstate{session = SPid}) ->
    SPid.

parser(#pstate{packet_size = Size, proto_ver = Ver}) ->
    emqx_frame:initial_state(#{max_packet_size => Size, version => Ver}).

%%------------------------------------------------------------------------------
%% Packet Received
%%------------------------------------------------------------------------------

set_protover(?CONNECT_PACKET(#mqtt_packet_connect{
                                proto_ver = ProtoVer}),
             PState) ->
    PState#pstate{ proto_ver = ProtoVer };
set_protover(_Packet, PState) ->
    PState.

-spec(received(emqx_mqtt_types:packet(), state()) ->
    {ok, state()} | {error, term()} | {error, term(), state()} | {stop, term(), state()}).
received(?PACKET(Type), PState = #pstate{connected = false}) when Type =/= ?CONNECT ->
    {error, proto_not_connected, PState};

received(?PACKET(?CONNECT), PState = #pstate{connected = true}) ->
    {error, proto_unexpected_connect, PState};

received(Packet = ?PACKET(Type), PState) ->
    PState1 = set_protover(Packet, PState),
    trace(recv, Packet),
    try emqx_packet:validate(Packet) of
        true ->
            case preprocess_properties(Packet, PState1) of
                {error, ReasonCode} ->
                    {error, ReasonCode, PState1};
                {Packet1, PState2} ->
                    process_packet(Packet1, inc_stats(recv, Type, PState2))
            end
    catch
        error : protocol_error ->
            deliver({disconnect, ?RC_PROTOCOL_ERROR}, PState1),
            {error, protocol_error, PState};
        error : subscription_identifier_invalid ->
            deliver({disconnect, ?RC_SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED}, PState1),
            {error, subscription_identifier_invalid, PState1};
        error : topic_alias_invalid ->
            deliver({disconnect, ?RC_TOPIC_ALIAS_INVALID}, PState1),
            {error, topic_alias_invalid, PState1};
        error : topic_filters_invalid ->
            deliver({disconnect, ?RC_TOPIC_FILTER_INVALID}, PState1),
            {error, topic_filters_invalid, PState1};
        error : topic_name_invalid ->
            deliver({disconnect, ?RC_TOPIC_FILTER_INVALID}, PState1),
            {error, topic_filters_invalid, PState1};
        error : Reason ->
            deliver({disconnect, ?RC_MALFORMED_PACKET}, PState1),
            {error, Reason, PState1}
    end.

%%------------------------------------------------------------------------------
%% Preprocess MQTT Properties
%%------------------------------------------------------------------------------
preprocess_properties(Packet = #mqtt_packet{
                                   variable = #mqtt_packet_connect{
                                                  properties = #{'Topic-Alias-Maximum' := ToClient}
                                              }
                               },
                      PState = #pstate{topic_alias_maximum = TopicAliasMaximum}) ->
    {Packet, PState#pstate{topic_alias_maximum = TopicAliasMaximum#{to_client => ToClient}}};

%% Subscription Identifier
preprocess_properties(Packet = #mqtt_packet{
                                  variable = Subscribe = #mqtt_packet_subscribe{
                                                            properties    = #{'Subscription-Identifier' := SubId},
                                                            topic_filters = TopicFilters
                                                           }
                                 },
                      PState = #pstate{proto_ver = ?MQTT_PROTO_V5}) ->
    TopicFilters1 = [{Topic, SubOpts#{subid => SubId}} || {Topic, SubOpts} <- TopicFilters],
    {Packet#mqtt_packet{variable = Subscribe#mqtt_packet_subscribe{topic_filters = TopicFilters1}}, PState};

%% Topic Alias Mapping
preprocess_properties(#mqtt_packet{
                          variable = #mqtt_packet_publish{
                              properties = #{'Topic-Alias' := 0}}
                      },
                      PState) ->
    deliver({disconnect, ?RC_TOPIC_ALIAS_INVALID}, PState),
    {error, ?RC_TOPIC_ALIAS_INVALID};

preprocess_properties(Packet = #mqtt_packet{
                                  variable = Publish = #mqtt_packet_publish{
                                                          topic_name = <<>>,
                                                          properties = #{'Topic-Alias' := AliasId}}
                                 },
                      PState = #pstate{proto_ver = ?MQTT_PROTO_V5, 
                                       topic_aliases = Aliases, 
                                       topic_alias_maximum = #{from_client := TopicAliasMaximum}}) ->
    case AliasId =< TopicAliasMaximum of
        true ->
            {Packet#mqtt_packet{variable = Publish#mqtt_packet_publish{
                                               topic_name = maps:get(AliasId, Aliases, <<>>)}}, PState};
        false ->
            deliver({disconnect, ?RC_TOPIC_ALIAS_INVALID}, PState),
            {error, ?RC_TOPIC_ALIAS_INVALID}
    end;

preprocess_properties(Packet = #mqtt_packet{
                                   variable = #mqtt_packet_publish{
                                                  topic_name = Topic,
                                                  properties = #{'Topic-Alias' := AliasId}}
                               },
                      PState = #pstate{proto_ver = ?MQTT_PROTO_V5, 
                                       topic_aliases = Aliases,
                                       topic_alias_maximum = #{from_client := TopicAliasMaximum}}) ->
    case AliasId =< TopicAliasMaximum of
        true ->
            {Packet, PState#pstate{topic_aliases = maps:put(AliasId, Topic, Aliases)}};
        false ->
            deliver({disconnect, ?RC_TOPIC_ALIAS_INVALID}, PState),
            {error, ?RC_TOPIC_ALIAS_INVALID}
    end;

preprocess_properties(Packet, PState) ->
    {Packet, PState}.

%%------------------------------------------------------------------------------
%% Process MQTT Packet
%%------------------------------------------------------------------------------
process_packet(?CONNECT_PACKET(
                  #mqtt_packet_connect{proto_name  = ProtoName,
                                       proto_ver   = ProtoVer,
                                       is_bridge   = IsBridge,
                                       clean_start = CleanStart,
                                       keepalive   = Keepalive,
                                       properties  = ConnProps,
                                       client_id   = ClientId,
                                       username    = Username,
                                       password    = Password} = Connect), PState) ->

    NewClientId = maybe_use_username_as_clientid(ClientId, Username, PState),

    emqx_logger:set_metadata_client_id(NewClientId),

    %% TODO: Mountpoint...
    %% Msg -> emqx_mountpoint:mount(MountPoint, Msg)
    WillMsg = make_will_msg(Connect),

    PState1 = set_username(Username,
                           PState#pstate{client_id    = NewClientId,
                                         proto_ver    = ProtoVer,
                                         proto_name   = ProtoName,
                                         clean_start  = CleanStart,
                                         keepalive    = Keepalive,
                                         conn_props   = ConnProps,
                                         will_msg     = WillMsg,
                                         is_bridge    = IsBridge,
                                         connected_at = os:timestamp()}),

    connack(
      case check_connect(Connect, PState1) of
          {ok, PState2} ->
              case authenticate(credentials(PState2), Password) of
                  {ok, IsSuper} ->
                      %% Maybe assign a clientId
                      PState3 = maybe_assign_client_id(PState2#pstate{is_super = IsSuper}),
                      emqx_logger:set_metadata_client_id(PState3#pstate.client_id),
                      %% Open session
                      case try_open_session(PState3) of
                          {ok, SPid, SP} ->
                              PState4 = PState3#pstate{session = SPid, connected = true},
                              ok = emqx_cm:register_connection(client_id(PState4)),
                              true = emqx_cm:set_conn_attrs(client_id(PState4), attrs(PState4)),
                              %% Start keepalive
                              start_keepalive(Keepalive, PState4),
                              %% Success
                              {?RC_SUCCESS, SP, PState4};
                          {error, Error} ->
                              ?LOG(error, "Failed to open session: ~p", [Error]),
                              {?RC_UNSPECIFIED_ERROR, PState1}
                      end;
                  {error, Reason} ->
                      ?LOG(error, "Username '~s' login failed for ~p", [Username, Reason]),
                      {?RC_NOT_AUTHORIZED, PState1}
              end;
          {error, ReasonCode} ->
              {ReasonCode, PState1}
      end);

process_packet(Packet = ?PUBLISH_PACKET(?QOS_0, Topic, _PacketId, _Payload), PState) ->
    case check_publish(Packet, PState) of
        {ok, PState1} ->
            do_publish(Packet, PState1);
        {error, ReasonCode} ->
            ?LOG(warning, "Cannot publish qos0 message to ~s for ~s",
                [Topic, emqx_reason_codes:text(ReasonCode)]),
            {error, ReasonCode, PState}
    end;

process_packet(Packet = ?PUBLISH_PACKET(?QOS_1, Topic, PacketId, _Payload), PState) ->
    case check_publish(Packet, PState) of
        {ok, PState1} ->
            do_publish(Packet, PState1);
        {error, ReasonCode} ->
            ?LOG(warning, "Cannot publish qos1 message to ~s for ~s",
                [Topic, emqx_reason_codes:text(ReasonCode)]),
            deliver({puback, PacketId, ReasonCode}, PState)
    end;

process_packet(Packet = ?PUBLISH_PACKET(?QOS_2, Topic, PacketId, _Payload), PState) ->
    case check_publish(Packet, PState) of
        {ok, PState1} ->
            do_publish(Packet, PState1);
        {error, ReasonCode} ->
            ?LOG(warning, "Cannot publish qos2 message to ~s for ~s",
                [Topic, emqx_reason_codes:text(ReasonCode)]),
            deliver({pubrec, PacketId, ReasonCode}, PState)
    end;

process_packet(?PUBACK_PACKET(PacketId, ReasonCode), PState = #pstate{session = SPid}) ->
    {ok = emqx_session:puback(SPid, PacketId, ReasonCode), PState};

process_packet(?PUBREC_PACKET(PacketId, ReasonCode), PState = #pstate{session = SPid}) ->
    case emqx_session:pubrec(SPid, PacketId, ReasonCode) of
        ok ->
            send(?PUBREL_PACKET(PacketId), PState);
        {error, NotFound} ->
            send(?PUBREL_PACKET(PacketId, NotFound), PState)
    end;

process_packet(?PUBREL_PACKET(PacketId, ReasonCode), PState = #pstate{session = SPid}) ->
    case emqx_session:pubrel(SPid, PacketId, ReasonCode) of
        ok ->
            send(?PUBCOMP_PACKET(PacketId), PState);
        {error, NotFound} ->
            send(?PUBCOMP_PACKET(PacketId, NotFound), PState)
    end;

process_packet(?PUBCOMP_PACKET(PacketId, ReasonCode), PState = #pstate{session = SPid}) ->
    {ok = emqx_session:pubcomp(SPid, PacketId, ReasonCode), PState};

process_packet(?SUBSCRIBE_PACKET(PacketId, Properties, RawTopicFilters),
               PState = #pstate{session = SPid, mountpoint = Mountpoint,
                                proto_ver = ProtoVer, is_bridge = IsBridge,
                                ignore_loop = IgnoreLoop}) ->
    RawTopicFilters1 =  if ProtoVer < ?MQTT_PROTO_V5 ->
                            IfIgnoreLoop = case IgnoreLoop of true -> 1; false -> 0 end,
                            case IsBridge of
                                true -> [{RawTopic, SubOpts#{rap => 1, nl => IfIgnoreLoop}} || {RawTopic, SubOpts} <- RawTopicFilters];
                                false -> [{RawTopic, SubOpts#{rap => 0, nl => IfIgnoreLoop}} || {RawTopic, SubOpts} <- RawTopicFilters]
                            end;
                           true ->
                               RawTopicFilters
                        end,
    case check_subscribe(
           parse_topic_filters(?SUBSCRIBE, RawTopicFilters1), PState) of
        {ok, TopicFilters} ->
            case emqx_hooks:run('client.subscribe', [credentials(PState)], TopicFilters) of
                {ok, TopicFilters1} ->
                    ok = emqx_session:subscribe(SPid, PacketId, Properties,
                                                emqx_mountpoint:mount(Mountpoint, TopicFilters1)),
                    {ok, PState};
                {stop, _} ->
                    ReasonCodes = lists:duplicate(length(TopicFilters),
                                                  ?RC_IMPLEMENTATION_SPECIFIC_ERROR),
                    deliver({suback, PacketId, ReasonCodes}, PState)
            end;
        {error, TopicFilters} ->
            {SubTopics, ReasonCodes} =
                lists:foldr(fun({Topic, #{rc := ?RC_SUCCESS}}, {Topics, Codes}) ->
                                    {[Topic|Topics], [?RC_IMPLEMENTATION_SPECIFIC_ERROR | Codes]};
                                ({Topic, #{rc := Code}}, {Topics, Codes}) ->
                                    {[Topic|Topics], [Code|Codes]}
                            end, {[], []}, TopicFilters),
            ?LOG(warning, "Cannot subscribe ~p for ~p",
                [SubTopics, [emqx_reason_codes:text(R) || R <- ReasonCodes]]),
            deliver({suback, PacketId, ReasonCodes}, PState)
    end;

process_packet(?UNSUBSCRIBE_PACKET(PacketId, Properties, RawTopicFilters),
               PState = #pstate{session = SPid, mountpoint = MountPoint}) ->
    case emqx_hooks:run('client.unsubscribe', [credentials(PState)],
                        parse_topic_filters(?UNSUBSCRIBE, RawTopicFilters)) of
        {ok, TopicFilters} ->
            ok = emqx_session:unsubscribe(SPid, PacketId, Properties,
                                          emqx_mountpoint:mount(MountPoint, TopicFilters)),
            {ok, PState};
        {stop, _Acc} ->
            ReasonCodes = lists:duplicate(length(RawTopicFilters),
                                          ?RC_IMPLEMENTATION_SPECIFIC_ERROR),
            deliver({unsuback, PacketId, ReasonCodes}, PState)
    end;

process_packet(?PACKET(?PINGREQ), PState) ->
    send(?PACKET(?PINGRESP), PState);

process_packet(?DISCONNECT_PACKET(?RC_SUCCESS, #{'Session-Expiry-Interval' := Interval}),
                PState = #pstate{session = SPid, conn_props = #{'Session-Expiry-Interval' := OldInterval}}) ->
    case Interval =/= 0 andalso OldInterval =:= 0 of
        true ->
            deliver({disconnect, ?RC_PROTOCOL_ERROR}, PState),
            {error, protocol_error, PState#pstate{will_msg = undefined}};
        false ->
            emqx_session:update_expiry_interval(SPid, Interval),
            %% Clean willmsg
            {stop, normal, PState#pstate{will_msg = undefined}}
    end;
process_packet(?DISCONNECT_PACKET(?RC_SUCCESS), PState) ->
    {stop, normal, PState#pstate{will_msg = undefined}};
process_packet(?DISCONNECT_PACKET(_), PState) ->
    {stop, normal, PState}.

%%------------------------------------------------------------------------------
%% ConnAck --> Client
%%------------------------------------------------------------------------------

connack({?RC_SUCCESS, SP, PState}) ->
    emqx_hooks:run('client.connected', [credentials(PState), ?RC_SUCCESS, attrs(PState)]),
    deliver({connack, ?RC_SUCCESS, sp(SP)}, update_mountpoint(PState));

connack({ReasonCode, PState = #pstate{proto_ver = ProtoVer}}) ->
    emqx_hooks:run('client.connected', [credentials(PState), ReasonCode, attrs(PState)]),
    ReasonCode1 = if ProtoVer =:= ?MQTT_PROTO_V5 ->
                         ReasonCode;
                     true ->
                         emqx_reason_codes:compat(connack, ReasonCode)
                  end,
    _ = deliver({connack, ReasonCode1}, PState),
    {error, emqx_reason_codes:name(ReasonCode1, ProtoVer), PState}.

%%------------------------------------------------------------------------------
%% Publish Message -> Broker
%%------------------------------------------------------------------------------

do_publish(Packet = ?PUBLISH_PACKET(QoS, PacketId),
           PState = #pstate{session = SPid, mountpoint = MountPoint}) ->
    Msg = emqx_mountpoint:mount(MountPoint,
                                emqx_packet:to_message(credentials(PState), Packet)),
    puback(QoS, PacketId, emqx_session:publish(SPid, PacketId, Msg), PState).

%%------------------------------------------------------------------------------
%% Puback -> Client
%%------------------------------------------------------------------------------

puback(?QOS_0, _PacketId, _Result, PState) ->
    {ok, PState};
puback(?QOS_1, PacketId, [], PState) ->
    deliver({puback, PacketId, ?RC_NO_MATCHING_SUBSCRIBERS}, PState);
puback(?QOS_1, PacketId, [_|_], PState) -> %%TODO: check the dispatch?
    deliver({puback, PacketId, ?RC_SUCCESS}, PState);
puback(?QOS_1, PacketId, {error, ReasonCode}, PState) ->
    deliver({puback, PacketId, ReasonCode}, PState);
puback(?QOS_2, PacketId, [], PState) ->
    deliver({pubrec, PacketId, ?RC_NO_MATCHING_SUBSCRIBERS}, PState);
puback(?QOS_2, PacketId, [_|_], PState) -> %%TODO: check the dispatch?
    deliver({pubrec, PacketId, ?RC_SUCCESS}, PState);
puback(?QOS_2, PacketId, {error, ReasonCode}, PState) ->
    deliver({pubrec, PacketId, ReasonCode}, PState).

%%------------------------------------------------------------------------------
%% Deliver Packet -> Client
%%------------------------------------------------------------------------------

-spec(deliver(tuple(), state()) -> {ok, state()} | {error, term()}).
deliver({connack, ReasonCode}, PState) ->
    send(?CONNACK_PACKET(ReasonCode), PState);

deliver({connack, ?RC_SUCCESS, SP}, PState = #pstate{zone = Zone,
                                                     proto_ver = ?MQTT_PROTO_V5,
                                                     client_id = ClientId,
                                                     conn_props = ConnProps,
                                                     is_assigned = IsAssigned,
                                                     topic_alias_maximum = TopicAliasMaximum}) ->
    ResponseInformation = case maps:find('Request-Response-Information', ConnProps) of
                              {ok, 1} ->
                                  iolist_to_binary(emqx_config:get_env(response_topic_prefix));
                              _ ->
                                  <<>>
                          end,
    #{max_packet_size := MaxPktSize,
      max_qos_allowed := MaxQoS,
      mqtt_retain_available := Retain,
      max_topic_alias := MaxAlias,
      mqtt_shared_subscription := Shared,
      mqtt_wildcard_subscription := Wildcard} = caps(PState),
    Props = #{'Retain-Available' => flag(Retain),
              'Maximum-Packet-Size' => MaxPktSize,
              'Topic-Alias-Maximum' => MaxAlias,
              'Wildcard-Subscription-Available' => flag(Wildcard),
              'Subscription-Identifier-Available' => 1,
              'Response-Information' => ResponseInformation,
              'Shared-Subscription-Available' => flag(Shared)},

    Props1 = if
                MaxQoS =:= ?QOS_2 ->
                    Props;
                true ->
                    maps:put('Maximum-QoS', MaxQoS, Props)
            end,

    Props2 = if IsAssigned ->
                    Props1#{'Assigned-Client-Identifier' => ClientId};
                true -> Props1

             end,

    Props3 = case emqx_zone:get_env(Zone, server_keepalive) of
                 undefined -> Props2;
                 Keepalive -> Props2#{'Server-Keep-Alive' => Keepalive}
             end,

    PState1 = PState#pstate{topic_alias_maximum = TopicAliasMaximum#{from_client => MaxAlias}},

    send(?CONNACK_PACKET(?RC_SUCCESS, SP, Props3), PState1);

deliver({connack, ReasonCode, SP}, PState) ->
    send(?CONNACK_PACKET(ReasonCode, SP), PState);

deliver({publish, PacketId, Msg = #message{headers = Headers}}, 
        PState = #pstate{topic_alias_maximum = #{to_client := TopicAliasMaximum},
                         mountpoint          = MountPoint}) ->
    TopicAlias = maps:get('Topic-Alias', Headers, undefined),
    case TopicAlias =:= undefined 
         orelse (TopicAlias =/= 0 andalso TopicAlias =< TopicAliasMaximum) of
        true ->
            _ = emqx_hooks:run('message.delivered', [credentials(PState)], Msg),
            Msg1 = emqx_message:update_expiry(Msg),
            Msg2 = emqx_mountpoint:unmount(MountPoint, Msg1),
            send(emqx_packet:from_message(PacketId, Msg2), PState);
        false ->
            deliver({disconnect, ?RC_TOPIC_ALIAS_INVALID}, PState),
            {error, topic_alias_invalid}
    end;
    
deliver({puback, PacketId, ReasonCode}, PState) ->
    send(?PUBACK_PACKET(PacketId, ReasonCode), PState);

deliver({pubrel, PacketId}, PState) ->
    send(?PUBREL_PACKET(PacketId), PState);

deliver({pubrec, PacketId, ReasonCode}, PState) ->
    send(?PUBREC_PACKET(PacketId, ReasonCode), PState);

deliver({suback, PacketId, ReasonCodes}, PState = #pstate{proto_ver = ProtoVer}) ->
    send(?SUBACK_PACKET(PacketId,
                        if ProtoVer =:= ?MQTT_PROTO_V5 ->
                               ReasonCodes;
                           true ->
                               [emqx_reason_codes:compat(suback, RC) || RC <- ReasonCodes]
                        end), PState);

deliver({unsuback, PacketId, ReasonCodes}, PState) ->
    send(?UNSUBACK_PACKET(PacketId, ReasonCodes), PState);

%% Deliver a disconnect for mqtt 5.0
deliver({disconnect, ReasonCode}, PState = #pstate{proto_ver = ?MQTT_PROTO_V5}) ->
    send(?DISCONNECT_PACKET(ReasonCode), PState);

deliver({disconnect, _ReasonCode}, PState) ->
    {ok, PState}.

%%------------------------------------------------------------------------------
%% Send Packet to Client

-spec(send(emqx_mqtt_types:packet(), state()) -> {ok, state()} | {error, term()}).
send(Packet = ?PACKET(Type), PState = #pstate{proto_ver = Ver, sendfun = SendFun}) ->
    trace(send, Packet),
    case SendFun(Packet, #{version => Ver}) of
        ok ->
            emqx_metrics:sent(Packet),
            {ok, inc_stats(send, Type, PState)};
        {error, Reason} ->
            {error, Reason}
    end.

%%------------------------------------------------------------------------------
%% Maybe use username replace client id

maybe_use_username_as_clientid(ClientId, undefined, _PState) ->
    ClientId;
maybe_use_username_as_clientid(ClientId, Username, #pstate{zone = Zone}) ->
    case emqx_zone:get_env(Zone, use_username_as_clientid, false) of
        true -> Username;
        false -> ClientId
    end.

%%------------------------------------------------------------------------------
%% Assign a clientId

maybe_assign_client_id(PState = #pstate{client_id = <<>>, ack_props = AckProps}) ->
    ClientId = emqx_guid:to_base62(emqx_guid:gen()),
    AckProps1 = set_property('Assigned-Client-Identifier', ClientId, AckProps),
    PState#pstate{client_id = ClientId, is_assigned = true, ack_props = AckProps1};
maybe_assign_client_id(PState) ->
    PState.

try_open_session(PState = #pstate{zone        = Zone,
                                  client_id   = ClientId,
                                  conn_pid    = ConnPid,
                                  username    = Username,
                                  clean_start = CleanStart,
                                  will_msg    = WillMsg}) ->

    SessAttrs = #{
        zone        => Zone,
        client_id   => ClientId,
        conn_pid    => ConnPid,
        username    => Username,
        clean_start => CleanStart,
        will_msg    => WillMsg
    },
    SessAttrs1 = lists:foldl(fun set_session_attrs/2, SessAttrs, [{max_inflight, PState}, {expiry_interval, PState}]),
    case emqx_sm:open_session(SessAttrs1) of
        {ok, SPid} ->
            {ok, SPid, false};
        Other -> Other
    end.

set_session_attrs({max_inflight, #pstate{proto_ver = ?MQTT_PROTO_V5, conn_props = ConnProps}}, SessAttrs) ->
    maps:put(max_inflight, get_property('Receive-Maximum', ConnProps, 65535), SessAttrs);

set_session_attrs({max_inflight, #pstate{zone = Zone}}, SessAttrs) ->
    maps:put(max_inflight, emqx_zone:get_env(Zone, max_inflight, 65535), SessAttrs);

set_session_attrs({expiry_interval, #pstate{proto_ver = ?MQTT_PROTO_V5, conn_props = ConnProps}}, SessAttrs) ->
    maps:put(expiry_interval, get_property('Session-Expiry-Interval', ConnProps, 0), SessAttrs);

set_session_attrs({expiry_interval, #pstate{zone = Zone, clean_start = CleanStart}}, SessAttrs) ->
    maps:put(expiry_interval, case CleanStart of
                                  true -> 0;
                                  false -> emqx_zone:get_env(Zone, session_expiry_interval, 16#ffffffff)
                              end, SessAttrs);

set_session_attrs(_, SessAttrs) ->
    SessAttrs.

authenticate(Credentials, Password) ->
    case emqx_access_control:authenticate(Credentials, Password) of
        ok -> {ok, false};
        {ok, IsSuper} when is_boolean(IsSuper) ->
            {ok, IsSuper};
        {ok, Result} when is_map(Result) ->
            {ok, maps:get(is_superuser, Result, false)};
        {error, Error} ->
            {error, Error}
    end.

set_property(Name, Value, ?NO_PROPS) ->
    #{Name => Value};
set_property(Name, Value, Props) ->
    Props#{Name => Value}.

get_property(_Name, undefined, Default) ->
    Default;
get_property(Name, Props, Default) ->
    maps:get(Name, Props, Default).

make_will_msg(#mqtt_packet_connect{proto_ver   = ProtoVer,
                                   will_props  = WillProps} = Connect) ->
    emqx_packet:will_msg(if
                             ProtoVer =:= ?MQTT_PROTO_V5 ->
                                 WillDelayInterval = get_property('Will-Delay-Interval', WillProps, 0),
                                 Connect#mqtt_packet_connect{will_props = set_property('Will-Delay-Interval', WillDelayInterval, WillProps)};
                             true ->
                                 Connect
                         end).

%%------------------------------------------------------------------------------
%% Check Packet
%%------------------------------------------------------------------------------

check_connect(Packet, PState) ->
    run_check_steps([fun check_proto_ver/2,
                     fun check_client_id/2,
                     fun check_banned/2], Packet, PState).

check_proto_ver(#mqtt_packet_connect{proto_ver  = Ver,
                                     proto_name = Name}, _PState) ->
    case lists:member({Ver, Name}, ?PROTOCOL_NAMES) of
        true  -> ok;
        false -> {error, ?RC_PROTOCOL_ERROR}
    end.

%% MQTT3.1 does not allow null clientId
check_client_id(#mqtt_packet_connect{proto_ver = ?MQTT_PROTO_V3,
                                     client_id = <<>>}, _PState) ->
    {error, ?RC_CLIENT_IDENTIFIER_NOT_VALID};

%% Issue#599: Null clientId and clean_start = false
check_client_id(#mqtt_packet_connect{client_id   = <<>>,
                                     clean_start = false}, _PState) ->
    {error, ?RC_CLIENT_IDENTIFIER_NOT_VALID};

check_client_id(#mqtt_packet_connect{client_id   = <<>>,
                                     clean_start = true}, _PState) ->
    ok;

check_client_id(#mqtt_packet_connect{client_id = ClientId}, #pstate{zone = Zone}) ->
    Len = byte_size(ClientId),
    MaxLen = emqx_zone:get_env(Zone, max_clientid_len),
    case (1 =< Len) andalso (Len =< MaxLen) of
        true  -> ok;
        false -> {error, ?RC_CLIENT_IDENTIFIER_NOT_VALID}
    end.

check_banned(_Connect, #pstate{enable_ban = false}) ->
    ok;
check_banned(#mqtt_packet_connect{client_id = ClientId, username = Username},
             #pstate{peername = Peername}) ->
    case emqx_banned:check(#{client_id => ClientId,
                             username  => Username,
                             peername  => Peername}) of
        true  -> {error, ?RC_BANNED};
        false -> ok
    end.

check_publish(Packet, PState) ->
    run_check_steps([fun check_pub_caps/2,
                     fun check_pub_acl/2], Packet, PState).

check_pub_caps(#mqtt_packet{header = #mqtt_packet_header{qos = QoS, retain = Retain},
                            variable = #mqtt_packet_publish{ properties = _Properties}},
               #pstate{zone = Zone}) ->
    emqx_mqtt_caps:check_pub(Zone, #{qos => QoS, retain => Retain}).

check_pub_acl(_Packet, #pstate{is_super = IsSuper, enable_acl = EnableAcl})
    when IsSuper orelse (not EnableAcl) ->
    ok;

check_pub_acl(#mqtt_packet{variable = #mqtt_packet_publish{topic_name = Topic}}, PState) ->
    case emqx_access_control:check_acl(credentials(PState), publish, Topic) of
        allow -> ok;
        deny  ->
            {error, ?RC_NOT_AUTHORIZED}
    end.

run_check_steps([], _Packet, PState) ->
    {ok, PState};
run_check_steps([Check|Steps], Packet, PState) ->
    case Check(Packet, PState) of
        ok ->
            run_check_steps(Steps, Packet, PState);
        {ok, PState1} ->
            run_check_steps(Steps, Packet, PState1);
        Error = {error, _RC} ->
            Error
    end.

check_subscribe(TopicFilters, PState = #pstate{zone = Zone}) ->
    case emqx_mqtt_caps:check_sub(Zone, TopicFilters) of
        {ok, TopicFilter1} ->
            check_sub_acl(TopicFilter1, PState);
        {error, TopicFilter1} ->
            {error, TopicFilter1}
    end.

check_sub_acl(TopicFilters, #pstate{is_super = IsSuper, enable_acl = EnableAcl})
    when IsSuper orelse (not EnableAcl) ->
    {ok, TopicFilters};

check_sub_acl(TopicFilters, PState) ->
    Credentials = credentials(PState),
    lists:foldr(
      fun({Topic, SubOpts}, {Ok, Acc}) ->
              case emqx_access_control:check_acl(Credentials, subscribe, Topic) of
                  allow -> {Ok, [{Topic, SubOpts}|Acc]};
                  deny  ->
                      {error, [{Topic, SubOpts#{rc := ?RC_NOT_AUTHORIZED}}|Acc]}
              end
      end, {ok, []}, TopicFilters).

trace(recv, Packet) ->
    ?LOG(debug, "RECV ~s", [emqx_packet:format(Packet)]);
trace(send, Packet) ->
    ?LOG(debug, "SEND ~s", [emqx_packet:format(Packet)]).

inc_stats(recv, Type, PState = #pstate{recv_stats = Stats}) ->
    PState#pstate{recv_stats = inc_stats(Type, Stats)};

inc_stats(send, Type, PState = #pstate{send_stats = Stats}) ->
    PState#pstate{send_stats = inc_stats(Type, Stats)}.

inc_stats(Type, Stats = #{pkt := PktCnt, msg := MsgCnt}) ->
    Stats#{pkt := PktCnt + 1, msg := case Type =:= ?PUBLISH of
                                         true  -> MsgCnt + 1;
                                         false -> MsgCnt
                                     end}.

shutdown(_Reason, #pstate{client_id = undefined}) ->
    ok;
shutdown(_Reason, #pstate{connected = false}) ->
    ok;
shutdown(conflict, _PState) ->
    ok;
shutdown(discard, _PState) ->
    ok;
shutdown(Reason, PState) ->
    ?LOG(info, "Shutdown for ~p", [Reason]),
    emqx_hooks:run('client.disconnected', [credentials(PState), Reason]).

start_keepalive(0, _PState) ->
    ignore;
start_keepalive(Secs, #pstate{zone = Zone}) when Secs > 0 ->
    Backoff = emqx_zone:get_env(Zone, keepalive_backoff, 0.75),
    self() ! {keepalive, start, round(Secs * Backoff)}.

%%-----------------------------------------------------------------------------
%% Parse topic filters
%%-----------------------------------------------------------------------------

parse_topic_filters(?SUBSCRIBE, RawTopicFilters) ->
    [emqx_topic:parse(RawTopic, SubOpts) || {RawTopic, SubOpts} <- RawTopicFilters];

parse_topic_filters(?UNSUBSCRIBE, RawTopicFilters) ->
    lists:map(fun emqx_topic:parse/1, RawTopicFilters).

%%------------------------------------------------------------------------------
%% Update mountpoint

update_mountpoint(PState = #pstate{mountpoint = undefined}) ->
    PState;
update_mountpoint(PState = #pstate{mountpoint = MountPoint}) ->
    PState#pstate{mountpoint = emqx_mountpoint:replvar(MountPoint, credentials(PState))}.

sp(true)  -> 1;
sp(false) -> 0.

flag(false) -> 0;
flag(true)  -> 1.