yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

2.0 - improve design of session and hook

This commit is contained in:
Feng Lee 2016-08-16 14:29:39 +08:00
parent c72a07dbb8
commit d0be556f33
8 changed files with 108 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
src/emqttd_cli.erl
View File

@ -491,13 +491,13 @@ print(Routes = [#mqtt_route{topic = Topic} | _]) ->
Nodes = [atom_to_list(Node) || #mqtt_route{node = Node} <- Routes],
?PRINT("~s -> ~s~n", [Topic, string:join(Nodes, ",")]);
print(Subscriptions = [#mqtt_subscription{subid = ClientId} | _]) ->
TopicTable = [io_lib:format("~s:~w", [Topic, Qos])
|| #mqtt_subscription{topic = Topic, qos = Qos} <- Subscriptions],
?PRINT("~s -> ~s~n", [ClientId, string:join(TopicTable, ",")]);
%% print(Subscriptions = [#mqtt_subscription{subid = ClientId} | _]) ->
%% TopicTable = [io_lib:format("~s:~w", [Topic, Qos])
%% || #mqtt_subscription{topic = Topic, qos = Qos} <- Subscriptions],
%% ?PRINT("~s -> ~s~n", [ClientId, string:join(TopicTable, ",")]);
print(Topics = [#mqtt_topic{}|_]) ->
foreach(fun print/1, Topics);
%% print(Topics = [#mqtt_topic{}|_]) ->
%% foreach(fun print/1, Topics);
print(#mqtt_plugin{name = Name, version = Ver, descr = Descr, active = Active}) ->
?PRINT("Plugin(~s, version=~s, description=~s, active=~s)~n",
@ -509,8 +509,8 @@ print(#mqtt_client{client_id = ClientId, clean_sess = CleanSess, username = User
[ClientId, CleanSess, Username, emqttd_net:format(Peername),
emqttd_time:now_to_secs(ConnectedAt)]);
print(#mqtt_topic{topic = Topic, flags = Flags}) ->
?PRINT("~s: ~s~n", [Topic, string:join([atom_to_list(F) || F <- Flags], ",")]);
%% print(#mqtt_topic{topic = Topic, flags = Flags}) ->
%% ?PRINT("~s: ~s~n", [Topic, string:join([atom_to_list(F) || F <- Flags], ",")]);
print(#mqtt_route{topic = Topic, node = Node}) ->
?PRINT("~s -> ~s~n", [Topic, Node]);

11
src/emqttd_message.erl
View File

@ -81,8 +81,7 @@ from_packet(#mqtt_packet_connect{client_id = ClientId,
will_msg = Msg}) ->
#mqtt_message{msgid = msgid(Qos),
topic = Topic,
from = ClientId,
sender = Username,
from = {ClientId, Username},
retain = Retain,
qos = Qos,
dup = false,
@ -95,7 +94,7 @@ from_packet(ClientId, Packet) ->
from_packet(Username, ClientId, Packet) ->
Msg = from_packet(Packet),
Msg#mqtt_message{from = ClientId, sender = Username}.
Msg#mqtt_message{from = {ClientId, Username}}.
msgid(?QOS_0) ->
undefined;
@ -150,10 +149,10 @@ unset_flag(retain, Msg = #mqtt_message{retain = true}) ->
unset_flag(Flag, Msg) when Flag =:= dup orelse Flag =:= retain -> Msg.
%% @doc Format MQTT Message
format(#mqtt_message{msgid = MsgId, pktid = PktId, from = From, sender = Sender,
format(#mqtt_message{msgid = MsgId, pktid = PktId, from = {ClientId, Username},
qos = Qos, retain = Retain, dup = Dup, topic =Topic}) ->
io_lib:format("Message(Q~p, R~p, D~p, MsgId=~p, PktId=~p, From=~s, Sender=~s, Topic=~s)",
[i(Qos), i(Retain), i(Dup), MsgId, PktId, From, Sender, Topic]).
io_lib:format("Message(Q~p, R~p, D~p, MsgId=~p, PktId=~p, From=~s/~s, Topic=~s)",
[i(Qos), i(Retain), i(Dup), MsgId, PktId, Username, ClientId, Topic]).
i(true) -> 1;
i(false) -> 0;

2
src/emqttd_mod_presence.erl
View File

@ -45,7 +45,7 @@ on_client_connected(ConnAck, Client = #mqtt_client{client_id = ClientId,
emqttd:publish(emqttd_message:set_flag(sys, Msg)),
{ok, Client}.
on_client_disconnected(Reason, ClientId, Opts) ->
on_client_disconnected(Reason, #mqtt_client{client_id = ClientId}, Opts) ->
Json = mochijson2:encode([{clientid, ClientId},
{reason, reason(Reason)},
{ts, emqttd_time:now_to_secs()}]),

15
src/emqttd_protocol.erl
View File

@ -146,7 +146,7 @@ process(Packet = ?CONNECT_PACKET(Var), State0) ->
State2 = maybe_set_clientid(State1),
%% Start session
case emqttd_sm:start_session(CleanSess, clientid(State2)) of
case emqttd_sm:start_session(CleanSess, {clientid(State2), Username}) of
{ok, Session, SP} ->
%% Register the client
emqttd_cm:reg(client(State2)),
@ -280,10 +280,11 @@ shutdown(conflict, #proto_state{client_id = _ClientId}) ->
%% emqttd_cm:unreg(ClientId);
ignore;
shutdown(Error, State = #proto_state{client_id = ClientId, will_msg = WillMsg}) ->
shutdown(Error, State = #proto_state{will_msg = WillMsg}) ->
?LOG(info, "Shutdown for ~p", [Error], State),
send_willmsg(ClientId, WillMsg),
emqttd:run_hooks('client.disconnected', [Error], ClientId),
Client = client(State),
send_willmsg(Client, WillMsg),
emqttd:run_hooks('client.disconnected', [Error], Client),
%% let it down
%% emqttd_cm:unreg(ClientId).
ok.
@ -301,10 +302,10 @@ maybe_set_clientid(State = #proto_state{client_id = NullId})
maybe_set_clientid(State) ->
State.
send_willmsg(_ClientId, undefined) ->
send_willmsg(_Client, undefined) ->
ignore;
send_willmsg(ClientId, WillMsg) ->
emqttd:publish(WillMsg#mqtt_message{from = ClientId}).
send_willmsg(#mqtt_client{client_id = ClientId, username = Username}, WillMsg) ->
emqttd:publish(WillMsg#mqtt_message{from = {ClientId, Username}}).
start_keepalive(0) -> ignore;

97
src/emqttd_session.erl
View File

@ -77,6 +77,9 @@
%% Old Client Pid that has been kickout
old_client_pid :: pid(),
%% Username
username :: binary() | undefined,
%% Last packet id of the session
packet_id = 1,
@ -136,9 +139,9 @@
"Session(~s): " ++ Format, [State#session.client_id | Args])).
%% @doc Start a session.
-spec(start_link(boolean(), mqtt_client_id(), pid()) -> {ok, pid()} | {error, any()}).
start_link(CleanSess, ClientId, ClientPid) ->
gen_server2:start_link(?MODULE, [CleanSess, ClientId, ClientPid], []).
-spec(start_link(boolean(), {mqtt_client_id(), mqtt_username()}, pid()) -> {ok, pid()} | {error, any()}).
start_link(CleanSess, {ClientId, Username}, ClientPid) ->
gen_server2:start_link(?MODULE, [CleanSess, {ClientId, Username}, ClientPid], []).
%% @doc Resume a session.
-spec(resume(pid(), mqtt_client_id(), pid()) -> ok).
@ -208,10 +211,10 @@ unsubscribe(SessPid, Topics) ->
gen_server2:cast(SessPid, {unsubscribe, Topics}).
%%--------------------------------------------------------------------
%% gen_server callbacks
%% gen_server Callbacks
%%--------------------------------------------------------------------
init([CleanSess, ClientId, ClientPid]) ->
init([CleanSess, {ClientId, Username}, ClientPid]) ->
process_flag(trap_exit, true),
true = link(ClientPid),
SessEnv = emqttd_conf:session(),
@ -219,6 +222,7 @@ init([CleanSess, ClientId, ClientPid]) ->
clean_sess = CleanSess,
client_id = ClientId,
client_pid = ClientPid,
username = Username,
subscriptions = dict:new(),
inflight_queue = [],
max_inflight = get_value(max_inflight, SessEnv, 0),
@ -284,68 +288,67 @@ handle_call({publish, Msg = #mqtt_message{qos = ?QOS_2, pktid = PktId}},
handle_call(Req, _From, State) ->
?UNEXPECTED_REQ(Req, State).
handle_cast({subscribe, RawTopicTable, AckFun}, Session = #session{client_id = ClientId,
%%TODO: 2.0 FIX
handle_cast({subscribe, TopicTable, AckFun}, Session = #session{client_id = ClientId,
username = Username,
subscriptions = Subscriptions}) ->
%% TODO: Ugly...
TopicTable0 = lists:map(fun({T, Q}) ->
{T1, Opts} = emqttd_topic:strip(T),
{T1, [{qos, Q} | Opts]}
end, RawTopicTable),
case emqttd:run_hooks('client.subscribe', [ClientId], TopicTable0) of
{ok, TopicTable} ->
?LOG(info, "Subscribe ~p", [TopicTable], Session),
Subscriptions1 = lists:foldl(
fun({Topic, Opts = [{qos, Qos}|_]}, SubDict) ->
case dict:find(Topic, SubDict) of
{ok, Qos} ->
?LOG(warning, "duplicated subscribe: ~s, qos = ~w", [Topic, Qos], Session),
{GrantedQos, Subscriptions1} =
lists:foldl(fun({RawTopic, Qos}, {QosAcc, SubDict}) ->
{Topic, Opts} = emqttd_topic:strip(RawTopic),
case emqttd:run_hooks('client.subscribe', [{ClientId, Username}], {Topic, Opts}) of
{ok, {Topic1, Opts1}} ->
NewQos = proplists:get_value(qos, Opts1, Qos),
{[NewQos | QosAcc], case dict:find(Topic, SubDict) of
{ok, NewQos} ->
?LOG(warning, "duplicated subscribe: ~s, qos = ~w", [Topic, NewQos], Session),
SubDict;
{ok, OldQos} ->
emqttd:setqos(Topic, ClientId, Qos),
?LOG(warning, "duplicated subscribe ~s, old_qos=~w, new_qos=~w", [Topic, OldQos, Qos], Session),
dict:store(Topic, Qos, SubDict);
emqttd:setqos(Topic, ClientId, NewQos),
?LOG(warning, "duplicated subscribe ~s, old_qos=~w, new_qos=~w", [Topic, OldQos, NewQos], Session),
dict:store(Topic, NewQos, SubDict);
error ->
emqttd:subscribe(Topic, ClientId, Opts),
emqttd:subscribe(Topic1, ClientId, Opts1),
%%TODO: the design is ugly...
%% <MQTT V3.1.1>: 3.8.4
%% Where the Topic Filter is not identical to any existing Subscription’s filter,
%% a new Subscription is created and all matching retained messages are sent.
emqttd_retainer:dispatch(Topic, self()),
dict:store(Topic, Qos, SubDict)
emqttd_retainer:dispatch(Topic1, self()),
dict:store(Topic1, NewQos, SubDict)
end};
{stop, _} ->
?LOG(error, "Cannot subscribe: ~p", [Topic], Session),
{[128 | QosAcc], SubDict}
end
end, Subscriptions, TopicTable),
AckFun([Qos || {_, Qos} <- RawTopicTable]),
emqttd:run_hooks('client.subscribe.after', [ClientId], TopicTable),
end, {[], Subscriptions}, TopicTable),
AckFun(lists:reverse(GrantedQos)),
%%emqttd:run_hooks('client.subscribe.after', [ClientId], TopicTable),
hibernate(Session#session{subscriptions = Subscriptions1});
{stop, TopicTable} ->
?LOG(error, "Cannot subscribe: ~p", [TopicTable], Session),
hibernate(Session)
end;
handle_cast({unsubscribe, RawTopics}, Session = #session{client_id = ClientId,
%%TODO: 2.0 FIX
handle_cast({unsubscribe, Topics}, Session = #session{client_id = ClientId,
username = Username,
subscriptions = Subscriptions}) ->
Topics0 = lists:map(fun(Topic) ->
{T, _Opts} = emqttd_topic:strip(Topic), T
end, RawTopics),
case emqttd:run_hooks('client.unsubscribe', [ClientId], Topics0) of
{ok, Topics} ->
?LOG(info, "unsubscribe ~p", [Topics], Session),
Subscriptions1 = lists:foldl(
fun(Topic, SubDict) ->
case dict:find(Topic, SubDict) of
Subscriptions1 =
lists:foldl(fun(RawTopic, SubDict) ->
{Topic0, _Opts} = emqttd_topic:strip(RawTopic),
case emqttd:run_hooks('client.unsubscribe', [ClientId, Username], Topic0) of
{ok, Topic1} ->
case dict:find(Topic1, SubDict) of
{ok, _Qos} ->
emqttd:unsubscribe(Topic, ClientId),
dict:erase(Topic, SubDict);
emqttd:unsubscribe(Topic1, ClientId),
dict:erase(Topic1, SubDict);
error ->
SubDict
end;
{stop, _} ->
SubDict
end
end, Subscriptions, Topics),
hibernate(Session#session{subscriptions = Subscriptions1});
{stop, Topics} ->
?LOG(info, "Cannot unsubscribe: ~p", [Topics], Session),
hibernate(Session)
end;
handle_cast({destroy, ClientId}, Session = #session{client_id = ClientId}) ->
?LOG(warning, "destroyed", [], Session),

6
src/emqttd_session_sup.erl
View File

@ -29,9 +29,9 @@ start_link() ->
supervisor:start_link({local, ?MODULE}, ?MODULE, []).
%% @doc Start a session
-spec(start_session(boolean(), binary(), pid()) -> {ok, pid()}).
start_session(CleanSess, ClientId, ClientPid) ->
supervisor:start_child(?MODULE, [CleanSess, ClientId, ClientPid]).
-spec(start_session(boolean(), {binary(), binary() | undefined} , pid()) -> {ok, pid()}).
start_session(CleanSess, {ClientId, Username}, ClientPid) ->
supervisor:start_child(?MODULE, [CleanSess, {ClientId, Username}, ClientPid]).
%%--------------------------------------------------------------------
%% Supervisor callbacks

33
src/emqttd_sm.erl
View File

@ -32,7 +32,7 @@
%% API Function Exports
-export([start_link/2]).
-export([start_session/2, lookup_session/1, register_session/3, unregister_session/1]).
-export([start_session/2, lookup_session/1, reg_session/3, unreg_session/1]).
-export([dispatch/3]).
@ -77,10 +77,10 @@ start_link(Pool, Id) ->
gen_server2:start_link({local, ?PROC_NAME(?MODULE, Id)}, ?MODULE, [Pool, Id], []).
%% @doc Start a session
-spec(start_session(boolean(), binary()) -> {ok, pid(), boolean()} | {error, any()}).
start_session(CleanSess, ClientId) ->
-spec(start_session(boolean(), {binary(), binary() | undefined}) -> {ok, pid(), boolean()} | {error, any()}).
start_session(CleanSess, {ClientId, Username}) ->
SM = gproc_pool:pick_worker(?POOL, ClientId),
call(SM, {start_session, {CleanSess, ClientId, self()}}).
call(SM, {start_session, CleanSess, {ClientId, Username}, self()}).
%% @doc Lookup a Session
-spec(lookup_session(binary()) -> mqtt_session() | undefined).
@ -91,18 +91,18 @@ lookup_session(ClientId) ->
end.
%% @doc Register a session with info.
-spec(register_session(binary(), boolean(), [tuple()]) -> true).
register_session(ClientId, CleanSess, Properties) ->
-spec(reg_session(binary(), boolean(), [tuple()]) -> true).
reg_session(ClientId, CleanSess, Properties) ->
ets:insert(mqtt_local_session, {ClientId, self(), CleanSess, Properties}).
%% @doc Unregister a session.
-spec(unregister_session(binary()) -> true).
unregister_session(ClientId) ->
-spec(unreg_session(binary()) -> true).
unreg_session(ClientId) ->
ets:delete(mqtt_local_session, ClientId).
dispatch(ClientId, Topic, Msg) ->
try ets:lookup_element(mqtt_local_session, ClientId, 2) of
Pid -> Pid ! {dispatch, Topic, Msg}
Pid -> Pid ! {deliver, Topic, Msg}
catch
error:badarg -> io:format("Session Not Found: ~p~n", [ClientId]), ok %%TODO: How??
end.
@ -128,11 +128,11 @@ prioritise_info(_Msg, _Len, _State) ->
2.
%% Persistent Session
handle_call({start_session, Client = {false, ClientId, ClientPid}}, _From, State) ->
handle_call({start_session, false, {ClientId, Username}, ClientPid}, _From, State) ->
case lookup_session(ClientId) of
undefined ->
%% Create session locally
create_session(Client, State);
create_session({false, {ClientId, Username}, ClientPid}, State);
Session ->
case resume_session(Session, ClientPid) of
{ok, SessPid} ->
@ -143,7 +143,8 @@ handle_call({start_session, Client = {false, ClientId, ClientPid}}, _From, State
end;
%% Transient Session
handle_call({start_session, Client = {true, ClientId, _ClientPid}}, _From, State) ->
handle_call({start_session, true, {ClientId, Username}, ClientPid}, _From, State) ->
Client = {true, {ClientId, Username}, ClientPid},
case lookup_session(ClientId) of
undefined ->
create_session(Client, State);
@ -195,8 +196,8 @@ code_change(_OldVsn, State, _Extra) ->
%%--------------------------------------------------------------------
%% Create Session Locally
create_session({CleanSess, ClientId, ClientPid}, State) ->
case create_session(CleanSess, ClientId, ClientPid) of
create_session({CleanSess, {ClientId, Username}, ClientPid}, State) ->
case create_session(CleanSess, {ClientId, Username}, ClientPid) of
{ok, SessPid} ->
{reply, {ok, SessPid, false},
monitor_session(ClientId, SessPid, State)};
@ -204,8 +205,8 @@ create_session({CleanSess, ClientId, ClientPid}, State) ->
{reply, {error, Error}, State}
end.
create_session(CleanSess, ClientId, ClientPid) ->
case emqttd_session_sup:start_session(CleanSess, ClientId, ClientPid) of
create_session(CleanSess, {ClientId, Username}, ClientPid) ->
case emqttd_session_sup:start_session(CleanSess, {ClientId, Username}, ClientPid) of
{ok, SessPid} ->
Session = #mqtt_session{client_id = ClientId, sess_pid = SessPid, persistent = not CleanSess},
case insert_session(Session) of

1
src/emqttd_topic.erl
View File

@ -17,6 +17,7 @@
-module(emqttd_topic).
-import(lists, [reverse/1]).
-export([match/2, validate/1, triples/1, words/1, wildcard/1]).
-export([join/1, feed_var/3, systop/1]).