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

Fix the 'subscriptions' CLI

This commit is contained in:
Feng Lee 2017-11-18 11:17:19 +08:00
parent c751e958db
commit 405d5d9d29
1 changed files with 28 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
src/emqttd_cli.erl
View File

@ -243,25 +243,32 @@ subscriptions(["list"]) ->
end, ets:tab2list(mqtt_subscription)); end, ets:tab2list(mqtt_subscription));
subscriptions(["show", ClientId]) -> subscriptions(["show", ClientId]) ->
case ets:lookup(mqtt_subscription, bin(ClientId)) of case emqttd:subscriptions(bin(ClientId)) of
[] -> ?PRINT_MSG("Not Found.~n"); [] ->
Records -> [print(subscription, Subscription) || Subscription <- Records] ?PRINT_MSG("Not Found.~n");
Subscriptions ->
[print(subscription, Sub) || Sub<- Subscriptions]
end; end;
subscriptions(["add", ClientId, Topic, QoS]) -> subscriptions(["add", ClientId, Topic, QoS]) ->
Add = fun(IntQos) -> if_valid_qos(QoS, fun(IntQos) ->
case emqttd:subscribe(bin(Topic), bin(ClientId), [{qos, IntQos}]) of case emqttd_sm:lookup_session(bin(ClientId)) of
ok -> undefined ->
?PRINT_MSG("ok~n"); ?PRINT_MSG("Error: Session not found!");
{error, Reason} -> #mqtt_session{sess_pid = SessPid} ->
?PRINT("Error: ~p~n", [Reason]) emqttd_session:subscribe(SessPid, [{bin(Topic), IntQos}]),
?PRINT_MSG("ok~n")
end end
end, end);
if_valid_qos(QoS, Add);
subscriptions(["del", ClientId, Topic]) -> subscriptions(["del", ClientId, Topic]) ->
Ok = emqttd:unsubscribe(bin(Topic), bin(ClientId)), case emqttd_sm:lookup_session(bin(ClientId)) of
?PRINT("~p~n", [Ok]); undefined ->
?PRINT_MSG("Error: Session not found!");
#mqtt_session{sess_pid = SessPid} ->
emqttd_session:unsubscribe(SessPid, [bin(Topic)]),
?PRINT_MSG("ok~n")
end;
subscriptions(_) -> subscriptions(_) ->
?USAGE([{"subscriptions list", "List all subscriptions"}, ?USAGE([{"subscriptions list", "List all subscriptions"},
@ -570,14 +577,16 @@ print({ClientId, _ClientPid, _Persistent, SessInfo}) ->
"deliver_msg=~w, enqueue_msg=~w, created_at=~w)~n", "deliver_msg=~w, enqueue_msg=~w, created_at=~w)~n",
[ClientId | [format(Key, get_value(Key, Data)) || Key <- InfoKeys]]). [ClientId | [format(Key, get_value(Key, Data)) || Key <- InfoKeys]]).
print(subscription, {Sub, {_Share, Topic}}) when is_pid(Sub) -> print(subscription, {Sub, {share, _Share, Topic}}) when is_pid(Sub) ->
?PRINT("~p -> ~s~n", [Sub, Topic]); ?PRINT("~p -> ~s~n", [Sub, Topic]);
print(subscription, {Sub, Topic}) when is_pid(Sub) -> print(subscription, {Sub, Topic}) when is_pid(Sub) ->
?PRINT("~p -> ~s~n", [Sub, Topic]); ?PRINT("~p -> ~s~n", [Sub, Topic]);
print(subscription, {Sub, {_Share, Topic}}) -> print(subscription, {{SubId, SubPid}, {share, _Share, Topic}})
?PRINT("~s -> ~s~n", [Sub, Topic]); when is_binary(SubId), is_pid(SubPid) ->
print(subscription, {Sub, Topic}) -> ?PRINT("~s~p -> ~s~n", [SubId, SubPid, Topic]);
?PRINT("~s -> ~s~n", [Sub, Topic]). print(subscription, {{SubId, SubPid}, Topic})
when is_binary(SubId), is_pid(SubPid) ->
?PRINT("~s~p -> ~s~n", [SubId, SubPid, Topic]).
format(created_at, Val) -> format(created_at, Val) ->
emqttd_time:now_secs(Val); emqttd_time:now_secs(Val);