这一次，主要分析client subscribe 某个topic 的处理流程。
由protocol开始
是的，还是要从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 模块处理
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
更新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的方式。
创建subscribe
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了。
总结
应该也不需要，只是这代码贴的有点多了。（示意图待补）