emqtt 4 (我要publish消息了)

這次，分析處理publish msg的流程。

由protocol開始

publish 類型的packet的處理是：

process(Packet = ?PUBLISH_PACKET(_Qos, Topic, _PacketId, _Payload), State) ->
    %% ACL check
    ...
    publish(Packet, State);
    ...
publish(Packet = ?PUBLISH_PACKET(?QOS_0, _PacketId),
        #proto_state{client_id = ClientId, username = Username, session = Session}) ->
    %% 處理packet 獲得msg
    Msg = emqttd_message:from_packet(Username, ClientId, Packet),
    %% 調用emqttd_session module的publish/2 函數
    %% subscribe的時候，也是由protocol 進入的emqttd_session module
    emqttd_session:publish(Session, Msg);

1、ACL 檢查

2、處理packet 獲得msg

3、調用session module進行處理

emqttd_session 模塊處理

和subscribe的處理流程類似，emqttd_session:publish/2 也只是一個接口函數，該函數要根據QoS的不同，決定是

1. 自己調用后續函數完成處理
2. call session process 完成后續處理

%% @doc Publish message
-spec(publish(pid(), mqtt_message()) -> ok | {error, any()}).
publish(_SessPid, Msg = #mqtt_message{qos = ?QOS_0}) ->
    %% publish qos0 directly
    emqttd:publish(Msg);
publish(_SessPid, Msg = #mqtt_message{qos = ?QOS_1}) ->
    %% publish qos1 directly, and client will puback automatically
    emqttd:publish(Msg);
publish(SessPid, Msg = #mqtt_message{qos = ?QOS_2}) ->
    %% publish qos2 by session 
    gen_server2:call(SessPid, {publish, Msg}, ?PUBSUB_TIMEOUT). 

直接處理

如果是自己調用后續函數完成處理的話，則繼續調用emqttd:publish/2，則在emqttd module 中繼續調用emqttd_server:publish/1：

%% @doc Publish a Message
-spec(publish(Msg :: mqtt_message()) -> any()).
publish(Msg = #mqtt_message{from = From}) ->
    ...
    %% 處理topic
    ...
    %% pulish
    emqttd_pubsub:publish(Topic, Msg2),
    ...

 

還是subscribe處理套路：

emqttd_protocol ---> emqttd_session ---> emqttd ---> emqttd_server ---> emqttd_pubsub

在emqttd_pubsub module中的處理是：

%% @doc Publish message to Topic.
-spec(publish(binary(), any()) -> any()).
publish(Topic, Msg) ->
    lists:foreach(
        fun(#mqtt_route{topic = To, node = Node}) when Node =:= node() ->
            ?MODULE:dispatch(To, Msg);
           (#mqtt_route{topic = To, node = Node}) ->
            rpc:cast(Node, ?MODULE, dispatch, [To, Msg])
        end, emqttd_router:lookup(Topic)).

dispatch(Topic, Msg) ->
    case subscribers(Topic) of
        [] ->
            dropped(Topic);
        [SubPid] ->
            SubPid ! {dispatch, Topic, Msg};
        SubPids ->
            lists:foreach(fun(SubPid) ->
                SubPid ! {dispatch, Topic, Msg}
            end, SubPids)
    end.
%% @private
%% @doc Find all subscribers
subscribers(Topic) ->
    case ets:member(subscriber, Topic) of
        true -> %% faster then lookup?
            try ets:lookup_element(subscriber, Topic, 2) catch error:badarg -> [] end;
        false ->
            []
    end.

 

至此，msg 就已經以{dispatch, Topic, Msg}的形式發送給 clientid 對應的session process了。

那么，就需要在emqttd_session module中的handle_info callback 函數處進行處理：

%% Dispatch Message
handle_info({dispatch, Topic, Msg}, Session = #session{subscriptions = Subscriptions})
    when is_record(Msg, mqtt_message) ->
    dispatch(tune_qos(Topic, Msg, Subscriptions), Session);

%% Deliver qos0 message directly to client
dispatch(Msg = #mqtt_message{qos = ?QOS0}, Session = #session{client_pid = ClientPid}) ->
    ClientPid ! {deliver, Msg},
    hibernate(Session);
dispatch(Msg = #mqtt_message{qos = QoS}, Session = #session{message_queue = MsgQ})
    when QoS =:= ?QOS1 orelse QoS =:= ?QOS2 ->
    case check_inflight(Session) of
        true  ->
            noreply(deliver(Msg, Session));
        false ->
            hibernate(Session#session{message_queue = emqttd_mqueue:in(Msg, MsgQ)})
    end.

 

繼而，將信息發送給socket controlling process，然后根據QoS的不同，判斷是否需要等待ack。

總結

（流程示意圖待補）