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

Merge pull request #7599 from HJianBo/merge-main-v4.3-into-v4.4 

Merge main v4.3 into v4.4
This commit is contained in:
Xinyu Liu 2022-04-13 09:17:32 +08:00 committed by GitHub
parent c28e50b6a7 fab71ddcfe
commit a42de63791
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 22 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
apps/emqx_bridge_mqtt/priv/emqx_bridge_mqtt.schema
View File

@ -214,8 +214,19 @@
end, end,
Subscriptions = fun(Name) -> Subscriptions = fun(Name) ->
Configs = cuttlefish_variable:filter_by_prefix("bridge.mqtt." ++ Name ++ ".subscription", Conf), Configs = cuttlefish_variable:filter_by_prefix("bridge.mqtt." ++ Name ++ ".subscription", Conf),
lists:zip([Topic || {_, Topic} <- lists:sort([{I, Topic} || {[_, _, _, "subscription", I, "topic"], Topic} <- Configs])], Configs = cuttlefish_variable:filter_by_prefix("bridge.mqtt." ++ Name ++ ".subscription", Conf),
[QoS || {_, QoS} <- lists:sort([{I, QoS} || {[_, _, _, "subscription", I, "qos"], QoS} <- Configs])]) SubConfId = lists:usort([{T, I} || {[_, _, _, "subscription", I, "topic"], T} <- Configs]),
FindQoS = fun({T, I}) ->
QoSFilter = fun
({[_, _, _, "subscription", Index, "qos"], QoS}) -> I == Index andalso {true, QoS};
(_Topic) -> false
end,
case lists:filtermap(QoSFilter, Configs) of
[] -> {T, 0};
QoSList -> {T, hd(QoSList)}
end
end,
[FindQoS(TopicAndIndex) || TopicAndIndex <- SubConfId]
end, end,
IsNodeAddr = fun(Addr) -> IsNodeAddr = fun(Addr) ->
case string:tokens(Addr, "@") of case string:tokens(Addr, "@") of

2
priv/emqx.schema
View File

@ -2434,7 +2434,7 @@ end}.
{translation, "emqx.shared_subscription_strategy_per_group", fun(Conf) -> {translation, "emqx.shared_subscription_strategy_per_group", fun(Conf) ->
Conf0 = cuttlefish_variable:filter_by_prefix("broker", Conf), Conf0 = cuttlefish_variable:filter_by_prefix("broker", Conf),
Groups = lists:filtermap(fun({["broker", Group, "shared_subscription_strategy"], Strategy}) -> Groups = lists:filtermap(fun({["broker", Group, "shared_subscription_strategy"], Strategy}) ->
{true, {Group, Strategy}}; {true, {Group, list_to_binary(Strategy)}};
(_) -> (_) ->
false false
end, Conf0), end, Conf0),

10
test/emqx_frame_SUITE.erl
View File

@ -46,6 +46,8 @@ groups() ->
t_parse_frame_malformed_variable_byte_integer, t_parse_frame_malformed_variable_byte_integer,
t_parse_frame_variable_byte_integer, t_parse_frame_variable_byte_integer,
t_parse_malformed_utf8_string, t_parse_malformed_utf8_string,
t_parse_empty_topic_name,
t_parse_empty_topic_name_with_alias,
t_parse_frame_proxy_protocol %% proxy_protocol_config_disabled packet. t_parse_frame_proxy_protocol %% proxy_protocol_config_disabled packet.
]}, ]},
{connect, [parallel], {connect, [parallel],
@ -170,8 +172,12 @@ t_parse_empty_topic_name(_) ->
t_parse_empty_topic_name_with_alias(_) -> t_parse_empty_topic_name_with_alias(_) ->
Props = #{'Topic-Alias' => 16#AB}, Props = #{'Topic-Alias' => 16#AB},
Packet = ?PUBLISH_PACKET(?QOS_1, <<>>, 1, Props, <<>>), Packet = ?PUBLISH_PACKET(?QOS_1, <<>>, 1, Props, <<>>),
?assertEqual(Packet, parse_serialize(Packet, #{strict_mode => false})), ?assertEqual(
?assertEqual(Packet, parse_serialize(Packet, #{strict_mode => true})). Packet, parse_serialize(Packet, #{strict_mode => false, version => ?MQTT_PROTO_V5})
),
?assertEqual(
Packet, parse_serialize(Packet, #{strict_mode => true, version => ?MQTT_PROTO_V5})
).
t_parse_frame_proxy_protocol(_) -> t_parse_frame_proxy_protocol(_) ->
BinList = [ <<"PROXY TCP4 ">>, <<"PROXY TCP6 ">>, <<"PROXY UNKNOWN">> BinList = [ <<"PROXY TCP4 ">>, <<"PROXY TCP6 ">>, <<"PROXY UNKNOWN">>