fix(mgmt_client): `nl` not allowed for shared-sub
This commit is contained in:
parent
c1928f874e
commit
c6b1102b2b
|
@ -283,6 +283,10 @@ parse(#share{topic = Topic = <<?QUEUE, "/", _/binary>>}, _Options) ->
|
||||||
error({invalid_topic_filter, Topic});
|
error({invalid_topic_filter, Topic});
|
||||||
parse(#share{topic = Topic = <<?SHARE, "/", _/binary>>}, _Options) ->
|
parse(#share{topic = Topic = <<?SHARE, "/", _/binary>>}, _Options) ->
|
||||||
error({invalid_topic_filter, Topic});
|
error({invalid_topic_filter, Topic});
|
||||||
|
parse(#share{} = T, #{nl := 1} = _Options) ->
|
||||||
|
%% Protocol Error and Should Disconnect
|
||||||
|
%% MQTT-5.0 [MQTT-3.8.3-4] and [MQTT-4.13.1-1]
|
||||||
|
error({invalid_subopts_nl, maybe_format_share(T)});
|
||||||
parse(<<?QUEUE, "/", Topic/binary>>, Options) ->
|
parse(<<?QUEUE, "/", Topic/binary>>, Options) ->
|
||||||
parse(#share{group = <<?QUEUE>>, topic = Topic}, Options);
|
parse(#share{group = <<?QUEUE>>, topic = Topic}, Options);
|
||||||
parse(TopicFilter = <<?SHARE, "/", Rest/binary>>, Options) ->
|
parse(TopicFilter = <<?SHARE, "/", Rest/binary>>, Options) ->
|
||||||
|
|
|
@ -758,6 +758,12 @@ subscribe(#{clientid := ClientID, topic := Topic} = Sub) ->
|
||||||
case do_subscribe(ClientID, Topic, Opts) of
|
case do_subscribe(ClientID, Topic, Opts) of
|
||||||
{error, channel_not_found} ->
|
{error, channel_not_found} ->
|
||||||
{404, ?CLIENTID_NOT_FOUND};
|
{404, ?CLIENTID_NOT_FOUND};
|
||||||
|
{error, invalid_subopts_nl} ->
|
||||||
|
{400, #{
|
||||||
|
code => <<"INVALID_PARAMETER">>,
|
||||||
|
message =>
|
||||||
|
<<"Invalid Subscribe options: `no_local` not allowed for shared-sub. See [MQTT-3.8.3-4]">>
|
||||||
|
}};
|
||||||
{error, Reason} ->
|
{error, Reason} ->
|
||||||
Message = list_to_binary(io_lib:format("~p", [Reason])),
|
Message = list_to_binary(io_lib:format("~p", [Reason])),
|
||||||
{500, #{code => <<"UNKNOWN_ERROR">>, message => Message}};
|
{500, #{code => <<"UNKNOWN_ERROR">>, message => Message}};
|
||||||
|
@ -804,18 +810,25 @@ unsubscribe_batch(#{clientid := ClientID, topics := Topics}) ->
|
||||||
%% internal function
|
%% internal function
|
||||||
|
|
||||||
do_subscribe(ClientID, Topic0, Options) ->
|
do_subscribe(ClientID, Topic0, Options) ->
|
||||||
{Topic, Opts} = emqx_topic:parse(Topic0, Options),
|
try emqx_topic:parse(Topic0, Options) of
|
||||||
TopicTable = [{Topic, Opts}],
|
{Topic, Opts} ->
|
||||||
case emqx_mgmt:subscribe(ClientID, TopicTable) of
|
TopicTable = [{Topic, Opts}],
|
||||||
{error, Reason} ->
|
case emqx_mgmt:subscribe(ClientID, TopicTable) of
|
||||||
{error, Reason};
|
{error, Reason} ->
|
||||||
{subscribe, Subscriptions, Node} ->
|
{error, Reason};
|
||||||
case proplists:is_defined(Topic, Subscriptions) of
|
{subscribe, Subscriptions, Node} ->
|
||||||
true ->
|
case proplists:is_defined(Topic, Subscriptions) of
|
||||||
{ok, Options#{node => Node, clientid => ClientID, topic => Topic0}};
|
true ->
|
||||||
false ->
|
{ok, Options#{node => Node, clientid => ClientID, topic => Topic0}};
|
||||||
{error, unknow_error}
|
false ->
|
||||||
|
{error, unknow_error}
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
catch
|
||||||
|
error:{invalid_subopts_nl, _} ->
|
||||||
|
{error, invalid_subopts_nl};
|
||||||
|
_:Reason ->
|
||||||
|
{error, Reason}
|
||||||
end.
|
end.
|
||||||
|
|
||||||
-spec do_unsubscribe(emqx_types:clientid(), emqx_types:topic()) ->
|
-spec do_unsubscribe(emqx_types:clientid(), emqx_types:topic()) ->
|
||||||
|
|
Loading…
Reference in New Issue