emqtt 4 (我要publish消息了)

 1 process(Packet = ?PUBLISH_PACKET(_Qos, Topic, _PacketId, _Payload), State) ->
 2     %% ACL check
 3     ...
 4     publish(Packet, State);
 5     ...
 6 publish(Packet = ?PUBLISH_PACKET(?QOS_0, _PacketId),
 7         #proto_state{client_id = ClientId, username = Username, session = Session}) ->
 8     %% 处理packet 获得msg
 9     Msg = emqttd_message:from_packet(Username, ClientId, Packet),
10     %% 调用emqttd_session module的publish/2 函数
11     %% subscribe的时候，也是由protocol 进入的emqttd_session module
12     emqttd_session:publish(Session, Msg);

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

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

 1 %% @doc Publish message to Topic.
 2 -spec(publish(binary(), any()) -> any()).
 3 publish(Topic, Msg) ->
 4     lists:foreach(
 5         fun(#mqtt_route{topic = To, node = Node}) when Node =:= node() ->
 6             ?MODULE:dispatch(To, Msg);
 7            (#mqtt_route{topic = To, node = Node}) ->
 8             rpc:cast(Node, ?MODULE, dispatch, [To, Msg])
 9         end, emqttd_router:lookup(Topic)).
10 
11 dispatch(Topic, Msg) ->
12     case subscribers(Topic) of
13         [] ->
14             dropped(Topic);
15         [SubPid] ->
16             SubPid ! {dispatch, Topic, Msg};
17         SubPids ->
18             lists:foreach(fun(SubPid) ->
19                 SubPid ! {dispatch, Topic, Msg}
20             end, SubPids)
21     end.
22 %% @private
23 %% @doc Find all subscribers
24 subscribers(Topic) ->
25     case ets:member(subscriber, Topic) of
26         true -> %% faster then lookup?
27             try ets:lookup_element(subscriber, Topic, 2) catch error:badarg -> [] end;
28         false ->
29             []
30     end.

 1 %% Dispatch Message
 2 handle_info({dispatch, Topic, Msg}, Session = #session{subscriptions = Subscriptions})
 3     when is_record(Msg, mqtt_message) ->
 4     dispatch(tune_qos(Topic, Msg, Subscriptions), Session);
 5 
 6 %% Deliver qos0 message directly to client
 7 dispatch(Msg = #mqtt_message{qos = ?QOS0}, Session = #session{client_pid = ClientPid}) ->
 8     ClientPid ! {deliver, Msg},
 9     hibernate(Session);
10 dispatch(Msg = #mqtt_message{qos = QoS}, Session = #session{message_queue = MsgQ})
11     when QoS =:= ?QOS1 orelse QoS =:= ?QOS2 ->
12     case check_inflight(Session) of
13         true  ->
14             noreply(deliver(Msg, Session));
15         false ->
16             hibernate(Session#session{message_queue = emqttd_mqueue:in(Msg, MsgQ)})
17     end.