标签：
这一次，主要分析client subscribe 某个topic 的处理流程。
是的，还是要从protocol开始，至于为什么，之前就说过了。
subscribe 类型的packet的处理是：
1 %% 直接过滤掉topic 为空的情况 2 process(?SUBSCRIBE_PACKET(PacketId, []), State) -> 3 send(?SUBACK_PACKET(PacketId, []), State); 4 5 process(?SUBSCRIBE_PACKET(PacketId, TopicTable), State = #proto_state{session = Session}) -> 6 %% 组装client 信息 7 Client = client(State), 8 %% 检查ACL 9 ... 10 %% session 为clientid 对应的session pid 11 %% TopicTable 为 [{TopicName, QoS}] 12 emqttd_session:subscribe(Session, PacketId, TopicTable) 13 ... 14 ; 15
1、过滤掉topictable 为空的情况
2、组装必要的client 信息，完成ACL检查
3、获取clientid 对应的session pid，并调用emqttd_session:subscribe/3 函数
emqttd_session:subscribe/3 只是一个接口函数，实际的处理逻辑是在emqttd_session 模块的handle_cast callback 中实现。
-spec(subscribe(pid(), mqtt_packet_id(), [{binary(), mqtt_qos()}]) -> ok).
subscribe(SessPid, PacketId, TopicTable) ->
From = self(), %%这里的self 是client process id
AckFun = fun(GrantedQos) ->
From ! {suback, PacketId, GrantedQos}
end,
gen_server2:cast(SessPid, {subscribe, TopicTable, AckFun}).
接口函数的定义如上。
handle_cast callback 的实现如下：
handle_cast({subscribe, TopicTable0, AckFun}, Session = #session{client_id = ClientId,
%% subscription 是dict subscriptions = Subscriptions}) -> %% rewrite topic name 对topic name 做一些处理 Subscriptions1 = lists:foldl( fun({Topic, Qos}, SubDict) -> case dict:find(Topic, SubDict) of {ok, Qos} -> %% 已经存在，并且QoS 未更新，所以什么都不需要做 SubDict; {ok, OldQos} -> %% 已经存在，但是QoS 更新，所以，需要更新一下 emqttd_server:update_subscription(ClientId, Topic, OldQos, Qos), dict:store(Topic, Qos, SubDict); error -> %% 不存在，直接添加 emqttd:subscribe(ClientId, Topic, Qos), dict:store(Topic, Qos, SubDict) end end, Subscriptions, TopicTable),
更新subscribe，也就是调用emqttd_server:update_subscription/4 。
emqttd_server 也是由pool 组织的gen_server进程，主要作用是subscription 的增删改查，subscription 信息是保存在 subscription mnesia table 中的，subscription mnesia table的字段信息如下：
-record(mqtt_subscription, {subid :: binary() | atom(), topic :: binary(), qos = 0 :: 0 | 1 | 2 }).
其中，subid 即为subscriber id，也就是clientid，topic 即为topic的名称。
而，update subscription 的逻辑：
1 %% 外部接口 2 update_subscription(ClientId, Topic, OldQos, NewQos) -> 3 call(server(self()), {update_subscription, ClientId, Topic, ?QOS_I(OldQos), ?QOS_I(NewQos)}). 4 5 handle_call({update_subscription, ClientId, Topic, OldQos, NewQos}, _From, State) -> 6 if_subsciption(State, fun() -> 7 OldSub = #mqtt_subscription{subid = ClientId, topic = Topic, qos = OldQos}, 8 NewSub = #mqtt_subscription{subid = ClientId, topic = Topic, qos = NewQos}, 9 %% 使用事物 10 mnesia:transaction(fun update_subscription_/2, [OldSub, NewSub]), 11 set_subscription_stats() 12 end), ok(State); 13 14 update_subscription_(OldSub, NewSub) -> 15 %% 删除旧的 subscription 16 mnesia:delete_object(subscription, OldSub, write), 17 %% 写入新的 subscription 18 mnesia:write(subscription, NewSub, write).
因为 subscription mnesia table的类型为bag，也就是一个clientid 可能会和多个topic 相对应，所以，不能依据key 进行delete，必须使用delete_object的方式。
create subscribe的处理略微有些绕，不知道是作者有意而为之，还是其他什么原因。
首先，create subscribe的入口函数在emqttd module中，
-spec(subscribe(binary(), binary(), mqtt_qos()) -> {ok, mqtt_qos()}). subscribe(ClientId, Topic, Qos) -> emqttd_server:subscribe(ClientId, Topic, Qos).
在此调用emqttd_server:subscribe/3 函数，并请求emqttd_server 进程，emqttd_server 进程调用handle_call callback 函数，处理请求。
1 %% 外部接口 2 -spec(subscribe(binary(), binary(), mqtt_qos()) -> ok). 3 subscribe(ClientId, Topic, Qos) -> 4 %% 这里的self 是emqttd_session 进程，这个调用是在emqttd_session 5 %% module 中的 handle_cast callback 发起的 6 From = self(), 7 call(server(From), {subscribe, From, ClientId, Topic, ?QOS_I(Qos)}). 8 9 handle_call({subscribe, SubPid, ClientId, Topic, Qos}, _From, State) -> 10 %% call pubsub process 11 pubsub_subscribe_(SubPid, Topic), 12 if_subsciption(State, fun() -> 13 %% 将subscription 信息写入到 subscription mnesia table 中 14 add_subscription_(ClientId, Topic, Qos), 15 set_subscription_stats() 16 end), 17 %% monitor session pid，当起DOWN 之后，去掉subscribe 并移除相关信息 18 ok(monitor_subscriber_(ClientId, SubPid, State)); 19 20 %% @private 21 %% @doc Call pubsub to subscribe 22 pubsub_subscribe_(SubPid, Topic) -> 23 case ets:match(subscribed, {SubPid, Topic}) of 24 [] -> 25 emqttd_pubsub:async_subscribe(Topic, SubPid), 26 ets:insert(subscribed, {SubPid, Topic}); 27 [_] -> 28 false 29 end.
L26处，用了subscribed ets table，记录session pid subscribe 的所有topic，这样在 session pid DOWN的时候，就可以移除所有的topic 中session pid 相关的信息了。
而，emqttd_pubsub 同样是由pool 组织的gen_server 进程。
1 %% 外部接口，发起请求 2 -spec(async_subscribe(binary(), pid()) -> ok). 3 async_subscribe(Topic, SubPid) when is_binary(Topic) -> 4 cast(pick(Topic), {subscribe, Topic, SubPid}). 5 6 handle_cast({subscribe, Topic, SubPid}, State) -> 7 %% 实际的处理函数 8 add_subscriber_(Topic, SubPid), 9 {noreply, setstats(State)}; 10 11 add_subscriber_(Topic, SubPid) -> 12 %% 检查该Topic 是否已经存在 13 %% 若不存在，则先增加{Topic，Node}信息，为多node 场景服务 14 ... 15 ets:insert(subscriber, {Topic, SubPid}). %% 这里的subscriber 是一张ets table，接下来的publish 主要就是用的这张表 ********
至此，subscribe 操作的处理逻辑就ok了。
应该也不需要，只是这代码贴的有点多了。（示意图待补）
标签：
原文地址：http://www.cnblogs.com/--00/p/emqtt_learning_subscribe_topic.html
