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

Merge remote-tracking branch 'upstream/master' into release-58

This commit is contained in:
Ivan Dyachkov 2024-08-05 10:59:59 +02:00
parent 4250d01363 86853ac6ef
commit 4865999606
292 changed files with 8893 additions and 1649 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
.ci/docker-compose-file/openldap/README.md Normal file
View File

@ -0,0 +1,61 @@
# LDAP authentication
To run manual tests with the default docker-compose files.
Expose openldap container port by uncommenting the `ports` config in `docker-compose-ldap.yaml `
To start openldap:
```
docker-compose -f ./.ci/docker-compose-file/docker-compose.yaml -f ./.ci/docker-compose-file/docker-compose-ldap.yaml up -docker
```
## LDAP database
LDAP database is populated from below files:
```
apps/emqx_ldap/test/data/emqx.io.ldif /usr/local/etc/openldap/schema/emqx.io.ldif
apps/emqx_ldap/test/data/emqx.schema /usr/local/etc/openldap/schema/emqx.schema
```
## Minimal EMQX config
```
authentication = [
{
backend = ldap
base_dn = "uid=${username},ou=testdevice,dc=emqx,dc=io"
filter = "(& (objectClass=mqttUser) (uid=${username}))"
mechanism = password_based
method {
is_superuser_attribute = isSuperuser
password_attribute = userPassword
type = hash
}
password = public
pool_size = 8
query_timeout = "5s"
request_timeout = "10s"
server = "localhost:1389"
username = "cn=root,dc=emqx,dc=io"
}
]
```
## Example ldapsearch command
```
ldapsearch -x -H ldap://localhost:389 -D "cn=root,dc=emqx,dc=io" -W -b "uid=mqttuser0007,ou=testdevice,dc=emqx,dc=io" "(&(objectClass=mqttUser)(uid=mqttuser0007))"
```
## Example mqttx command
The client password hashes are generated from their username.
```
# disabled user
mqttx pub -t 't/1' -h localhost -p 1883 -m x -u mqttuser0006 -P mqttuser0006
# enabled super-user
mqttx pub -t 't/1' -h localhost -p 1883 -m x -u mqttuser0007 -P mqttuser0007
```

9
.github/workflows/run_docker_tests.yaml vendored
View File

@ -69,7 +69,6 @@ jobs:
shell: bash
env:
EMQX_NAME: ${{ matrix.profile }}
_EMQX_TEST_DB_BACKEND: ${{ matrix.cluster_db_backend }}
strategy:
fail-fast: false
@ -78,15 +77,17 @@ jobs:
- emqx
- emqx-enterprise
- emqx-elixir
cluster_db_backend:
- mnesia
- rlog
steps:
- uses: actions/checkout@692973e3d937129bcbf40652eb9f2f61becf3332 # v4.1.7
- name: Set up environment
id: env
run: |
source env.sh
if [ "$EMQX_NAME" = "emqx-enterprise" ]; then
_EMQX_TEST_DB_BACKEND='rlog'
else
_EMQX_TEST_DB_BACKEND='mnesia'
fi
PKG_VSN=$(docker run --rm -v $(pwd):$(pwd) -w $(pwd) -u $(id -u) "$EMQX_BUILDER" ./pkg-vsn.sh "$EMQX_NAME")
echo "PKG_VSN=$PKG_VSN" >> "$GITHUB_ENV"
- uses: actions/download-artifact@65a9edc5881444af0b9093a5e628f2fe47ea3b2e # v4.1.7

13
apps/emqx/include/emqx.hrl
View File

@ -65,9 +65,20 @@
%% Route
%%--------------------------------------------------------------------
-record(share_dest, {
session_id :: emqx_session:session_id(),
group :: emqx_types:group()
}).
-record(route, {
topic :: binary(),
dest :: node() | {binary(), node()} | emqx_session:session_id() | emqx_external_broker:dest()
dest ::
node()
| {binary(), node()}
| emqx_session:session_id()
%% One session can also have multiple subscriptions to the same topic through different groups
| #share_dest{}
| emqx_external_broker:dest()
}).
%%--------------------------------------------------------------------

8
apps/emqx/include/logger.hrl
View File

@ -41,16 +41,20 @@
).
%% NOTE: do not forget to use atom for msg and add every used msg to
%% the default value of `log.thorttling.msgs` list.
%% the default value of `log.throttling.msgs` list.
-define(SLOG_THROTTLE(Level, Data),
?SLOG_THROTTLE(Level, Data, #{})
).
-define(SLOG_THROTTLE(Level, Data, Meta),
?SLOG_THROTTLE(Level, undefined, Data, Meta)
).
-define(SLOG_THROTTLE(Level, UniqueKey, Data, Meta),
case logger:allow(Level, ?MODULE) of
true ->
(fun(#{msg := __Msg} = __Data) ->
case emqx_log_throttler:allow(__Msg) of
case emqx_log_throttler:allow(__Msg, UniqueKey) of
true ->
logger:log(Level, __Data, Meta);
false ->

2
apps/emqx/priv/bpapi.versions
View File

@ -10,6 +10,7 @@
{emqx_bridge,5}.
{emqx_bridge,6}.
{emqx_broker,1}.
{emqx_cluster_link,1}.
{emqx_cm,1}.
{emqx_cm,2}.
{emqx_cm,3}.
@ -26,6 +27,7 @@
{emqx_ds,2}.
{emqx_ds,3}.
{emqx_ds,4}.
{emqx_ds_shared_sub,1}.
{emqx_eviction_agent,1}.
{emqx_eviction_agent,2}.
{emqx_eviction_agent,3}.

5
apps/emqx/rebar.config
View File

@ -31,12 +31,11 @@
{esockd, {git, "https://github.com/emqx/esockd", {tag, "5.11.3"}}},
{ekka, {git, "https://github.com/emqx/ekka", {tag, "0.19.5"}}},
{gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "3.3.1"}}},
{hocon, {git, "https://github.com/emqx/hocon.git", {tag, "0.43.1"}}},
{hocon, {git, "https://github.com/emqx/hocon.git", {tag, "0.43.2"}}},
{emqx_http_lib, {git, "https://github.com/emqx/emqx_http_lib.git", {tag, "0.5.3"}}},
{pbkdf2, {git, "https://github.com/emqx/erlang-pbkdf2.git", {tag, "2.0.4"}}},
{recon, {git, "https://github.com/ferd/recon", {tag, "2.5.1"}}},
{snabbkaffe, {git, "https://github.com/kafka4beam/snabbkaffe.git", {tag, "1.0.10"}}},
{ra, "2.7.3"}
{snabbkaffe, {git, "https://github.com/kafka4beam/snabbkaffe.git", {tag, "1.0.10"}}}
]}.
{plugins, [{rebar3_proper, "0.12.1"}, rebar3_path_deps]}.

7
apps/emqx/src/emqx_exclusive_subscription.erl
View File

@ -117,6 +117,13 @@ try_subscribe(ClientId, Topic) ->
write
),
allow;
[#exclusive_subscription{clientid = ClientId, topic = Topic}] ->
%% Fixed the issue-13476
%% In this feature, the user must manually call `unsubscribe` to release the lock,
%% but sometimes the node may go down for some reason,
%% then the client will reconnect to this node and resubscribe.
%% We need to allow resubscription, otherwise the lock will never be released.
allow;
[_] ->
deny
end.

10
apps/emqx/src/emqx_external_broker.erl
View File

@ -43,7 +43,9 @@
add_shared_route/2,
delete_shared_route/2,
add_persistent_route/2,
delete_persistent_route/2
delete_persistent_route/2,
add_persistent_shared_route/3,
delete_persistent_shared_route/3
]).
-export_type([dest/0]).
@ -129,6 +131,12 @@ add_persistent_route(Topic, ID) ->
delete_persistent_route(Topic, ID) ->
?safe_with_provider(?FUNCTION_NAME(Topic, ID), ok).
add_persistent_shared_route(Topic, Group, ID) ->
?safe_with_provider(?FUNCTION_NAME(Topic, Group, ID), ok).
delete_persistent_shared_route(Topic, Group, ID) ->
?safe_with_provider(?FUNCTION_NAME(Topic, Group, ID), ok).
%%--------------------------------------------------------------------
%% Internal functions
%%--------------------------------------------------------------------

72
apps/emqx/src/emqx_log_throttler.erl
View File

@ -25,7 +25,7 @@
-export([start_link/0]).
%% throttler API
-export([allow/1]).
-export([allow/2]).
%% gen_server callbacks
-export([
@ -40,23 +40,29 @@
-define(SEQ_ID(Msg), {?MODULE, Msg}).
-define(NEW_SEQ, atomics:new(1, [{signed, false}])).
-define(GET_SEQ(Msg), persistent_term:get(?SEQ_ID(Msg), undefined)).
-define(ERASE_SEQ(Msg), persistent_term:erase(?SEQ_ID(Msg))).
-define(RESET_SEQ(SeqRef), atomics:put(SeqRef, 1, 0)).
-define(INC_SEQ(SeqRef), atomics:add(SeqRef, 1, 1)).
-define(GET_DROPPED(SeqRef), atomics:get(SeqRef, 1) - 1).
-define(IS_ALLOWED(SeqRef), atomics:add_get(SeqRef, 1, 1) =:= 1).
-define(NEW_THROTTLE(Msg, SeqRef), persistent_term:put(?SEQ_ID(Msg), SeqRef)).
-define(MSGS_LIST, emqx:get_config([log, throttling, msgs], [])).
-define(TIME_WINDOW_MS, timer:seconds(emqx:get_config([log, throttling, time_window], 60))).
-spec allow(atom()) -> boolean().
allow(Msg) when is_atom(Msg) ->
%% @doc Check if a throttled log message is allowed to pass down to the logger this time.
%% The Msg has to be an atom, and the second argument `UniqueKey' should be `undefined'
%% for predefined message IDs.
%% For relatively static resources created from configurations such as data integration
%% resource IDs `UniqueKey' should be of `binary()' type.
-spec allow(atom(), undefined | binary()) -> boolean().
allow(Msg, UniqueKey) when
is_atom(Msg) andalso (is_binary(UniqueKey) orelse UniqueKey =:= undefined)
->
case emqx_logger:get_primary_log_level() of
debug ->
true;
_ ->
do_allow(Msg)
do_allow(Msg, UniqueKey)
end.
-spec start_link() -> startlink_ret().
@ -68,7 +74,8 @@ start_link() ->
%%--------------------------------------------------------------------
init([]) ->
ok = lists:foreach(fun(Msg) -> ?NEW_THROTTLE(Msg, ?NEW_SEQ) end, ?MSGS_LIST),
process_flag(trap_exit, true),
ok = lists:foreach(fun new_throttler/1, ?MSGS_LIST),
CurrentPeriodMs = ?TIME_WINDOW_MS,
TimerRef = schedule_refresh(CurrentPeriodMs),
{ok, #{timer_ref => TimerRef, current_period_ms => CurrentPeriodMs}}.
@ -86,16 +93,22 @@ handle_info(refresh, #{current_period_ms := PeriodMs} = State) ->
DroppedStats = lists:foldl(
fun(Msg, Acc) ->
case ?GET_SEQ(Msg) of
%% Should not happen, unless the static ids list is updated at run-time.
undefined ->
?NEW_THROTTLE(Msg, ?NEW_SEQ),
%% Should not happen, unless the static ids list is updated at run-time.
new_throttler(Msg),
?tp(log_throttler_new_msg, #{throttled_msg => Msg}),
Acc;
SeqMap when is_map(SeqMap) ->
maps:fold(
fun(Key, Ref, Acc0) ->
ID = iolist_to_binary([atom_to_binary(Msg), $:, Key]),
drop_stats(Ref, ID, Acc0)
end,
Acc,
SeqMap
);
SeqRef ->
Dropped = ?GET_DROPPED(SeqRef),
ok = ?RESET_SEQ(SeqRef),
?tp(log_throttler_dropped, #{dropped_count => Dropped, throttled_msg => Msg}),
maybe_add_dropped(Msg, Dropped, Acc)
drop_stats(SeqRef, Msg, Acc)
end
end,
#{},
@ -112,7 +125,16 @@ handle_info(Info, State) ->
?SLOG(error, #{msg => "unxpected_info", info => Info}),
{noreply, State}.
drop_stats(SeqRef, Msg, Acc) ->
Dropped = ?GET_DROPPED(SeqRef),
ok = ?RESET_SEQ(SeqRef),
?tp(log_throttler_dropped, #{dropped_count => Dropped, throttled_msg => Msg}),
maybe_add_dropped(Msg, Dropped, Acc).
terminate(_Reason, _State) ->
%% atomics do not have delete/remove/release/deallocate API
%% after the reference is garbage-collected the resource is released
lists:foreach(fun(Msg) -> ?ERASE_SEQ(Msg) end, ?MSGS_LIST),
ok.
code_change(_OldVsn, State, _Extra) ->
@ -122,17 +144,27 @@ code_change(_OldVsn, State, _Extra) ->
%% internal functions
%%--------------------------------------------------------------------
do_allow(Msg) ->
do_allow(Msg, UniqueKey) ->
case persistent_term:get(?SEQ_ID(Msg), undefined) of
undefined ->
%% This is either a race condition (emqx_log_throttler is not started yet)
%% or a developer mistake (msg used in ?SLOG_THROTTLE/2,3 macro is
%% not added to the default value of `log.throttling.msgs`.
?SLOG(info, #{
msg => "missing_log_throttle_sequence",
?SLOG(debug, #{
msg => "log_throttle_disabled",
throttled_msg => Msg
}),
true;
%% e.g: unrecoverable msg throttle according resource_id
SeqMap when is_map(SeqMap) ->
case maps:find(UniqueKey, SeqMap) of
{ok, SeqRef} ->
?IS_ALLOWED(SeqRef);
error ->
SeqRef = ?NEW_SEQ,
new_throttler(Msg, SeqMap#{UniqueKey => SeqRef}),
true
end;
SeqRef ->
?IS_ALLOWED(SeqRef)
end.
@ -154,3 +186,11 @@ maybe_log_dropped(_DroppedStats, _PeriodMs) ->
schedule_refresh(PeriodMs) ->
?tp(log_throttler_sched_refresh, #{new_period_ms => PeriodMs}),
erlang:send_after(PeriodMs, ?MODULE, refresh).
new_throttler(unrecoverable_resource_error = Msg) ->
new_throttler(Msg, #{});
new_throttler(Msg) ->
new_throttler(Msg, ?NEW_SEQ).
new_throttler(Msg, AtomicOrEmptyMap) ->
persistent_term:put(?SEQ_ID(Msg), AtomicOrEmptyMap).

39
apps/emqx/src/emqx_persistent_session_ds.erl
View File

@ -621,9 +621,13 @@ handle_timeout(ClientInfo, ?TIMER_RETRY_REPLAY, Session0) ->
Session = replay_streams(Session0, ClientInfo),
{ok, [], Session};
handle_timeout(ClientInfo, ?TIMER_GET_STREAMS, Session0 = #{s := S0, shared_sub_s := SharedSubS0}) ->
S1 = emqx_persistent_session_ds_subs:gc(S0),
S2 = emqx_persistent_session_ds_stream_scheduler:renew_streams(S1),
{S, SharedSubS} = emqx_persistent_session_ds_shared_subs:renew_streams(S2, SharedSubS0),
%% `gc` and `renew_streams` methods may drop unsubscribed streams.
%% Shared subscription handler must have a chance to see unsubscribed streams
%% in the fully replayed state.
{S1, SharedSubS1} = emqx_persistent_session_ds_shared_subs:pre_renew_streams(S0, SharedSubS0),
S2 = emqx_persistent_session_ds_subs:gc(S1),
S3 = emqx_persistent_session_ds_stream_scheduler:renew_streams(S2),
{S, SharedSubS} = emqx_persistent_session_ds_shared_subs:renew_streams(S3, SharedSubS1),
Interval = get_config(ClientInfo, [renew_streams_interval]),
Session = emqx_session:ensure_timer(
?TIMER_GET_STREAMS,
@ -757,7 +761,7 @@ skip_batch(StreamKey, SRS0, Session = #{s := S0}, ClientInfo, Reason) ->
%%--------------------------------------------------------------------
-spec disconnect(session(), emqx_types:conninfo()) -> {shutdown, session()}.
disconnect(Session = #{id := Id, s := S0}, ConnInfo) ->
disconnect(Session = #{id := Id, s := S0, shared_sub_s := SharedSubS0}, ConnInfo) ->
S1 = maybe_set_offline_info(S0, Id),
S2 = emqx_persistent_session_ds_state:set_last_alive_at(now_ms(), S1),
S3 =
@ -767,8 +771,9 @@ disconnect(Session = #{id := Id, s := S0}, ConnInfo) ->
_ ->
S2
end,
S = emqx_persistent_session_ds_state:commit(S3),
{shutdown, Session#{s => S}}.
{S4, SharedSubS} = emqx_persistent_session_ds_shared_subs:on_disconnect(S3, SharedSubS0),
S = emqx_persistent_session_ds_state:commit(S4),
{shutdown, Session#{s => S, shared_sub_s => SharedSubS}}.
-spec terminate(Reason :: term(), session()) -> ok.
terminate(_Reason, Session = #{id := Id, s := S}) ->
@ -816,10 +821,12 @@ list_client_subscriptions(ClientId) ->
{error, not_found}
end.
-spec get_client_subscription(emqx_types:clientid(), emqx_types:topic()) ->
-spec get_client_subscription(emqx_types:clientid(), topic_filter() | share_topic_filter()) ->
subscription() | undefined.
get_client_subscription(ClientId, Topic) ->
emqx_persistent_session_ds_subs:cold_get_subscription(ClientId, Topic).
get_client_subscription(ClientId, #share{} = ShareTopicFilter) ->
emqx_persistent_session_ds_shared_subs:cold_get_subscription(ClientId, ShareTopicFilter);
get_client_subscription(ClientId, TopicFilter) ->
emqx_persistent_session_ds_subs:cold_get_subscription(ClientId, TopicFilter).
%%--------------------------------------------------------------------
%% Session tables operations
@ -986,14 +993,14 @@ do_ensure_all_iterators_closed(_DSSessionID) ->
%% Normal replay:
%%--------------------------------------------------------------------
fetch_new_messages(Session0 = #{s := S0}, ClientInfo) ->
LFS = maps:get(last_fetched_stream, Session0, beginning),
ItStream = emqx_persistent_session_ds_stream_scheduler:iter_next_streams(LFS, S0),
fetch_new_messages(Session0 = #{s := S0, shared_sub_s := SharedSubS0}, ClientInfo) ->
{S1, SharedSubS1} = emqx_persistent_session_ds_shared_subs:on_streams_replay(S0, SharedSubS0),
Session1 = Session0#{s => S1, shared_sub_s => SharedSubS1},
LFS = maps:get(last_fetched_stream, Session1, beginning),
ItStream = emqx_persistent_session_ds_stream_scheduler:iter_next_streams(LFS, S1),
BatchSize = get_config(ClientInfo, [batch_size]),
Session1 = fetch_new_messages(ItStream, BatchSize, Session0, ClientInfo),
#{s := S1, shared_sub_s := SharedSubS0} = Session1,
{S2, SharedSubS1} = emqx_persistent_session_ds_shared_subs:on_streams_replayed(S1, SharedSubS0),
Session1#{s => S2, shared_sub_s => SharedSubS1}.
Session2 = fetch_new_messages(ItStream, BatchSize, Session1, ClientInfo),
Session2#{shared_sub_s => SharedSubS1}.
fetch_new_messages(ItStream0, BatchSize, Session0, ClientInfo) ->
#{inflight := Inflight} = Session0,

8
apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_router.erl
View File

@ -17,7 +17,7 @@
-module(emqx_persistent_session_ds_router).
-include("emqx.hrl").
-include("emqx_persistent_session_ds/emqx_ps_ds_int.hrl").
-include("emqx_ps_ds_int.hrl").
-export([init_tables/0]).
@ -47,7 +47,7 @@
-endif.
-type route() :: #ps_route{}.
-type dest() :: emqx_persistent_session_ds:id().
-type dest() :: emqx_persistent_session_ds:id() | #share_dest{}.
-export_type([dest/0, route/0]).
@ -161,7 +161,7 @@ topics() ->
print_routes(Topic) ->
lists:foreach(
fun(#ps_route{topic = To, dest = Dest}) ->
io:format("~ts -> ~ts~n", [To, Dest])
io:format("~ts -> ~tp~n", [To, Dest])
end,
match_routes(Topic)
).
@ -247,6 +247,8 @@ mk_filtertab_fold_fun(FoldFun) ->
match_filters(Topic) ->
emqx_topic_index:matches(Topic, ?PS_FILTERS_TAB, []).
get_dest_session_id(#share_dest{session_id = DSSessionId}) ->
DSSessionId;
get_dest_session_id({_, DSSessionId}) ->
DSSessionId;
get_dest_session_id(DSSessionId) ->

792
apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_shared_subs.erl
View File

@ -2,11 +2,37 @@
%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
%% @doc This module
%% * handles creation and management of _shared_ subscriptions for the session;
%% * provides streams to the session;
%% * handles progress of stream replay.
%%
%% The logic is quite straightforward; most of the parts resemble the logic of the
%% `emqx_persistent_session_ds_subs` (subscribe/unsubscribe) and
%% `emqx_persistent_session_ds_scheduler` (providing new streams),
%% but some data is sent or received from the `emqx_persistent_session_ds_shared_subs_agent`
%% which communicates with remote shared subscription leaders.
%%
%% A tricky part is the concept of "scheduled actions". When we unsubscribe from a topic
%% we may have some streams that have unacked messages. So we do not have a reliable
%% progress for them. Sending the current progress to the leader and disconnecting
%% will lead to the duplication of messages. So after unsubscription, we need to wait
%% some time until all streams are acked, and only then we disconnect from the leader.
%%
%% For this purpose we have the `scheduled_actions` map in the state of the module.
%% We preserve there the streams that we need to wait for and collect their progress.
%% We also use `scheduled_actions` for resubscriptions. If a client quickly resubscribes
%% after unsubscription, we may still have the mentioned streams unacked. If we abandon
%% them, just connect to the leader, then it may lease us the same streams again, but with
%% the previous progress. So messages may duplicate.
-module(emqx_persistent_session_ds_shared_subs).
-include("emqx_mqtt.hrl").
-include("emqx.hrl").
-include("logger.hrl").
-include("session_internals.hrl").
-include_lib("snabbkaffe/include/trace.hrl").
-export([
@ -15,16 +41,51 @@
on_subscribe/3,
on_unsubscribe/4,
on_disconnect/2,
on_streams_replayed/2,
on_streams_replay/2,
on_info/3,
pre_renew_streams/2,
renew_streams/2,
to_map/2
]).
%% Management API:
-export([
cold_get_subscription/2
]).
-define(schedule_subscribe, schedule_subscribe).
-define(schedule_unsubscribe, schedule_unsubscribe).
-type stream_key() :: {emqx_persistent_session_ds:id(), emqx_ds:stream()}.
-type scheduled_action_type() ::
{?schedule_subscribe, emqx_types:subopts()} | ?schedule_unsubscribe.
-type agent_stream_progress() :: #{
stream := emqx_ds:stream(),
progress := progress(),
use_finished := boolean()
}.
-type progress() ::
#{
iterator := emqx_ds:iterator()
}.
-type scheduled_action() :: #{
type := scheduled_action_type(),
stream_keys_to_wait := [stream_key()],
progresses := [agent_stream_progress()]
}.
-type t() :: #{
agent := emqx_persistent_session_ds_shared_subs_agent:t()
agent := emqx_persistent_session_ds_shared_subs_agent:t(),
scheduled_actions := #{
share_topic_filter() => scheduled_action()
}
}.
-type share_topic_filter() :: emqx_persistent_session_ds:share_topic_filter().
-type opts() :: #{
@ -34,184 +95,90 @@
-define(rank_x, rank_shared).
-define(rank_y, 0).
-export_type([
progress/0,
agent_stream_progress/0
]).
%%--------------------------------------------------------------------
%% API
%%--------------------------------------------------------------------
%%--------------------------------------------------------------------
%% new
-spec new(opts()) -> t().
new(Opts) ->
#{
agent => emqx_persistent_session_ds_shared_subs_agent:new(
agent_opts(Opts)
)
),
scheduled_actions => #{}
}.
%%--------------------------------------------------------------------
%% open
-spec open(emqx_persistent_session_ds_state:t(), opts()) ->
{ok, emqx_persistent_session_ds_state:t(), t()}.
open(S, Opts) ->
open(S0, Opts) ->
SharedSubscriptions = fold_shared_subs(
fun(#share{} = TopicFilter, Sub, Acc) ->
[{TopicFilter, to_agent_subscription(S, Sub)} | Acc]
fun(#share{} = ShareTopicFilter, Sub, Acc) ->
[{ShareTopicFilter, to_agent_subscription(S0, Sub)} | Acc]
end,
[],
S
S0
),
Agent = emqx_persistent_session_ds_shared_subs_agent:open(
SharedSubscriptions, agent_opts(Opts)
),
SharedSubS = #{agent => Agent},
{ok, S, SharedSubS}.
SharedSubS = #{agent => Agent, scheduled_actions => #{}},
S1 = revoke_all_streams(S0),
{ok, S1, SharedSubS}.
%%--------------------------------------------------------------------
%% on_subscribe
-spec on_subscribe(
share_topic_filter(),
emqx_types:subopts(),
emqx_persistent_session_ds:session()
) -> {ok, emqx_persistent_session_ds_state:t(), t()} | {error, emqx_types:reason_code()}.
on_subscribe(TopicFilter, SubOpts, #{s := S} = Session) ->
Subscription = emqx_persistent_session_ds_state:get_subscription(TopicFilter, S),
on_subscribe(Subscription, TopicFilter, SubOpts, Session).
-spec on_unsubscribe(
emqx_persistent_session_ds:id(),
emqx_persistent_session_ds:topic_filter(),
emqx_persistent_session_ds_state:t(),
t()
) ->
{ok, emqx_persistent_session_ds_state:t(), t(), emqx_persistent_session_ds:subscription()}
| {error, emqx_types:reason_code()}.
on_unsubscribe(SessionId, TopicFilter, S0, #{agent := Agent0} = SharedSubS0) ->
case lookup(TopicFilter, S0) of
undefined ->
{error, ?RC_NO_SUBSCRIPTION_EXISTED};
Subscription ->
?tp(persistent_session_ds_subscription_delete, #{
session_id => SessionId, topic_filter => TopicFilter
}),
Agent1 = emqx_persistent_session_ds_shared_subs_agent:on_unsubscribe(
Agent0, TopicFilter
),
SharedSubS = SharedSubS0#{agent => Agent1},
S = emqx_persistent_session_ds_state:del_subscription(TopicFilter, S0),
{ok, S, SharedSubS, Subscription}
end.
-spec renew_streams(emqx_persistent_session_ds_state:t(), t()) ->
{emqx_persistent_session_ds_state:t(), t()}.
renew_streams(S0, #{agent := Agent0} = SharedSubS0) ->
{StreamLeaseEvents, Agent1} = emqx_persistent_session_ds_shared_subs_agent:renew_streams(
Agent0
),
?tp(info, shared_subs_new_stream_lease_events, #{stream_lease_events => StreamLeaseEvents}),
S1 = lists:foldl(
fun
(#{type := lease} = Event, S) -> accept_stream(Event, S);
(#{type := revoke} = Event, S) -> revoke_stream(Event, S)
end,
S0,
StreamLeaseEvents
),
SharedSubS1 = SharedSubS0#{agent => Agent1},
{S1, SharedSubS1}.
-spec on_streams_replayed(
emqx_persistent_session_ds_state:t(),
t()
) -> {emqx_persistent_session_ds_state:t(), t()}.
on_streams_replayed(S, #{agent := Agent0} = SharedSubS0) ->
%% TODO
%% Is it sufficient for a report?
Progress = fold_shared_stream_states(
fun(TopicFilter, Stream, SRS, Acc) ->
#srs{it_begin = BeginIt} = SRS,
StreamProgress = #{
topic_filter => TopicFilter,
stream => Stream,
iterator => BeginIt
},
[StreamProgress | Acc]
end,
[],
S
),
Agent1 = emqx_persistent_session_ds_shared_subs_agent:on_stream_progress(
Agent0, Progress
),
SharedSubS1 = SharedSubS0#{agent => Agent1},
{S, SharedSubS1}.
-spec on_info(emqx_persistent_session_ds_state:t(), t(), term()) ->
{emqx_persistent_session_ds_state:t(), t()}.
on_info(S, #{agent := Agent0} = SharedSubS0, Info) ->
Agent1 = emqx_persistent_session_ds_shared_subs_agent:on_info(Agent0, Info),
SharedSubS1 = SharedSubS0#{agent => Agent1},
{S, SharedSubS1}.
-spec to_map(emqx_persistent_session_ds_state:t(), t()) -> map().
to_map(_S, _SharedSubS) ->
%% TODO
#{}.
on_subscribe(#share{} = ShareTopicFilter, SubOpts, #{s := S} = Session) ->
Subscription = emqx_persistent_session_ds_state:get_subscription(ShareTopicFilter, S),
on_subscribe(Subscription, ShareTopicFilter, SubOpts, Session).
%%--------------------------------------------------------------------
%% Internal functions
%%--------------------------------------------------------------------
%% on_subscribe internal functions
fold_shared_subs(Fun, Acc, S) ->
emqx_persistent_session_ds_state:fold_subscriptions(
fun
(#share{} = TopicFilter, Sub, Acc0) -> Fun(TopicFilter, Sub, Acc0);
(_, _Sub, Acc0) -> Acc0
end,
Acc,
S
).
fold_shared_stream_states(Fun, Acc, S) ->
%% TODO
%% Optimize or cache
TopicFilters = fold_shared_subs(
fun
(#share{} = TopicFilter, #{id := Id} = _Sub, Acc0) ->
Acc0#{Id => TopicFilter};
(_, _, Acc0) ->
Acc0
end,
#{},
S
),
emqx_persistent_session_ds_state:fold_streams(
fun({SubId, Stream}, SRS, Acc0) ->
case TopicFilters of
#{SubId := TopicFilter} ->
Fun(TopicFilter, Stream, SRS, Acc0);
_ ->
Acc0
end
end,
Acc,
S
).
on_subscribe(undefined, TopicFilter, SubOpts, #{props := Props, s := S} = Session) ->
on_subscribe(undefined, ShareTopicFilter, SubOpts, #{props := Props, s := S} = Session) ->
#{max_subscriptions := MaxSubscriptions} = Props,
case emqx_persistent_session_ds_state:n_subscriptions(S) < MaxSubscriptions of
true ->
create_new_subscription(TopicFilter, SubOpts, Session);
create_new_subscription(ShareTopicFilter, SubOpts, Session);
false ->
{error, ?RC_QUOTA_EXCEEDED}
end;
on_subscribe(Subscription, TopicFilter, SubOpts, Session) ->
update_subscription(Subscription, TopicFilter, SubOpts, Session).
on_subscribe(Subscription, ShareTopicFilter, SubOpts, Session) ->
update_subscription(Subscription, ShareTopicFilter, SubOpts, Session).
-dialyzer({nowarn_function, create_new_subscription/3}).
create_new_subscription(TopicFilter, SubOpts, #{
id := SessionId, s := S0, shared_sub_s := #{agent := Agent0} = SharedSubS0, props := Props
create_new_subscription(#share{topic = TopicFilter, group = Group} = ShareTopicFilter, SubOpts, #{
id := SessionId,
s := S0,
shared_sub_s := #{agent := Agent} = SharedSubS0,
props := Props
}) ->
case
emqx_persistent_session_ds_shared_subs_agent:on_subscribe(
Agent0, TopicFilter, SubOpts
emqx_persistent_session_ds_shared_subs_agent:can_subscribe(
Agent, ShareTopicFilter, SubOpts
)
of
{ok, Agent1} ->
ok ->
ok = emqx_persistent_session_ds_router:do_add_route(TopicFilter, #share_dest{
session_id = SessionId, group = Group
}),
_ = emqx_external_broker:add_persistent_shared_route(TopicFilter, Group, SessionId),
#{upgrade_qos := UpgradeQoS} = Props,
{SubId, S1} = emqx_persistent_session_ds_state:new_id(S0),
{SStateId, S2} = emqx_persistent_session_ds_state:new_id(S1),
@ -227,20 +194,20 @@ create_new_subscription(TopicFilter, SubOpts, #{
start_time => now_ms()
},
S = emqx_persistent_session_ds_state:put_subscription(
TopicFilter, Subscription, S3
ShareTopicFilter, Subscription, S3
),
SharedSubS = SharedSubS0#{agent => Agent1},
?tp(persistent_session_ds_shared_subscription_added, #{
topic_filter => TopicFilter, session => SessionId
}),
SharedSubS = schedule_subscribe(SharedSubS0, ShareTopicFilter, SubOpts),
{ok, S, SharedSubS};
{error, _} = Error ->
Error
end.
update_subscription(#{current_state := SStateId0, id := SubId} = Sub0, TopicFilter, SubOpts, #{
update_subscription(
#{current_state := SStateId0, id := SubId} = Sub0, ShareTopicFilter, SubOpts, #{
s := S0, shared_sub_s := SharedSubS, props := Props
}) ->
}
) ->
#{upgrade_qos := UpgradeQoS} = Props,
SState = #{parent_subscription => SubId, upgrade_qos => UpgradeQoS, subopts => SubOpts},
case emqx_persistent_session_ds_state:get_subscription_state(SStateId0, S0) of
@ -254,36 +221,173 @@ update_subscription(#{current_state := SStateId0, id := SubId} = Sub0, TopicFilt
SStateId, SState, S1
),
Sub = Sub0#{current_state => SStateId},
S = emqx_persistent_session_ds_state:put_subscription(TopicFilter, Sub, S2),
S = emqx_persistent_session_ds_state:put_subscription(ShareTopicFilter, Sub, S2),
{ok, S, SharedSubS}
end.
lookup(TopicFilter, S) ->
case emqx_persistent_session_ds_state:get_subscription(TopicFilter, S) of
Sub = #{current_state := SStateId} ->
case emqx_persistent_session_ds_state:get_subscription_state(SStateId, S) of
#{subopts := SubOpts} ->
Sub#{subopts => SubOpts};
-dialyzer({nowarn_function, schedule_subscribe/3}).
schedule_subscribe(
#{agent := Agent0, scheduled_actions := ScheduledActions0} = SharedSubS0,
ShareTopicFilter,
SubOpts
) ->
case ScheduledActions0 of
#{ShareTopicFilter := ScheduledAction} ->
ScheduledActions1 = ScheduledActions0#{
ShareTopicFilter => ScheduledAction#{type => {?schedule_subscribe, SubOpts}}
},
?tp(warning, shared_subs_schedule_subscribe_override, #{
share_topic_filter => ShareTopicFilter,
new_type => {?schedule_subscribe, SubOpts},
old_action => format_schedule_action(ScheduledAction)
}),
SharedSubS0#{scheduled_actions := ScheduledActions1};
_ ->
?tp(warning, shared_subs_schedule_subscribe_new, #{
share_topic_filter => ShareTopicFilter, subopts => SubOpts
}),
Agent1 = emqx_persistent_session_ds_shared_subs_agent:on_subscribe(
Agent0, ShareTopicFilter, SubOpts
),
SharedSubS0#{agent => Agent1}
end.
%%--------------------------------------------------------------------
%% on_unsubscribe
-spec on_unsubscribe(
emqx_persistent_session_ds:id(),
share_topic_filter(),
emqx_persistent_session_ds_state:t(),
t()
) ->
{ok, emqx_persistent_session_ds_state:t(), t(), emqx_persistent_session_ds:subscription()}
| {error, emqx_types:reason_code()}.
on_unsubscribe(
SessionId, #share{topic = TopicFilter, group = Group} = ShareTopicFilter, S0, SharedSubS0
) ->
case lookup(ShareTopicFilter, S0) of
undefined ->
undefined
end;
undefined ->
undefined
{error, ?RC_NO_SUBSCRIPTION_EXISTED};
#{id := SubId} = Subscription ->
?tp(persistent_session_ds_subscription_delete, #{
session_id => SessionId, share_topic_filter => ShareTopicFilter
}),
_ = emqx_external_broker:delete_persistent_shared_route(TopicFilter, Group, SessionId),
ok = emqx_persistent_session_ds_router:do_delete_route(TopicFilter, #share_dest{
session_id = SessionId, group = Group
}),
S = emqx_persistent_session_ds_state:del_subscription(ShareTopicFilter, S0),
SharedSubS = schedule_unsubscribe(S, SharedSubS0, SubId, ShareTopicFilter),
{ok, S, SharedSubS, Subscription}
end.
%%--------------------------------------------------------------------
%% on_unsubscribe internal functions
schedule_unsubscribe(
S, #{scheduled_actions := ScheduledActions0} = SharedSubS0, UnsubscridedSubId, ShareTopicFilter
) ->
case ScheduledActions0 of
#{ShareTopicFilter := ScheduledAction0} ->
ScheduledAction1 = ScheduledAction0#{type => ?schedule_unsubscribe},
ScheduledActions1 = ScheduledActions0#{
ShareTopicFilter => ScheduledAction1
},
?tp(warning, shared_subs_schedule_unsubscribe_override, #{
share_topic_filter => ShareTopicFilter,
new_type => ?schedule_unsubscribe,
old_action => format_schedule_action(ScheduledAction0)
}),
SharedSubS0#{scheduled_actions := ScheduledActions1};
_ ->
StreamKeys = stream_keys_by_sub_id(S, UnsubscridedSubId),
ScheduledActions1 = ScheduledActions0#{
ShareTopicFilter => #{
type => ?schedule_unsubscribe,
stream_keys_to_wait => StreamKeys,
progresses => []
}
},
?tp(warning, shared_subs_schedule_unsubscribe_new, #{
share_topic_filter => ShareTopicFilter,
stream_keys => format_stream_keys(StreamKeys)
}),
SharedSubS0#{scheduled_actions := ScheduledActions1}
end.
%%--------------------------------------------------------------------
%% pre_renew_streams
-spec pre_renew_streams(emqx_persistent_session_ds_state:t(), t()) ->
{emqx_persistent_session_ds_state:t(), t()}.
pre_renew_streams(S, SharedSubS) ->
on_streams_replay(S, SharedSubS).
%%--------------------------------------------------------------------
%% renew_streams
-spec renew_streams(emqx_persistent_session_ds_state:t(), t()) ->
{emqx_persistent_session_ds_state:t(), t()}.
renew_streams(S0, #{agent := Agent0, scheduled_actions := ScheduledActions} = SharedSubS0) ->
{StreamLeaseEvents, Agent1} = emqx_persistent_session_ds_shared_subs_agent:renew_streams(
Agent0
),
StreamLeaseEvents =/= [] andalso
?tp(warning, shared_subs_new_stream_lease_events, #{
stream_lease_events => format_lease_events(StreamLeaseEvents)
}),
S1 = lists:foldl(
fun
(#{type := lease} = Event, S) -> accept_stream(Event, S, ScheduledActions);
(#{type := revoke} = Event, S) -> revoke_stream(Event, S)
end,
S0,
StreamLeaseEvents
),
SharedSubS1 = SharedSubS0#{agent => Agent1},
{S1, SharedSubS1}.
%%--------------------------------------------------------------------
%% renew_streams internal functions
accept_stream(#{share_topic_filter := ShareTopicFilter} = Event, S, ScheduledActions) ->
%% If we have a pending action (subscribe or unsubscribe) for this topic filter,
%% we should not accept a stream and start replaying it. We won't use it anyway:
%% * if subscribe is pending, we will reset agent obtain a new lease
%% * if unsubscribe is pending, we will drop connection
case ScheduledActions of
#{ShareTopicFilter := _Action} ->
S;
_ ->
accept_stream(Event, S)
end.
accept_stream(
#{topic_filter := TopicFilter, stream := Stream, iterator := Iterator}, S0
#{
share_topic_filter := ShareTopicFilter,
stream := Stream,
progress := #{iterator := Iterator} = _Progress
} = _Event,
S0
) ->
case emqx_persistent_session_ds_state:get_subscription(TopicFilter, S0) of
case emqx_persistent_session_ds_state:get_subscription(ShareTopicFilter, S0) of
undefined ->
%% This should not happen.
%% Agent should have received unsubscribe callback
%% and should not have passed this stream as a new one
error(new_stream_without_sub);
%% We unsubscribed
S0;
#{id := SubId, current_state := SStateId} ->
Key = {SubId, Stream},
NeedCreateStream =
case emqx_persistent_session_ds_state:get_stream(Key, S0) of
undefined ->
true;
#srs{unsubscribed = true} ->
true;
_SRS ->
false
end,
case NeedCreateStream of
true ->
NewSRS =
#srs{
rank_x = ?rank_x,
@ -294,15 +398,15 @@ accept_stream(
},
S1 = emqx_persistent_session_ds_state:put_stream(Key, NewSRS, S0),
S1;
_SRS ->
false ->
S0
end
end.
revoke_stream(
#{topic_filter := TopicFilter, stream := Stream}, S0
#{share_topic_filter := ShareTopicFilter, stream := Stream}, S0
) ->
case emqx_persistent_session_ds_state:get_subscription(TopicFilter, S0) of
case emqx_persistent_session_ds_state:get_subscription(ShareTopicFilter, S0) of
undefined ->
%% This should not happen.
%% Agent should have received unsubscribe callback
@ -320,19 +424,363 @@ revoke_stream(
end
end.
-spec to_agent_subscription(
emqx_persistent_session_ds_state:t(), emqx_persistent_session_ds:subscription()
%%--------------------------------------------------------------------
%% on_streams_replay
-spec on_streams_replay(
emqx_persistent_session_ds_state:t(),
t()
) -> {emqx_persistent_session_ds_state:t(), t()}.
on_streams_replay(S0, SharedSubS0) ->
{S1, #{agent := Agent0, scheduled_actions := ScheduledActions0} = SharedSubS1} =
renew_streams(S0, SharedSubS0),
Progresses = all_stream_progresses(S1, Agent0),
Agent1 = emqx_persistent_session_ds_shared_subs_agent:on_stream_progress(
Agent0, Progresses
),
{Agent2, ScheduledActions1} = run_scheduled_actions(S1, Agent1, ScheduledActions0),
SharedSubS2 = SharedSubS1#{
agent => Agent2,
scheduled_actions => ScheduledActions1
},
{S1, SharedSubS2}.
%%--------------------------------------------------------------------
%% on_streams_replay internal functions
all_stream_progresses(S, Agent) ->
all_stream_progresses(S, Agent, _NeedUnacked = false).
all_stream_progresses(S, _Agent, NeedUnacked) ->
CommQos1 = emqx_persistent_session_ds_state:get_seqno(?committed(?QOS_1), S),
CommQos2 = emqx_persistent_session_ds_state:get_seqno(?committed(?QOS_2), S),
fold_shared_stream_states(
fun(ShareTopicFilter, Stream, SRS, ProgressesAcc0) ->
case
is_stream_started(CommQos1, CommQos2, SRS) and
(NeedUnacked or is_stream_fully_acked(CommQos1, CommQos2, SRS))
of
true ->
StreamProgress = stream_progress(CommQos1, CommQos2, Stream, SRS),
maps:update_with(
ShareTopicFilter,
fun(Progresses) -> [StreamProgress | Progresses] end,
[StreamProgress],
ProgressesAcc0
);
false ->
ProgressesAcc0
end
end,
#{},
S
).
run_scheduled_actions(S, Agent, ScheduledActions) ->
maps:fold(
fun(ShareTopicFilter, Action0, {AgentAcc0, ScheduledActionsAcc}) ->
case run_scheduled_action(S, AgentAcc0, ShareTopicFilter, Action0) of
{ok, AgentAcc1} ->
{AgentAcc1, maps:remove(ShareTopicFilter, ScheduledActionsAcc)};
{continue, Action1} ->
{AgentAcc0, ScheduledActionsAcc#{ShareTopicFilter => Action1}}
end
end,
{Agent, ScheduledActions},
ScheduledActions
).
run_scheduled_action(
S,
Agent0,
ShareTopicFilter,
#{type := Type, stream_keys_to_wait := StreamKeysToWait0, progresses := Progresses0} = Action
) ->
emqx_persistent_session_ds_shared_subs_agent:subscription().
to_agent_subscription(_S, Subscription) ->
StreamKeysToWait1 = filter_unfinished_streams(S, StreamKeysToWait0),
Progresses1 = stream_progresses(S, StreamKeysToWait0 -- StreamKeysToWait1) ++ Progresses0,
case StreamKeysToWait1 of
[] ->
?tp(warning, shared_subs_schedule_action_complete, #{
share_topic_filter => ShareTopicFilter,
progresses => format_stream_progresses(Progresses1),
type => Type
}),
%% Regular progress won't se unsubscribed streams, so we need to
%% send the progress explicitly.
Agent1 = emqx_persistent_session_ds_shared_subs_agent:on_stream_progress(
Agent0, #{ShareTopicFilter => Progresses1}
),
case Type of
{?schedule_subscribe, SubOpts} ->
{ok,
emqx_persistent_session_ds_shared_subs_agent:on_subscribe(
Agent1, ShareTopicFilter, SubOpts
)};
?schedule_unsubscribe ->
{ok,
emqx_persistent_session_ds_shared_subs_agent:on_unsubscribe(
Agent1, ShareTopicFilter, Progresses1
)}
end;
_ ->
Action1 = Action#{stream_keys_to_wait => StreamKeysToWait1, progresses => Progresses1},
?tp(warning, shared_subs_schedule_action_continue, #{
share_topic_filter => ShareTopicFilter,
new_action => format_schedule_action(Action1)
}),
{continue, Action1}
end.
filter_unfinished_streams(S, StreamKeysToWait) ->
CommQos1 = emqx_persistent_session_ds_state:get_seqno(?committed(?QOS_1), S),
CommQos2 = emqx_persistent_session_ds_state:get_seqno(?committed(?QOS_2), S),
lists:filter(
fun(Key) ->
case emqx_persistent_session_ds_state:get_stream(Key, S) of
undefined ->
%% This should not happen: we should see any stream
%% in completed state before deletion
true;
SRS ->
not is_stream_fully_acked(CommQos1, CommQos2, SRS)
end
end,
StreamKeysToWait
).
stream_progresses(S, StreamKeys) ->
CommQos1 = emqx_persistent_session_ds_state:get_seqno(?committed(?QOS_1), S),
CommQos2 = emqx_persistent_session_ds_state:get_seqno(?committed(?QOS_2), S),
lists:map(
fun({_SubId, Stream} = Key) ->
SRS = emqx_persistent_session_ds_state:get_stream(Key, S),
stream_progress(CommQos1, CommQos2, Stream, SRS)
end,
StreamKeys
).
%%--------------------------------------------------------------------
%% on_disconnect
on_disconnect(S0, #{agent := Agent0} = SharedSubS0) ->
S1 = revoke_all_streams(S0),
Progresses = all_stream_progresses(S1, Agent0, _NeedUnacked = true),
Agent1 = emqx_persistent_session_ds_shared_subs_agent:on_disconnect(Agent0, Progresses),
SharedSubS1 = SharedSubS0#{agent => Agent1, scheduled_actions => #{}},
{S1, SharedSubS1}.
%%--------------------------------------------------------------------
%% on_disconnect helpers
revoke_all_streams(S0) ->
fold_shared_stream_states(
fun(ShareTopicFilter, Stream, _SRS, S) ->
revoke_stream(#{share_topic_filter => ShareTopicFilter, stream => Stream}, S)
end,
S0,
S0
).
%%--------------------------------------------------------------------
%% on_info
-spec on_info(emqx_persistent_session_ds_state:t(), t(), term()) ->
{emqx_persistent_session_ds_state:t(), t()}.
on_info(S, #{agent := Agent0} = SharedSubS0, Info) ->
Agent1 = emqx_persistent_session_ds_shared_subs_agent:on_info(Agent0, Info),
SharedSubS1 = SharedSubS0#{agent => Agent1},
{S, SharedSubS1}.
%%--------------------------------------------------------------------
%% to_map
-spec to_map(emqx_persistent_session_ds_state:t(), t()) -> map().
to_map(S, _SharedSubS) ->
fold_shared_subs(
fun(ShareTopicFilter, _, Acc) -> Acc#{ShareTopicFilter => lookup(ShareTopicFilter, S)} end,
#{},
S
).
%%--------------------------------------------------------------------
%% cold_get_subscription
-spec cold_get_subscription(emqx_persistent_session_ds:id(), share_topic_filter()) ->
emqx_persistent_session_ds:subscription() | undefined.
cold_get_subscription(SessionId, ShareTopicFilter) ->
case emqx_persistent_session_ds_state:cold_get_subscription(SessionId, ShareTopicFilter) of
[Sub = #{current_state := SStateId}] ->
case
emqx_persistent_session_ds_state:cold_get_subscription_state(SessionId, SStateId)
of
[#{subopts := Subopts}] ->
Sub#{subopts => Subopts};
_ ->
undefined
end;
_ ->
undefined
end.
%%--------------------------------------------------------------------
%% Generic helpers
%%--------------------------------------------------------------------
lookup(ShareTopicFilter, S) ->
case emqx_persistent_session_ds_state:get_subscription(ShareTopicFilter, S) of
Sub = #{current_state := SStateId} ->
case emqx_persistent_session_ds_state:get_subscription_state(SStateId, S) of
#{subopts := SubOpts} ->
Sub#{subopts => SubOpts};
undefined ->
undefined
end;
undefined ->
undefined
end.
stream_keys_by_sub_id(S, MatchSubId) ->
emqx_persistent_session_ds_state:fold_streams(
fun({SubId, _Stream} = StreamKey, _SRS, StreamKeys) ->
case SubId of
MatchSubId ->
[StreamKey | StreamKeys];
_ ->
StreamKeys
end
end,
[],
S
).
stream_progress(
CommQos1,
CommQos2,
Stream,
#srs{
it_end = EndIt,
it_begin = BeginIt
} = SRS
) ->
Iterator =
case is_stream_fully_acked(CommQos1, CommQos2, SRS) of
true -> EndIt;
false -> BeginIt
end,
#{
stream => Stream,
progress => #{
iterator => Iterator
},
use_finished => is_use_finished(SRS)
}.
fold_shared_subs(Fun, Acc, S) ->
emqx_persistent_session_ds_state:fold_subscriptions(
fun
(#share{} = ShareTopicFilter, Sub, Acc0) -> Fun(ShareTopicFilter, Sub, Acc0);
(_, _Sub, Acc0) -> Acc0
end,
Acc,
S
).
fold_shared_stream_states(Fun, Acc, S) ->
%% TODO
%% do we need anything from sub state?
%% Optimize or cache
ShareTopicFilters = fold_shared_subs(
fun
(#share{} = ShareTopicFilter, #{id := Id} = _Sub, Acc0) ->
Acc0#{Id => ShareTopicFilter};
(_, _, Acc0) ->
Acc0
end,
#{},
S
),
emqx_persistent_session_ds_state:fold_streams(
fun({SubId, Stream}, SRS, Acc0) ->
case ShareTopicFilters of
#{SubId := ShareTopicFilter} ->
Fun(ShareTopicFilter, Stream, SRS, Acc0);
_ ->
Acc0
end
end,
Acc,
S
).
to_agent_subscription(_S, Subscription) ->
maps:with([start_time], Subscription).
-spec agent_opts(opts()) -> emqx_persistent_session_ds_shared_subs_agent:opts().
agent_opts(#{session_id := SessionId}) ->
#{session_id => SessionId}.
-dialyzer({nowarn_function, now_ms/0}).
now_ms() ->
erlang:system_time(millisecond).
is_use_finished(#srs{unsubscribed = Unsubscribed}) ->
Unsubscribed.
is_stream_started(CommQos1, CommQos2, #srs{first_seqno_qos1 = Q1, last_seqno_qos1 = Q2}) ->
(CommQos1 >= Q1) or (CommQos2 >= Q2).
is_stream_fully_acked(_, _, #srs{
first_seqno_qos1 = Q1, last_seqno_qos1 = Q1, first_seqno_qos2 = Q2, last_seqno_qos2 = Q2
}) ->
%% Streams where the last chunk doesn't contain any QoS1 and 2
%% messages are considered fully acked:
true;
is_stream_fully_acked(Comm1, Comm2, #srs{last_seqno_qos1 = S1, last_seqno_qos2 = S2}) ->
(Comm1 >= S1) andalso (Comm2 >= S2).
%%--------------------------------------------------------------------
%% Formatters
%%--------------------------------------------------------------------
format_schedule_action(#{
type := Type, progresses := Progresses, stream_keys_to_wait := StreamKeysToWait
}) ->
#{
type => Type,
progresses => format_stream_progresses(Progresses),
stream_keys_to_wait => format_stream_keys(StreamKeysToWait)
}.
format_stream_progresses(Streams) ->
lists:map(
fun format_stream_progress/1,
Streams
).
format_stream_progress(#{stream := Stream, progress := Progress} = Value) ->
Value#{stream => format_opaque(Stream), progress => format_progress(Progress)}.
format_progress(#{iterator := Iterator} = Progress) ->
Progress#{iterator => format_opaque(Iterator)}.
format_stream_key(beginning) -> beginning;
format_stream_key({SubId, Stream}) -> {SubId, format_opaque(Stream)}.
format_stream_keys(StreamKeys) ->
lists:map(
fun format_stream_key/1,
StreamKeys
).
format_lease_events(Events) ->
lists:map(
fun format_lease_event/1,
Events
).
format_lease_event(#{stream := Stream, progress := Progress} = Event) ->
Event#{stream => format_opaque(Stream), progress => format_progress(Progress)};
format_lease_event(#{stream := Stream} = Event) ->
Event#{stream => format_opaque(Stream)}.
format_opaque(Opaque) ->
erlang:phash2(Opaque).

53
apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_shared_subs_agent.erl
View File

@ -15,7 +15,7 @@
}.
-type t() :: term().
-type topic_filter() :: emqx_persistent_session_ds:share_topic_filter().
-type share_topic_filter() :: emqx_persistent_session_ds:share_topic_filter().
-type opts() :: #{
session_id := session_id()
@ -28,41 +28,44 @@
-type stream_lease() :: #{
type => lease,
%% Used as "external" subscription_id
topic_filter := topic_filter(),
share_topic_filter := share_topic_filter(),
stream := emqx_ds:stream(),
iterator := emqx_ds:iterator()
}.
-type stream_revoke() :: #{
type => revoke,
topic_filter := topic_filter(),
share_topic_filter := share_topic_filter(),
stream := emqx_ds:stream()
}.
-type stream_lease_event() :: stream_lease() | stream_revoke().
-type stream_progress() :: #{
topic_filter := topic_filter(),
share_topic_filter := share_topic_filter(),
stream := emqx_ds:stream(),
iterator := emqx_ds:iterator()
iterator := emqx_ds:iterator(),
use_finished := boolean()
}.
-export_type([
t/0,
subscription/0,
session_id/0,
stream_lease/0,
stream_lease_event/0,
opts/0
]).
-export([
new/1,
open/2,
can_subscribe/3,
on_subscribe/3,
on_unsubscribe/2,
on_unsubscribe/3,
on_stream_progress/2,
on_info/2,
on_disconnect/2,
renew_streams/1
]).
@ -77,12 +80,13 @@
%%--------------------------------------------------------------------
-callback new(opts()) -> t().
-callback open([{topic_filter(), subscription()}], opts()) -> t().
-callback on_subscribe(t(), topic_filter(), emqx_types:subopts()) ->
{ok, t()} | {error, term()}.
-callback on_unsubscribe(t(), topic_filter()) -> t().
-callback open([{share_topic_filter(), subscription()}], opts()) -> t().
-callback can_subscribe(t(), share_topic_filter(), emqx_types:subopts()) -> ok | {error, term()}.
-callback on_subscribe(t(), share_topic_filter(), emqx_types:subopts()) -> t().
-callback on_unsubscribe(t(), share_topic_filter(), [stream_progress()]) -> t().
-callback on_disconnect(t(), [stream_progress()]) -> t().
-callback renew_streams(t()) -> {[stream_lease_event()], t()}.
-callback on_stream_progress(t(), [stream_progress()]) -> t().
-callback on_stream_progress(t(), #{share_topic_filter() => [stream_progress()]}) -> t().
-callback on_info(t(), term()) -> t().
%%--------------------------------------------------------------------
@ -93,24 +97,31 @@
new(Opts) ->
?shared_subs_agent:new(Opts).
-spec open([{topic_filter(), subscription()}], opts()) -> t().
-spec open([{share_topic_filter(), subscription()}], opts()) -> t().
open(Topics, Opts) ->
?shared_subs_agent:open(Topics, Opts).
-spec on_subscribe(t(), topic_filter(), emqx_types:subopts()) ->
{ok, t()} | {error, emqx_types:reason_code()}.
on_subscribe(Agent, TopicFilter, SubOpts) ->
?shared_subs_agent:on_subscribe(Agent, TopicFilter, SubOpts).
-spec can_subscribe(t(), share_topic_filter(), emqx_types:subopts()) -> ok | {error, term()}.
can_subscribe(Agent, ShareTopicFilter, SubOpts) ->
?shared_subs_agent:can_subscribe(Agent, ShareTopicFilter, SubOpts).
-spec on_unsubscribe(t(), topic_filter()) -> t().
on_unsubscribe(Agent, TopicFilter) ->
?shared_subs_agent:on_unsubscribe(Agent, TopicFilter).
-spec on_subscribe(t(), share_topic_filter(), emqx_types:subopts()) -> t().
on_subscribe(Agent, ShareTopicFilter, SubOpts) ->
?shared_subs_agent:on_subscribe(Agent, ShareTopicFilter, SubOpts).
-spec on_unsubscribe(t(), share_topic_filter(), [stream_progress()]) -> t().
on_unsubscribe(Agent, ShareTopicFilter, StreamProgresses) ->
?shared_subs_agent:on_unsubscribe(Agent, ShareTopicFilter, StreamProgresses).
-spec on_disconnect(t(), #{share_topic_filter() => [stream_progress()]}) -> t().
on_disconnect(Agent, StreamProgresses) ->
?shared_subs_agent:on_disconnect(Agent, StreamProgresses).
-spec renew_streams(t()) -> {[stream_lease_event()], t()}.
renew_streams(Agent) ->
?shared_subs_agent:renew_streams(Agent).
-spec on_stream_progress(t(), [stream_progress()]) -> t().
-spec on_stream_progress(t(), #{share_topic_filter() => [stream_progress()]}) -> t().
on_stream_progress(Agent, StreamProgress) ->
?shared_subs_agent:on_stream_progress(Agent, StreamProgress).

14
apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_shared_subs_null_agent.erl
View File

@ -9,11 +9,13 @@
-export([
new/1,
open/2,
can_subscribe/3,
on_subscribe/3,
on_unsubscribe/2,
on_unsubscribe/3,
on_stream_progress/2,
on_info/2,
on_disconnect/2,
renew_streams/1
]).
@ -30,10 +32,16 @@ new(_Opts) ->
open(_Topics, _Opts) ->
undefined.
on_subscribe(_Agent, _TopicFilter, _SubOpts) ->
can_subscribe(_Agent, _TopicFilter, _SubOpts) ->
{error, ?RC_SHARED_SUBSCRIPTIONS_NOT_SUPPORTED}.
on_unsubscribe(Agent, _TopicFilter) ->
on_subscribe(Agent, _TopicFilter, _SubOpts) ->
Agent.
on_unsubscribe(Agent, _TopicFilter, _Progresses) ->
Agent.
on_disconnect(Agent, _) ->
Agent.
renew_streams(Agent) ->

4
apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_state.erl
View File

@ -399,7 +399,9 @@ new_id(Rec) ->
get_subscription(TopicFilter, Rec) ->
gen_get(?subscriptions, TopicFilter, Rec).
-spec cold_get_subscription(emqx_persistent_session_ds:id(), emqx_types:topic()) ->
-spec cold_get_subscription(
emqx_persistent_session_ds:id(), emqx_types:topic() | emqx_types:share()
) ->
[emqx_persistent_session_ds_subs:subscription()].
cold_get_subscription(SessionId, Topic) ->
kv_pmap_read(?subscription_tab, SessionId, Topic).

2
apps/emqx/src/emqx_persistent_session_ds/emqx_ps_ds_int.hrl
View File

@ -21,7 +21,7 @@
-record(ps_route, {
topic :: binary(),
dest :: emqx_persistent_session_ds:id() | '_'
dest :: emqx_persistent_session_ds_router:dest() | '_'
}).
-record(ps_routeidx, {

1
apps/emqx/src/emqx_persistent_session_ds/shared_subs_agent.hrl
View File

@ -21,6 +21,7 @@
%% Till full implementation we need to dispach to the null agent.
%% It will report "not implemented" error for attempts to use shared subscriptions.
-define(shared_subs_agent, emqx_persistent_session_ds_shared_subs_null_agent).
% -define(shared_subs_agent, emqx_ds_shared_sub_agent).
%% end of -ifdef(TEST).
-endif.

13
apps/emqx/src/emqx_schema.erl
View File

@ -351,6 +351,7 @@ fields("authz_cache") ->
#{
default => true,
required => true,
importance => ?IMPORTANCE_NO_DOC,
desc => ?DESC(fields_cache_enable)
}
)},
@ -387,6 +388,7 @@ fields("flapping_detect") ->
boolean(),
#{
default => false,
%% importance => ?IMPORTANCE_NO_DOC,
desc => ?DESC(flapping_detect_enable)
}
)},
@ -423,6 +425,7 @@ fields("force_shutdown") ->
boolean(),
#{
default => true,
importance => ?IMPORTANCE_NO_DOC,
desc => ?DESC(force_shutdown_enable)
}
)},
@ -452,6 +455,7 @@ fields("overload_protection") ->
boolean(),
#{
desc => ?DESC(overload_protection_enable),
%% importance => ?IMPORTANCE_NO_DOC,
default => false
}
)},
@ -512,7 +516,11 @@ fields("force_gc") ->
{"enable",
sc(
boolean(),
#{default => true, desc => ?DESC(force_gc_enable)}
#{
default => true,
importance => ?IMPORTANCE_NO_DOC,
desc => ?DESC(force_gc_enable)
}
)},
{"count",
sc(
@ -1665,6 +1673,7 @@ fields("durable_sessions") ->
sc(
boolean(), #{
desc => ?DESC(durable_sessions_enable),
%% importance => ?IMPORTANCE_NO_DOC,
default => false
}
)},
@ -1888,6 +1897,7 @@ base_listener(Bind) ->
#{
default => true,
aliases => [enabled],
importance => ?IMPORTANCE_NO_DOC,
desc => ?DESC(fields_listener_enabled)
}
)},
@ -2416,6 +2426,7 @@ client_ssl_opts_schema(Defaults) ->
boolean(),
#{
default => false,
%% importance => ?IMPORTANCE_NO_DOC,
desc => ?DESC(client_ssl_opts_schema_enable)
}
)},

1
apps/emqx/test/emqx_common_test_helpers.erl
View File

@ -78,6 +78,7 @@
start_epmd/0,
start_peer/2,
stop_peer/1,
ebin_path/0,
listener_port/2
]).

15
apps/emqx/test/emqx_cth_suite.erl
View File

@ -79,6 +79,8 @@
%% "Unofficial" `emqx_config_handler' and `emqx_conf' APIs
-export([schema_module/0, upgrade_raw_conf/1]).
-export([skip_if_oss/0]).
-export_type([appspec/0]).
-export_type([appspec_opts/0]).
@ -389,6 +391,8 @@ default_appspec(emqx_schema_validation, _SuiteOpts) ->
#{schema_mod => emqx_schema_validation_schema, config => #{}};
default_appspec(emqx_message_transformation, _SuiteOpts) ->
#{schema_mod => emqx_message_transformation_schema, config => #{}};
default_appspec(emqx_ds_shared_sub, _SuiteOpts) ->
#{schema_mod => emqx_ds_shared_sub_schema, config => #{}};
default_appspec(_, _) ->
#{}.
@ -519,3 +523,14 @@ upgrade_raw_conf(Conf) ->
ce ->
emqx_conf_schema:upgrade_raw_conf(Conf)
end.
skip_if_oss() ->
try emqx_release:edition() of
ee ->
false;
_ ->
{skip, not_supported_in_oss}
catch
error:undef ->
{skip, standalone_not_supported}
end.

2
apps/emqx/test/emqx_exclusive_sub_SUITE.erl
View File

@ -56,6 +56,8 @@ t_exclusive_sub(_) ->
{ok, _} = emqtt:connect(C1),
?CHECK_SUB(C1, 0),
?CHECK_SUB(C1, 0),
{ok, C2} = emqtt:start_link([
{clientid, <<"client2">>},
{clean_start, false},

78
apps/emqx/test/emqx_log_throttler_SUITE.erl
View File

@ -26,6 +26,7 @@
%% Have to use real msgs, as the schema is guarded by enum.
-define(THROTTLE_MSG, authorization_permission_denied).
-define(THROTTLE_MSG1, cannot_publish_to_topic_due_to_not_authorized).
-define(THROTTLE_UNRECOVERABLE_MSG, unrecoverable_resource_error).
-define(TIME_WINDOW, <<"1s">>).
all() -> emqx_common_test_helpers:all(?MODULE).
@ -59,6 +60,11 @@ end_per_suite(Config) ->
emqx_cth_suite:stop(?config(suite_apps, Config)),
emqx_config:delete_override_conf_files().
init_per_testcase(t_throttle_recoverable_msg, Config) ->
ok = snabbkaffe:start_trace(),
[?THROTTLE_MSG] = Conf = emqx:get_config([log, throttling, msgs]),
{ok, _} = emqx_conf:update([log, throttling, msgs], [?THROTTLE_UNRECOVERABLE_MSG | Conf], #{}),
Config;
init_per_testcase(t_throttle_add_new_msg, Config) ->
ok = snabbkaffe:start_trace(),
[?THROTTLE_MSG] = Conf = emqx:get_config([log, throttling, msgs]),
@ -72,6 +78,10 @@ init_per_testcase(_TC, Config) ->
ok = snabbkaffe:start_trace(),
Config.
end_per_testcase(t_throttle_recoverable_msg, _Config) ->
ok = snabbkaffe:stop(),
{ok, _} = emqx_conf:update([log, throttling, msgs], [?THROTTLE_MSG], #{}),
ok;
end_per_testcase(t_throttle_add_new_msg, _Config) ->
ok = snabbkaffe:stop(),
{ok, _} = emqx_conf:update([log, throttling, msgs], [?THROTTLE_MSG], #{}),
@ -101,8 +111,8 @@ t_throttle(_Config) ->
5000
),
?assert(emqx_log_throttler:allow(?THROTTLE_MSG)),
?assertNot(emqx_log_throttler:allow(?THROTTLE_MSG)),
?assert(emqx_log_throttler:allow(?THROTTLE_MSG, undefined)),
?assertNot(emqx_log_throttler:allow(?THROTTLE_MSG, undefined)),
{ok, _} = ?block_until(
#{
?snk_kind := log_throttler_dropped,
@ -115,14 +125,48 @@ t_throttle(_Config) ->
[]
).
t_throttle_recoverable_msg(_Config) ->
ResourceId = <<"resource_id">>,
ThrottledMsg = iolist_to_binary([atom_to_list(?THROTTLE_UNRECOVERABLE_MSG), ":", ResourceId]),
?check_trace(
begin
%% Warm-up and block to increase the probability that next events
%% will be in the same throttling time window.
{ok, _} = ?block_until(
#{?snk_kind := log_throttler_new_msg, throttled_msg := ?THROTTLE_UNRECOVERABLE_MSG},
5000
),
{_, {ok, _}} = ?wait_async_action(
events(?THROTTLE_UNRECOVERABLE_MSG, ResourceId),
#{
?snk_kind := log_throttler_dropped,
throttled_msg := ThrottledMsg
},
5000
),
?assert(emqx_log_throttler:allow(?THROTTLE_UNRECOVERABLE_MSG, ResourceId)),
?assertNot(emqx_log_throttler:allow(?THROTTLE_UNRECOVERABLE_MSG, ResourceId)),
{ok, _} = ?block_until(
#{
?snk_kind := log_throttler_dropped,
throttled_msg := ThrottledMsg,
dropped_count := 1
},
3000
)
end,
[]
).
t_throttle_add_new_msg(_Config) ->
?check_trace(
begin
{ok, _} = ?block_until(
#{?snk_kind := log_throttler_new_msg, throttled_msg := ?THROTTLE_MSG1}, 5000
),
?assert(emqx_log_throttler:allow(?THROTTLE_MSG1)),
?assertNot(emqx_log_throttler:allow(?THROTTLE_MSG1)),
?assert(emqx_log_throttler:allow(?THROTTLE_MSG1, undefined)),
?assertNot(emqx_log_throttler:allow(?THROTTLE_MSG1, undefined)),
{ok, _} = ?block_until(
#{
?snk_kind := log_throttler_dropped,
@ -137,10 +181,15 @@ t_throttle_add_new_msg(_Config) ->
t_throttle_no_msg(_Config) ->
%% Must simply pass with no crashes
?assert(emqx_log_throttler:allow(no_test_throttle_msg)),
?assert(emqx_log_throttler:allow(no_test_throttle_msg)),
timer:sleep(10),
?assert(erlang:is_process_alive(erlang:whereis(emqx_log_throttler))).
Pid = erlang:whereis(emqx_log_throttler),
?assert(emqx_log_throttler:allow(no_test_throttle_msg, undefined)),
?assert(emqx_log_throttler:allow(no_test_throttle_msg, undefined)),
%% assert process is not restarted
?assertEqual(Pid, erlang:whereis(emqx_log_throttler)),
%% make a gen_call to ensure the process is alive
%% note: this call result in an 'unexpected_call' error log.
?assertEqual(ignored, gen_server:call(Pid, probe)),
ok.
t_update_time_window(_Config) ->
?check_trace(
@ -168,8 +217,8 @@ t_throttle_debug_primary_level(_Config) ->
#{?snk_kind := log_throttler_dropped, throttled_msg := ?THROTTLE_MSG},
5000
),
?assert(emqx_log_throttler:allow(?THROTTLE_MSG)),
?assertNot(emqx_log_throttler:allow(?THROTTLE_MSG)),
?assert(emqx_log_throttler:allow(?THROTTLE_MSG, undefined)),
?assertNot(emqx_log_throttler:allow(?THROTTLE_MSG, undefined)),
{ok, _} = ?block_until(
#{
?snk_kind := log_throttler_dropped,
@ -187,10 +236,13 @@ t_throttle_debug_primary_level(_Config) ->
%%--------------------------------------------------------------------
events(Msg) ->
events(100, Msg).
events(100, Msg, undefined).
events(N, Msg) ->
[emqx_log_throttler:allow(Msg) || _ <- lists:seq(1, N)].
events(Msg, Id) ->
events(100, Msg, Id).
events(N, Msg, Id) ->
[emqx_log_throttler:allow(Msg, Id) || _ <- lists:seq(1, N)].
module_exists(Mod) ->
case erlang:module_loaded(Mod) of

2
apps/emqx/test/emqx_persistent_messages_SUITE.erl
View File

@ -573,7 +573,7 @@ app_specs(Opts) ->
cluster() ->
ExtraConf = "\n durable_storage.messages.n_sites = 2",
Spec = #{role => core, apps => app_specs(#{extra_emqx_conf => ExtraConf})},
Spec = #{apps => app_specs(#{extra_emqx_conf => ExtraConf})},
[
{persistent_messages_SUITE1, Spec},
{persistent_messages_SUITE2, Spec}

10
apps/emqx/test/emqx_routing_SUITE.erl
View File

@ -64,10 +64,17 @@ init_per_group(routing_schema_v2, Config) ->
init_per_group(batch_sync_on, Config) ->
[{emqx_config, "broker.routing.batch_sync.enable_on = all"} | Config];
init_per_group(batch_sync_replicants, Config) ->
case emqx_cth_suite:skip_if_oss() of
false ->
[{emqx_config, "broker.routing.batch_sync.enable_on = replicant"} | Config];
True ->
True
end;
init_per_group(batch_sync_off, Config) ->
[{emqx_config, "broker.routing.batch_sync.enable_on = none"} | Config];
init_per_group(cluster, Config) ->
case emqx_cth_suite:skip_if_oss() of
false ->
WorkDir = emqx_cth_suite:work_dir(Config),
NodeSpecs = [
{emqx_routing_SUITE1, #{apps => [mk_emqx_appspec(1, Config)], role => core}},
@ -76,6 +83,9 @@ init_per_group(cluster, Config) ->
],
Nodes = emqx_cth_cluster:start(NodeSpecs, #{work_dir => WorkDir}),
[{cluster, Nodes} | Config];
True ->
True
end;
init_per_group(GroupName, Config) when
GroupName =:= single_batch_on;
GroupName =:= single

5
apps/emqx/test/emqx_shared_sub_SUITE.erl
View File

@ -1247,7 +1247,7 @@ recv_msgs(Count, Msgs) ->
start_peer(Name, Port) ->
{ok, Node} = emqx_cth_peer:start_link(
Name,
ebin_path()
emqx_common_test_helpers:ebin_path()
),
pong = net_adm:ping(Node),
setup_node(Node, Port),
@ -1261,9 +1261,6 @@ host() ->
[_, Host] = string:tokens(atom_to_list(node()), "@"),
Host.
ebin_path() ->
["-pa" | code:get_path()].
setup_node(Node, Port) ->
EnvHandler =
fun(_) ->

2
apps/emqx_auth/include/emqx_authn.hrl
View File

@ -28,7 +28,7 @@
-type authenticator_id() :: binary().
-define(AUTHN_RESOURCE_GROUP, <<"emqx_authn">>).
-define(AUTHN_RESOURCE_GROUP, <<"authn">>).
%% VAR_NS_CLIENT_ATTRS is added here because it can be initialized before authn.
%% NOTE: authn return may add more to (or even overwrite) client_attrs.

2
apps/emqx_auth/include/emqx_authz.hrl
View File

@ -156,7 +156,7 @@
count => 1
}).
-define(AUTHZ_RESOURCE_GROUP, <<"emqx_authz">>).
-define(AUTHZ_RESOURCE_GROUP, <<"authz">>).
-define(AUTHZ_FEATURES, [rich_actions]).

1
apps/emqx_auth/src/emqx_authn/emqx_authn_schema.erl
View File

@ -203,6 +203,7 @@ common_fields() ->
enable(type) -> boolean();
enable(default) -> true;
enable(importance) -> ?IMPORTANCE_NO_DOC;
enable(desc) -> ?DESC(?FUNCTION_NAME);
enable(_) -> undefined.

2
apps/emqx_auth/src/emqx_authz/emqx_authz_rule.erl
View File

@ -198,7 +198,7 @@ qos_from_opts(Opts) ->
)
end
catch
{bad_qos, QoS} ->
throw:{bad_qos, QoS} ->
throw(#{
reason => invalid_authorization_qos,
qos => QoS

7
apps/emqx_auth/src/emqx_authz/emqx_authz_schema.erl
View File

@ -170,7 +170,12 @@ api_authz_refs() ->
authz_common_fields(Type) ->
[
{type, ?HOCON(Type, #{required => true, desc => ?DESC(type)})},
{enable, ?HOCON(boolean(), #{default => true, desc => ?DESC(enable)})}
{enable,
?HOCON(boolean(), #{
default => true,
importance => ?IMPORTANCE_NO_DOC,
desc => ?DESC(enable)
})}
].
source_types() ->

57
apps/emqx_auth/src/emqx_authz/emqx_authz_utils.erl
View File

@ -16,6 +16,9 @@
-module(emqx_authz_utils).
-feature(maybe_expr, enable).
-include_lib("emqx/include/emqx_placeholder.hrl").
-include_lib("emqx_authz.hrl").
-include_lib("snabbkaffe/include/trace.hrl").
@ -28,7 +31,7 @@
remove_resource/1,
update_config/2,
vars_for_rule_query/2,
parse_rule_from_row/2
do_authorize/6
]).
-export([
@ -133,14 +136,18 @@ content_type(Headers) when is_list(Headers) ->
-define(RAW_RULE_KEYS, [<<"permission">>, <<"action">>, <<"topic">>, <<"qos">>, <<"retain">>]).
parse_rule_from_row(ColumnNames, Row) ->
RuleRaw = maps:with(?RAW_RULE_KEYS, maps:from_list(lists:zip(ColumnNames, to_list(Row)))),
case emqx_authz_rule_raw:parse_rule(RuleRaw) of
-spec parse_rule_from_row([binary()], [binary()] | map()) ->
{ok, emqx_authz_rule:rule()} | {error, term()}.
parse_rule_from_row(_ColumnNames, RuleMap = #{}) ->
case emqx_authz_rule_raw:parse_rule(RuleMap) of
{ok, {Permission, Action, Topics}} ->
emqx_authz_rule:compile({Permission, all, Action, Topics});
{ok, emqx_authz_rule:compile({Permission, all, Action, Topics})};
{error, Reason} ->
error(Reason)
end.
{error, Reason}
end;
parse_rule_from_row(ColumnNames, Row) ->
RuleMap = maps:with(?RAW_RULE_KEYS, maps:from_list(lists:zip(ColumnNames, to_list(Row)))),
parse_rule_from_row(ColumnNames, RuleMap).
vars_for_rule_query(Client, ?authz_action(PubSub, Qos) = Action) ->
Client#{
@ -157,3 +164,39 @@ to_list(Tuple) when is_tuple(Tuple) ->
tuple_to_list(Tuple);
to_list(List) when is_list(List) ->
List.
do_authorize(Type, Client, Action, Topic, ColumnNames, Row) ->
try
maybe
{ok, Rule} ?= parse_rule_from_row(ColumnNames, Row),
{matched, Permission} ?= emqx_authz_rule:match(Client, Action, Topic, Rule),
{matched, Permission}
else
nomatch ->
nomatch;
{error, Reason0} ->
log_match_rule_error(Type, Row, Reason0),
nomatch
end
catch
throw:Reason1 ->
log_match_rule_error(Type, Row, Reason1),
nomatch
end.
log_match_rule_error(Type, Row, Reason0) ->
Msg0 = #{
msg => "match_rule_error",
rule => Row,
type => Type
},
Msg1 =
case is_map(Reason0) of
true -> maps:merge(Msg0, Reason0);
false -> Msg0#{reason => Reason0}
end,
?SLOG(
error,
Msg1,
#{tag => "AUTHZ"}
).

8
apps/emqx_auth/test/emqx_authn/emqx_authn_schema_SUITE.erl
View File

@ -122,14 +122,6 @@ t_union_member_selector(_) ->
},
check(BadMechanism)
),
BadCombination = Base#{<<"mechanism">> => <<"scram">>, <<"backend">> => <<"http">>},
?assertThrow(
#{
reason := "unknown_mechanism",
expected := "password_based"
},
check(BadCombination)
),
ok.
t_http_auth_selector(_) ->

4
apps/emqx_auth/test/emqx_authz/emqx_authz_api_cluster_SUITE.erl
View File

@ -118,8 +118,8 @@ mk_cluster_spec(Opts) ->
Node1Apps = Apps ++ [{emqx_dashboard, "dashboard.listeners.http {enable=true,bind=18083}"}],
Node2Apps = Apps,
[
{emqx_authz_api_cluster_SUITE1, Opts#{role => core, apps => Node1Apps}},
{emqx_authz_api_cluster_SUITE2, Opts#{role => core, apps => Node2Apps}}
{emqx_authz_api_cluster_SUITE1, Opts#{apps => Node1Apps}},
{emqx_authz_api_cluster_SUITE2, Opts#{apps => Node2Apps}}
].
request(Method, URL, Body, Config) ->

7
apps/emqx_auth_http/include/emqx_auth_http.hrl
View File

@ -22,8 +22,15 @@
-define(AUTHN_MECHANISM, password_based).
-define(AUTHN_MECHANISM_BIN, <<"password_based">>).
-define(AUTHN_MECHANISM_SCRAM, scram).
-define(AUTHN_MECHANISM_SCRAM_BIN, <<"scram">>).
-define(AUTHN_BACKEND, http).
-define(AUTHN_BACKEND_BIN, <<"http">>).
-define(AUTHN_TYPE, {?AUTHN_MECHANISM, ?AUTHN_BACKEND}).
-define(AUTHN_TYPE_SCRAM, {?AUTHN_MECHANISM_SCRAM, ?AUTHN_BACKEND}).
-define(AUTHN_DATA_FIELDS, [is_superuser, client_attrs, expire_at, acl]).
-endif.

2
apps/emqx_auth_http/src/emqx_auth_http_app.erl
View File

@ -25,10 +25,12 @@
start(_StartType, _StartArgs) ->
ok = emqx_authz:register_source(?AUTHZ_TYPE, emqx_authz_http),
ok = emqx_authn:register_provider(?AUTHN_TYPE, emqx_authn_http),
ok = emqx_authn:register_provider(?AUTHN_TYPE_SCRAM, emqx_authn_scram_restapi),
{ok, Sup} = emqx_auth_http_sup:start_link(),
{ok, Sup}.
stop(_State) ->
ok = emqx_authn:deregister_provider(?AUTHN_TYPE),
ok = emqx_authn:deregister_provider(?AUTHN_TYPE_SCRAM),
ok = emqx_authz:unregister_source(?AUTHZ_TYPE),
ok.

56
apps/emqx_auth_http/src/emqx_authn_http.erl
View File

@ -28,6 +28,15 @@
destroy/1
]).
-export([
with_validated_config/2,
generate_request/2,
request_for_log/2,
response_for_log/1,
extract_auth_data/2,
safely_parse_body/2
]).
-define(DEFAULT_CONTENT_TYPE, <<"application/json">>).
%%------------------------------------------------------------------------------
@ -187,23 +196,28 @@ handle_response(Headers, Body) ->
case safely_parse_body(ContentType, Body) of
{ok, NBody} ->
body_to_auth_data(NBody);
{error, Reason} ->
?TRACE_AUTHN_PROVIDER(
error,
"parse_http_response_failed",
#{content_type => ContentType, body => Body, reason => Reason}
),
{error, _Reason} ->
ignore
end.
body_to_auth_data(Body) ->
case maps:get(<<"result">>, Body, <<"ignore">>) of
<<"allow">> ->
extract_auth_data(http, Body);
<<"deny">> ->
{error, not_authorized};
<<"ignore">> ->
ignore;
_ ->
ignore
end.
extract_auth_data(Source, Body) ->
IsSuperuser = emqx_authn_utils:is_superuser(Body),
Attrs = emqx_authn_utils:client_attrs(Body),
try
ExpireAt = expire_at(Body),
ACL = acl(ExpireAt, Body),
ACL = acl(ExpireAt, Source, Body),
Result = merge_maps([ExpireAt, IsSuperuser, ACL, Attrs]),
{ok, Result}
catch
@ -214,13 +228,6 @@ body_to_auth_data(Body) ->
throw:Reason ->
?TRACE_AUTHN_PROVIDER("bad_response_body", Reason#{http_body => Body}),
{error, bad_username_or_password}
end;
<<"deny">> ->
{error, not_authorized};
<<"ignore">> ->
ignore;
_ ->
ignore
end.
merge_maps([]) -> #{};
@ -261,40 +268,43 @@ expire_sec(#{<<"expire_at">> := _}) ->
expire_sec(_) ->
undefined.
acl(#{expire_at := ExpireTimeMs}, #{<<"acl">> := Rules}) ->
acl(#{expire_at := ExpireTimeMs}, Source, #{<<"acl">> := Rules}) ->
#{
acl => #{
source_for_logging => http,
source_for_logging => Source,
rules => emqx_authz_rule_raw:parse_and_compile_rules(Rules),
%% It's seconds level precision (like JWT) for authz
%% see emqx_authz_client_info:check/1
expire => erlang:convert_time_unit(ExpireTimeMs, millisecond, second)
}
};
acl(_NoExpire, #{<<"acl">> := Rules}) ->
acl(_NoExpire, Source, #{<<"acl">> := Rules}) ->
#{
acl => #{
source_for_logging => http,
source_for_logging => Source,
rules => emqx_authz_rule_raw:parse_and_compile_rules(Rules)
}
};
acl(_, _) ->
acl(_, _, _) ->
#{}.
safely_parse_body(ContentType, Body) ->
try
parse_body(ContentType, Body)
catch
_Class:_Reason ->
_Class:Reason ->
?TRACE_AUTHN_PROVIDER(
error,
"parse_http_response_failed",
#{content_type => ContentType, body => Body, reason => Reason}
),
{error, invalid_body}
end.
parse_body(<<"application/json", _/binary>>, Body) ->
{ok, emqx_utils_json:decode(Body, [return_maps])};
parse_body(<<"application/x-www-form-urlencoded", _/binary>>, Body) ->
Flags = [<<"result">>, <<"is_superuser">>],
RawMap = maps:from_list(cow_qs:parse_qs(Body)),
NBody = maps:with(Flags, RawMap),
NBody = maps:from_list(cow_qs:parse_qs(Body)),
{ok, NBody};
parse_body(ContentType, _) ->
{error, {unsupported_content_type, ContentType}}.

8
apps/emqx_auth_http/src/emqx_authn_http_schema.erl
View File

@ -27,6 +27,8 @@
namespace/0
]).
-export([url/1, headers/1, headers_no_content_type/1, request_timeout/1]).
-include("emqx_auth_http.hrl").
-include_lib("emqx_auth/include/emqx_authn.hrl").
-include_lib("hocon/include/hoconsc.hrl").
@ -61,12 +63,6 @@ select_union_member(
got => Else
})
end;
select_union_member(#{<<"backend">> := ?AUTHN_BACKEND_BIN}) ->
throw(#{
reason => "unknown_mechanism",
expected => "password_based",
got => undefined
});
select_union_member(_Value) ->
undefined.

161
apps/emqx_auth_http/src/emqx_authn_scram_restapi.erl Normal file
View File

@ -0,0 +1,161 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
%% Note:
%% This is not an implementation of the RFC 7804:
%% Salted Challenge Response HTTP Authentication Mechanism.
%% This backend is an implementation of scram,
%% which uses an external web resource as a source of user information.
-module(emqx_authn_scram_restapi).
-feature(maybe_expr, enable).
-include("emqx_auth_http.hrl").
-include_lib("emqx/include/logger.hrl").
-include_lib("emqx_auth/include/emqx_authn.hrl").
-behaviour(emqx_authn_provider).
-export([
create/2,
update/2,
authenticate/2,
destroy/1
]).
-define(REQUIRED_USER_INFO_KEYS, [
<<"stored_key">>,
<<"server_key">>,
<<"salt">>
]).
%%------------------------------------------------------------------------------
%% APIs
%%------------------------------------------------------------------------------
create(_AuthenticatorID, Config) ->
create(Config).
create(Config0) ->
emqx_authn_http:with_validated_config(Config0, fun(Config, State) ->
ResourceId = emqx_authn_utils:make_resource_id(?MODULE),
% {Config, State} = parse_config(Config0),
{ok, _Data} = emqx_authn_utils:create_resource(
ResourceId,
emqx_bridge_http_connector,
Config
),
{ok, merge_scram_conf(Config, State#{resource_id => ResourceId})}
end).
update(Config0, #{resource_id := ResourceId} = _State) ->
emqx_authn_http:with_validated_config(Config0, fun(Config, NState) ->
% {Config, NState} = parse_config(Config0),
case emqx_authn_utils:update_resource(emqx_bridge_http_connector, Config, ResourceId) of
{error, Reason} ->
error({load_config_error, Reason});
{ok, _} ->
{ok, merge_scram_conf(Config, NState#{resource_id => ResourceId})}
end
end).
authenticate(
#{
auth_method := AuthMethod,
auth_data := AuthData,
auth_cache := AuthCache
} = Credential,
State
) ->
RetrieveFun = fun(Username) ->
retrieve(Username, Credential, State)
end,
OnErrFun = fun(Msg, Reason) ->
?TRACE_AUTHN_PROVIDER(Msg, #{
reason => Reason
})
end,
emqx_utils_scram:authenticate(
AuthMethod, AuthData, AuthCache, State, RetrieveFun, OnErrFun, ?AUTHN_DATA_FIELDS
);
authenticate(_Credential, _State) ->
ignore.
destroy(#{resource_id := ResourceId}) ->
_ = emqx_resource:remove_local(ResourceId),
ok.
%%--------------------------------------------------------------------
%% Internal functions
%%--------------------------------------------------------------------
retrieve(
Username,
Credential,
#{
resource_id := ResourceId,
method := Method,
request_timeout := RequestTimeout
} = State
) ->
Request = emqx_authn_http:generate_request(Credential#{username := Username}, State),
Response = emqx_resource:simple_sync_query(ResourceId, {Method, Request, RequestTimeout}),
?TRACE_AUTHN_PROVIDER("scram_restapi_response", #{
request => emqx_authn_http:request_for_log(Credential, State),
response => emqx_authn_http:response_for_log(Response),
resource => ResourceId
}),
case Response of
{ok, 200, Headers, Body} ->
handle_response(Headers, Body);
{ok, _StatusCode, _Headers} ->
{error, bad_response};
{ok, _StatusCode, _Headers, _Body} ->
{error, bad_response};
{error, _Reason} = Error ->
Error
end.
handle_response(Headers, Body) ->
ContentType = proplists:get_value(<<"content-type">>, Headers),
maybe
{ok, NBody} ?= emqx_authn_http:safely_parse_body(ContentType, Body),
{ok, UserInfo} ?= body_to_user_info(NBody),
{ok, AuthData} ?= emqx_authn_http:extract_auth_data(scram_restapi, NBody),
{ok, maps:merge(AuthData, UserInfo)}
end.
body_to_user_info(Body) ->
Required0 = maps:with(?REQUIRED_USER_INFO_KEYS, Body),
case maps:size(Required0) =:= erlang:length(?REQUIRED_USER_INFO_KEYS) of
true ->
case safely_convert_hex(Required0) of
{ok, Required} ->
{ok, emqx_utils_maps:safe_atom_key_map(Required)};
Error ->
?TRACE_AUTHN_PROVIDER("decode_keys_failed", #{http_body => Body}),
Error
end;
_ ->
?TRACE_AUTHN_PROVIDER("missing_requried_keys", #{http_body => Body}),
{error, bad_response}
end.
safely_convert_hex(Required) ->
try
{ok,
maps:map(
fun(_Key, Hex) ->
binary:decode_hex(Hex)
end,
Required
)}
catch
_Class:Reason ->
{error, Reason}
end.
merge_scram_conf(Conf, State) ->
maps:merge(maps:with([algorithm, iteration_count], Conf), State).

81
apps/emqx_auth_http/src/emqx_authn_scram_restapi_schema.erl Normal file
View File

@ -0,0 +1,81 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
-module(emqx_authn_scram_restapi_schema).
-behaviour(emqx_authn_schema).
-export([
fields/1,
validations/0,
desc/1,
refs/0,
select_union_member/1,
namespace/0
]).
-include("emqx_auth_http.hrl").
-include_lib("emqx_auth/include/emqx_authn.hrl").
-include_lib("hocon/include/hoconsc.hrl").
namespace() -> "authn".
refs() ->
[?R_REF(scram_restapi_get), ?R_REF(scram_restapi_post)].
select_union_member(
#{<<"mechanism">> := ?AUTHN_MECHANISM_SCRAM_BIN, <<"backend">> := ?AUTHN_BACKEND_BIN} = Value
) ->
case maps:get(<<"method">>, Value, undefined) of
<<"get">> ->
[?R_REF(scram_restapi_get)];
<<"post">> ->
[?R_REF(scram_restapi_post)];
Else ->
throw(#{
reason => "unknown_http_method",
expected => "get | post",
field_name => method,
got => Else
})
end;
select_union_member(_Value) ->
undefined.
fields(scram_restapi_get) ->
[
{method, #{type => get, required => true, desc => ?DESC(emqx_authn_http_schema, method)}},
{headers, fun emqx_authn_http_schema:headers_no_content_type/1}
] ++ common_fields();
fields(scram_restapi_post) ->
[
{method, #{type => post, required => true, desc => ?DESC(emqx_authn_http_schema, method)}},
{headers, fun emqx_authn_http_schema:headers/1}
] ++ common_fields().
desc(scram_restapi_get) ->
?DESC(emqx_authn_http_schema, get);
desc(scram_restapi_post) ->
?DESC(emqx_authn_http_schema, post);
desc(_) ->
undefined.
validations() ->
emqx_authn_http_schema:validations().
common_fields() ->
emqx_authn_schema:common_fields() ++
[
{mechanism, emqx_authn_schema:mechanism(?AUTHN_MECHANISM_SCRAM)},
{backend, emqx_authn_schema:backend(?AUTHN_BACKEND)},
{algorithm, fun emqx_authn_scram_mnesia_schema:algorithm/1},
{iteration_count, fun emqx_authn_scram_mnesia_schema:iteration_count/1},
{url, fun emqx_authn_http_schema:url/1},
{body,
hoconsc:mk(typerefl:alias("map", map([{fuzzy, term(), binary()}])), #{
required => false, desc => ?DESC(emqx_authn_http_schema, body)
})},
{request_timeout, fun emqx_authn_http_schema:request_timeout/1}
] ++
proplists:delete(pool_type, emqx_bridge_http_connector:fields(config)).

6
apps/emqx_auth_http/src/emqx_authz_http.erl
View File

@ -67,7 +67,11 @@ description() ->
create(Config) ->
NConfig = parse_config(Config),
ResourceId = emqx_authn_utils:make_resource_id(?MODULE),
{ok, _Data} = emqx_authz_utils:create_resource(ResourceId, emqx_bridge_http_connector, NConfig),
{ok, _Data} = emqx_authz_utils:create_resource(
ResourceId,
emqx_bridge_http_connector,
NConfig
),
NConfig#{annotations => #{id => ResourceId}}.
update(Config) ->

509
apps/emqx_auth_http/test/emqx_authn_scram_restapi_SUITE.erl Normal file
View File

@ -0,0 +1,509 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
-module(emqx_authn_scram_restapi_SUITE).
-compile(export_all).
-compile(nowarn_export_all).
-include_lib("eunit/include/eunit.hrl").
-include_lib("common_test/include/ct.hrl").
-include_lib("emqx/include/emqx_mqtt.hrl").
-include_lib("emqx_auth/include/emqx_authn.hrl").
-define(PATH, [authentication]).
-define(HTTP_PORT, 34333).
-define(HTTP_PATH, "/user/[...]").
-define(ALGORITHM, sha512).
-define(ALGORITHM_STR, <<"sha512">>).
-define(ITERATION_COUNT, 4096).
-define(T_ACL_USERNAME, <<"username">>).
-define(T_ACL_PASSWORD, <<"password">>).
-include_lib("emqx/include/emqx_placeholder.hrl").
all() ->
case emqx_release:edition() of
ce ->
[];
_ ->
emqx_common_test_helpers:all(?MODULE)
end.
init_per_suite(Config) ->
Apps = emqx_cth_suite:start([cowboy, emqx, emqx_conf, emqx_auth, emqx_auth_http], #{
work_dir => ?config(priv_dir, Config)
}),
IdleTimeout = emqx_config:get([mqtt, idle_timeout]),
[{apps, Apps}, {idle_timeout, IdleTimeout} | Config].
end_per_suite(Config) ->
ok = emqx_config:put([mqtt, idle_timeout], ?config(idle_timeout, Config)),
emqx_authn_test_lib:delete_authenticators(
[authentication],
?GLOBAL
),
ok = emqx_cth_suite:stop(?config(apps, Config)),
ok.
init_per_testcase(_Case, Config) ->
{ok, _} = emqx_cluster_rpc:start_link(node(), emqx_cluster_rpc, 1000),
emqx_authn_test_lib:delete_authenticators(
[authentication],
?GLOBAL
),
{ok, _} = emqx_authn_scram_restapi_test_server:start_link(?HTTP_PORT, ?HTTP_PATH),
Config.
end_per_testcase(_Case, _Config) ->
ok = emqx_authn_scram_restapi_test_server:stop().
%%------------------------------------------------------------------------------
%% Tests
%%------------------------------------------------------------------------------
t_create(_Config) ->
AuthConfig = raw_config(),
{ok, _} = emqx:update_config(
?PATH,
{create_authenticator, ?GLOBAL, AuthConfig}
),
{ok, [#{provider := emqx_authn_scram_restapi}]} = emqx_authn_chains:list_authenticators(
?GLOBAL
).
t_create_invalid(_Config) ->
AuthConfig = raw_config(),
InvalidConfigs =
[
AuthConfig#{<<"headers">> => []},
AuthConfig#{<<"method">> => <<"delete">>},
AuthConfig#{<<"url">> => <<"localhost">>},
AuthConfig#{<<"url">> => <<"http://foo.com/xxx#fragment">>},
AuthConfig#{<<"url">> => <<"http://${foo}.com/xxx">>},
AuthConfig#{<<"url">> => <<"//foo.com/xxx">>},
AuthConfig#{<<"algorithm">> => <<"sha128">>}
],
lists:foreach(
fun(Config) ->
ct:pal("creating authenticator with invalid config: ~p", [Config]),
{error, _} =
try
emqx:update_config(
?PATH,
{create_authenticator, ?GLOBAL, Config}
)
catch
throw:Error ->
{error, Error}
end,
?assertEqual(
{error, {not_found, {chain, ?GLOBAL}}},
emqx_authn_chains:list_authenticators(?GLOBAL)
)
end,
InvalidConfigs
).
t_authenticate(_Config) ->
Username = <<"u">>,
Password = <<"p">>,
set_user_handler(Username, Password),
init_auth(),
ok = emqx_config:put([mqtt, idle_timeout], 500),
{ok, Pid} = create_connection(Username, Password),
emqx_authn_mqtt_test_client:stop(Pid).
t_authenticate_bad_props(_Config) ->
Username = <<"u">>,
Password = <<"p">>,
set_user_handler(Username, Password),
init_auth(),
{ok, Pid} = emqx_authn_mqtt_test_client:start_link("127.0.0.1", 1883),
ConnectPacket = ?CONNECT_PACKET(
#mqtt_packet_connect{
proto_ver = ?MQTT_PROTO_V5,
properties = #{
'Authentication-Method' => <<"SCRAM-SHA-512">>
}
}
),
ok = emqx_authn_mqtt_test_client:send(Pid, ConnectPacket),
?CONNACK_PACKET(?RC_NOT_AUTHORIZED) = receive_packet().
t_authenticate_bad_username(_Config) ->
Username = <<"u">>,
Password = <<"p">>,
set_user_handler(Username, Password),
init_auth(),
{ok, Pid} = emqx_authn_mqtt_test_client:start_link("127.0.0.1", 1883),
ClientFirstMessage = esasl_scram:client_first_message(<<"badusername">>),
ConnectPacket = ?CONNECT_PACKET(
#mqtt_packet_connect{
proto_ver = ?MQTT_PROTO_V5,
properties = #{
'Authentication-Method' => <<"SCRAM-SHA-512">>,
'Authentication-Data' => ClientFirstMessage
}
}
),
ok = emqx_authn_mqtt_test_client:send(Pid, ConnectPacket),
?CONNACK_PACKET(?RC_NOT_AUTHORIZED) = receive_packet().
t_authenticate_bad_password(_Config) ->
Username = <<"u">>,
Password = <<"p">>,
set_user_handler(Username, Password),
init_auth(),
{ok, Pid} = emqx_authn_mqtt_test_client:start_link("127.0.0.1", 1883),
ClientFirstMessage = esasl_scram:client_first_message(Username),
ConnectPacket = ?CONNECT_PACKET(
#mqtt_packet_connect{
proto_ver = ?MQTT_PROTO_V5,
properties = #{
'Authentication-Method' => <<"SCRAM-SHA-512">>,
'Authentication-Data' => ClientFirstMessage
}
}
),
ok = emqx_authn_mqtt_test_client:send(Pid, ConnectPacket),
?AUTH_PACKET(
?RC_CONTINUE_AUTHENTICATION,
#{'Authentication-Data' := ServerFirstMessage}
) = receive_packet(),
{continue, ClientFinalMessage, _ClientCache} =
esasl_scram:check_server_first_message(
ServerFirstMessage,
#{
client_first_message => ClientFirstMessage,
password => <<"badpassword">>,
algorithm => ?ALGORITHM
}
),
AuthContinuePacket = ?AUTH_PACKET(
?RC_CONTINUE_AUTHENTICATION,
#{
'Authentication-Method' => <<"SCRAM-SHA-512">>,
'Authentication-Data' => ClientFinalMessage
}
),
ok = emqx_authn_mqtt_test_client:send(Pid, AuthContinuePacket),
?CONNACK_PACKET(?RC_NOT_AUTHORIZED) = receive_packet().
t_destroy(_Config) ->
Username = <<"u">>,
Password = <<"p">>,
set_user_handler(Username, Password),
init_auth(),
ok = emqx_config:put([mqtt, idle_timeout], 500),
{ok, Pid} = emqx_authn_mqtt_test_client:start_link("127.0.0.1", 1883),
ConnectPacket = ?CONNECT_PACKET(
#mqtt_packet_connect{
proto_ver = ?MQTT_PROTO_V5,
properties = #{
'Authentication-Method' => <<"SCRAM-SHA-512">>
}
}
),
ok = emqx_authn_mqtt_test_client:send(Pid, ConnectPacket),
ok = ct:sleep(1000),
?CONNACK_PACKET(?RC_NOT_AUTHORIZED) = receive_packet(),
%% emqx_authn_mqtt_test_client:stop(Pid),
emqx_authn_test_lib:delete_authenticators(
[authentication],
?GLOBAL
),
{ok, Pid2} = emqx_authn_mqtt_test_client:start_link("127.0.0.1", 1883),
ok = emqx_authn_mqtt_test_client:send(Pid2, ConnectPacket),
ok = ct:sleep(1000),
?CONNACK_PACKET(
?RC_SUCCESS,
_,
_
) = receive_packet().
t_acl(_Config) ->
init_auth(),
ACL = emqx_authn_http_SUITE:acl_rules(),
set_user_handler(?T_ACL_USERNAME, ?T_ACL_PASSWORD, #{acl => ACL}),
{ok, Pid} = create_connection(?T_ACL_USERNAME, ?T_ACL_PASSWORD),
Cases = [
{allow, <<"http-authn-acl/#">>},
{deny, <<"http-authn-acl/1">>},
{deny, <<"t/#">>}
],
try
lists:foreach(
fun(Case) ->
test_acl(Case, Pid)
end,
Cases
)
after
ok = emqx_authn_mqtt_test_client:stop(Pid)
end.
t_auth_expire(_Config) ->
init_auth(),
ExpireSec = 3,
WaitTime = timer:seconds(ExpireSec + 1),
ACL = emqx_authn_http_SUITE:acl_rules(),
set_user_handler(?T_ACL_USERNAME, ?T_ACL_PASSWORD, #{
acl => ACL,
expire_at =>
erlang:system_time(second) + ExpireSec
}),
{ok, Pid} = create_connection(?T_ACL_USERNAME, ?T_ACL_PASSWORD),
timer:sleep(WaitTime),
?assertEqual(false, erlang:is_process_alive(Pid)).
t_is_superuser() ->
State = init_auth(),
ok = test_is_superuser(State, false),
ok = test_is_superuser(State, true),
ok = test_is_superuser(State, false).
test_is_superuser(State, ExpectedIsSuperuser) ->
Username = <<"u">>,
Password = <<"p">>,
set_user_handler(Username, Password, #{is_superuser => ExpectedIsSuperuser}),
ClientFirstMessage = esasl_scram:client_first_message(Username),
{continue, ServerFirstMessage, ServerCache} =
emqx_authn_scram_restapi:authenticate(
#{
auth_method => <<"SCRAM-SHA-512">>,
auth_data => ClientFirstMessage,
auth_cache => #{}
},
State
),
{continue, ClientFinalMessage, ClientCache} =
esasl_scram:check_server_first_message(
ServerFirstMessage,
#{
client_first_message => ClientFirstMessage,
password => Password,
algorithm => ?ALGORITHM
}
),
{ok, UserInfo1, ServerFinalMessage} =
emqx_authn_scram_restapi:authenticate(
#{
auth_method => <<"SCRAM-SHA-512">>,
auth_data => ClientFinalMessage,
auth_cache => ServerCache
},
State
),
ok = esasl_scram:check_server_final_message(
ServerFinalMessage, ClientCache#{algorithm => ?ALGORITHM}
),
?assertMatch(#{is_superuser := ExpectedIsSuperuser}, UserInfo1).
%%------------------------------------------------------------------------------
%% Helpers
%%------------------------------------------------------------------------------
raw_config() ->
#{
<<"mechanism">> => <<"scram">>,
<<"backend">> => <<"http">>,
<<"enable">> => <<"true">>,
<<"method">> => <<"get">>,
<<"url">> => <<"http://127.0.0.1:34333/user">>,
<<"body">> => #{<<"username">> => ?PH_USERNAME},
<<"headers">> => #{<<"X-Test-Header">> => <<"Test Value">>},
<<"algorithm">> => ?ALGORITHM_STR,
<<"iteration_count">> => ?ITERATION_COUNT
}.
set_user_handler(Username, Password) ->
set_user_handler(Username, Password, #{is_superuser => false}).
set_user_handler(Username, Password, Extra0) ->
%% HTTP Server
Handler = fun(Req0, State) ->
#{
username := Username
} = cowboy_req:match_qs([username], Req0),
UserInfo = make_user_info(Password, ?ALGORITHM, ?ITERATION_COUNT),
Extra = maps:merge(#{is_superuser => false}, Extra0),
Req = cowboy_req:reply(
200,
#{<<"content-type">> => <<"application/json">>},
emqx_utils_json:encode(maps:merge(Extra, UserInfo)),
Req0
),
{ok, Req, State}
end,
ok = emqx_authn_scram_restapi_test_server:set_handler(Handler).
init_auth() ->
init_auth(raw_config()).
init_auth(Config) ->
{ok, _} = emqx:update_config(
?PATH,
{create_authenticator, ?GLOBAL, Config}
),
{ok, [#{state := State}]} = emqx_authn_chains:list_authenticators(?GLOBAL),
State.
make_user_info(Password, Algorithm, IterationCount) ->
{StoredKey, ServerKey, Salt} = esasl_scram:generate_authentication_info(
Password,
#{
algorithm => Algorithm,
iteration_count => IterationCount
}
),
#{
stored_key => binary:encode_hex(StoredKey),
server_key => binary:encode_hex(ServerKey),
salt => binary:encode_hex(Salt)
}.
receive_packet() ->
receive
{packet, Packet} ->
ct:pal("Delivered packet: ~p", [Packet]),
Packet
after 1000 ->
ct:fail("Deliver timeout")
end.
create_connection(Username, Password) ->
{ok, Pid} = emqx_authn_mqtt_test_client:start_link("127.0.0.1", 1883),
ClientFirstMessage = esasl_scram:client_first_message(Username),
ConnectPacket = ?CONNECT_PACKET(
#mqtt_packet_connect{
proto_ver = ?MQTT_PROTO_V5,
properties = #{
'Authentication-Method' => <<"SCRAM-SHA-512">>,
'Authentication-Data' => ClientFirstMessage
}
}
),
ok = emqx_authn_mqtt_test_client:send(Pid, ConnectPacket),
%% Intentional sleep to trigger idle timeout for the connection not yet authenticated
ok = ct:sleep(1000),
?AUTH_PACKET(
?RC_CONTINUE_AUTHENTICATION,
#{'Authentication-Data' := ServerFirstMessage}
) = receive_packet(),
{continue, ClientFinalMessage, ClientCache} =
esasl_scram:check_server_first_message(
ServerFirstMessage,
#{
client_first_message => ClientFirstMessage,
password => Password,
algorithm => ?ALGORITHM
}
),
AuthContinuePacket = ?AUTH_PACKET(
?RC_CONTINUE_AUTHENTICATION,
#{
'Authentication-Method' => <<"SCRAM-SHA-512">>,
'Authentication-Data' => ClientFinalMessage
}
),
ok = emqx_authn_mqtt_test_client:send(Pid, AuthContinuePacket),
?CONNACK_PACKET(
?RC_SUCCESS,
_,
#{'Authentication-Data' := ServerFinalMessage}
) = receive_packet(),
ok = esasl_scram:check_server_final_message(
ServerFinalMessage, ClientCache#{algorithm => ?ALGORITHM}
),
{ok, Pid}.
test_acl({allow, Topic}, C) ->
?assertMatch(
[0],
send_subscribe(C, Topic)
);
test_acl({deny, Topic}, C) ->
?assertMatch(
[?RC_NOT_AUTHORIZED],
send_subscribe(C, Topic)
).
send_subscribe(Client, Topic) ->
TopicOpts = #{nl => 0, rap => 0, rh => 0, qos => 0},
Packet = ?SUBSCRIBE_PACKET(1, [{Topic, TopicOpts}]),
emqx_authn_mqtt_test_client:send(Client, Packet),
timer:sleep(200),
?SUBACK_PACKET(1, ReasonCode) = receive_packet(),
ReasonCode.

115
apps/emqx_auth_http/test/emqx_authn_scram_restapi_test_server.erl Normal file
View File

@ -0,0 +1,115 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
-module(emqx_authn_scram_restapi_test_server).
-behaviour(supervisor).
-behaviour(cowboy_handler).
% cowboy_server callbacks
-export([init/2]).
% supervisor callbacks
-export([init/1]).
% API
-export([
start_link/2,
start_link/3,
stop/0,
set_handler/1
]).
%%------------------------------------------------------------------------------
%% API
%%------------------------------------------------------------------------------
start_link(Port, Path) ->
start_link(Port, Path, false).
start_link(Port, Path, SSLOpts) ->
supervisor:start_link({local, ?MODULE}, ?MODULE, [Port, Path, SSLOpts]).
stop() ->
gen_server:stop(?MODULE).
set_handler(F) when is_function(F, 2) ->
true = ets:insert(?MODULE, {handler, F}),
ok.
%%------------------------------------------------------------------------------
%% supervisor API
%%------------------------------------------------------------------------------
init([Port, Path, SSLOpts]) ->
Dispatch = cowboy_router:compile(
[
{'_', [{Path, ?MODULE, []}]}
]
),
ProtoOpts = #{env => #{dispatch => Dispatch}},
Tab = ets:new(?MODULE, [set, named_table, public]),
ets:insert(Tab, {handler, fun default_handler/2}),
{Transport, TransOpts, CowboyModule} = transport_settings(Port, SSLOpts),
ChildSpec = ranch:child_spec(?MODULE, Transport, TransOpts, CowboyModule, ProtoOpts),
{ok, {#{}, [ChildSpec]}}.
%%------------------------------------------------------------------------------
%% cowboy_server API
%%------------------------------------------------------------------------------
init(Req, State) ->
[{handler, Handler}] = ets:lookup(?MODULE, handler),
Handler(Req, State).
%%------------------------------------------------------------------------------
%% Internal functions
%%------------------------------------------------------------------------------
transport_settings(Port, false) ->
TransOpts = #{
socket_opts => [{port, Port}],
connection_type => supervisor
},
{ranch_tcp, TransOpts, cowboy_clear};
transport_settings(Port, SSLOpts) ->
TransOpts = #{
socket_opts => [
{port, Port},
{next_protocols_advertised, [<<"h2">>, <<"http/1.1">>]},
{alpn_preferred_protocols, [<<"h2">>, <<"http/1.1">>]}
| SSLOpts
],
connection_type => supervisor
},
{ranch_ssl, TransOpts, cowboy_tls}.
default_handler(Req0, State) ->
Req = cowboy_req:reply(
400,
#{<<"content-type">> => <<"text/plain">>},
<<"">>,
Req0
),
{ok, Req, State}.
make_user_info(Password, Algorithm, IterationCount) ->
{StoredKey, ServerKey, Salt} = esasl_scram:generate_authentication_info(
Password,
#{
algorithm => Algorithm,
iteration_count => IterationCount
}
),
#{
stored_key => StoredKey,
server_key => ServerKey,
salt => Salt,
is_superuser => false
}.

3
apps/emqx_auth_jwt/src/emqx_authn_jwks_connector.erl
View File

@ -22,6 +22,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -32,6 +33,8 @@
-define(DEFAULT_POOL_SIZE, 8).
resource_type() -> jwks.
callback_mode() -> always_sync.
on_start(InstId, Opts) ->

3
apps/emqx_auth_jwt/src/emqx_authn_jwt.erl
View File

@ -188,7 +188,8 @@ do_create(
ResourceId,
?AUTHN_RESOURCE_GROUP,
emqx_authn_jwks_connector,
connector_opts(Config)
connector_opts(Config),
#{}
),
{ok, #{
jwk_resource => ResourceId,

2
apps/emqx_auth_ldap/src/emqx_auth_ldap.app.src
View File

@ -1,7 +1,7 @@
%% -*- mode: erlang -*-
{application, emqx_auth_ldap, [
{description, "EMQX LDAP Authentication and Authorization"},
{vsn, "0.1.2"},
{vsn, "0.1.3"},
{registered, []},
{mod, {emqx_auth_ldap_app, []}},
{applications, [

93
apps/emqx_auth_ldap/test/emqx_authn_ldap_SUITE.erl
View File

@ -21,6 +21,7 @@
-include_lib("emqx_auth/include/emqx_authn.hrl").
-include_lib("eunit/include/eunit.hrl").
-include_lib("common_test/include/ct.hrl").
-include_lib("snabbkaffe/include/snabbkaffe.hrl").
-define(LDAP_HOST, "ldap").
-define(LDAP_DEFAULT_PORT, 389).
@ -46,13 +47,6 @@ init_per_suite(Config) ->
Apps = emqx_cth_suite:start([emqx, emqx_conf, emqx_auth, emqx_auth_ldap], #{
work_dir => ?config(priv_dir, Config)
}),
{ok, _} = emqx_resource:create_local(
?LDAP_RESOURCE,
?AUTHN_RESOURCE_GROUP,
emqx_ldap,
ldap_config(),
#{}
),
[{apps, Apps} | Config];
false ->
{skip, no_ldap}
@ -63,7 +57,6 @@ end_per_suite(Config) ->
[authentication],
?GLOBAL
),
ok = emqx_resource:remove_local(?LDAP_RESOURCE),
ok = emqx_cth_suite:stop(?config(apps, Config)).
%%------------------------------------------------------------------------------
@ -128,6 +121,87 @@ t_create_invalid(_Config) ->
InvalidConfigs
).
t_authenticate_timeout_cause_reconnect(_Config) ->
TestPid = self(),
meck:new(eldap, [non_strict, no_link, passthrough]),
try
%% cause eldap process to be killed
meck:expect(
eldap,
search,
fun
(Pid, [{base, <<"uid=mqttuser0007", _/binary>>} | _]) ->
TestPid ! {eldap_pid, Pid},
{error, {gen_tcp_error, timeout}};
(Pid, Args) ->
meck:passthrough([Pid, Args])
end
),
Credentials = fun(Username) ->
#{
username => Username,
password => Username,
listener => 'tcp:default',
protocol => mqtt
}
end,
SpecificConfigParams = #{},
Result = {ok, #{is_superuser => true}},
Timeout = 1000,
Config0 = raw_ldap_auth_config(),
Config = Config0#{
<<"pool_size">> => 1,
<<"request_timeout">> => Timeout
},
AuthConfig = maps:merge(Config, SpecificConfigParams),
{ok, _} = emqx:update_config(
?PATH,
{create_authenticator, ?GLOBAL, AuthConfig}
),
%% 0006 is a disabled user
?assertEqual(
{error, user_disabled},
emqx_access_control:authenticate(Credentials(<<"mqttuser0006">>))
),
?assertEqual(
{error, not_authorized},
emqx_access_control:authenticate(Credentials(<<"mqttuser0007">>))
),
ok = wait_for_ldap_pid(1000),
[#{id := ResourceID}] = emqx_resource_manager:list_all(),
?retry(1_000, 10, {ok, connected} = emqx_resource_manager:health_check(ResourceID)),
%% turn back to normal
meck:expect(
eldap,
search,
2,
fun(Pid2, Query) ->
meck:passthrough([Pid2, Query])
end
),
%% expect eldap process to be restarted
?assertEqual(Result, emqx_access_control:authenticate(Credentials(<<"mqttuser0007">>))),
emqx_authn_test_lib:delete_authenticators(
[authentication],
?GLOBAL
)
after
meck:unload(eldap)
end.
wait_for_ldap_pid(After) ->
receive
{eldap_pid, Pid} ->
?assertNot(is_process_alive(Pid)),
ok
after After ->
error(timeout)
end.
t_authenticate(_Config) ->
ok = lists:foreach(
fun(Sample) ->
@ -300,6 +374,3 @@ user_seeds() ->
ldap_server() ->
iolist_to_binary(io_lib:format("~s:~B", [?LDAP_HOST, ?LDAP_DEFAULT_PORT])).
ldap_config() ->
emqx_ldap_SUITE:ldap_config([]).

14
apps/emqx_auth_ldap/test/emqx_authz_ldap_SUITE.erl
View File

@ -44,7 +44,6 @@ init_per_suite(Config) ->
],
#{work_dir => emqx_cth_suite:work_dir(Config)}
),
ok = create_ldap_resource(),
[{apps, Apps} | Config];
false ->
{skip, no_ldap}
@ -167,21 +166,8 @@ setup_config(SpecialParams) ->
ldap_server() ->
iolist_to_binary(io_lib:format("~s:~B", [?LDAP_HOST, ?LDAP_DEFAULT_PORT])).
ldap_config() ->
emqx_ldap_SUITE:ldap_config([]).
start_apps(Apps) ->
lists:foreach(fun application:ensure_all_started/1, Apps).
stop_apps(Apps) ->
lists:foreach(fun application:stop/1, Apps).
create_ldap_resource() ->
{ok, _} = emqx_resource:create_local(
?LDAP_RESOURCE,
?AUTHZ_RESOURCE_GROUP,
emqx_ldap,
ldap_config(),
#{}
),
ok.

71
apps/emqx_auth_mnesia/src/emqx_authn_scram_mnesia.erl
View File

@ -133,17 +133,17 @@ authenticate(
},
State
) ->
case ensure_auth_method(AuthMethod, AuthData, State) of
true ->
case AuthCache of
#{next_step := client_final} ->
check_client_final_message(AuthData, AuthCache, State);
_ ->
check_client_first_message(AuthData, AuthCache, State)
end;
false ->
ignore
end;
RetrieveFun = fun(Username) ->
retrieve(Username, State)
end,
OnErrFun = fun(Msg, Reason) ->
?TRACE_AUTHN_PROVIDER(Msg, #{
reason => Reason
})
end,
emqx_utils_scram:authenticate(
AuthMethod, AuthData, AuthCache, State, RetrieveFun, OnErrFun, [is_superuser]
);
authenticate(_Credential, _State) ->
ignore.
@ -257,55 +257,6 @@ run_fuzzy_filter(
%% Internal functions
%%------------------------------------------------------------------------------
ensure_auth_method(_AuthMethod, undefined, _State) ->
false;
ensure_auth_method(<<"SCRAM-SHA-256">>, _AuthData, #{algorithm := sha256}) ->
true;
ensure_auth_method(<<"SCRAM-SHA-512">>, _AuthData, #{algorithm := sha512}) ->
true;
ensure_auth_method(_AuthMethod, _AuthData, _State) ->
false.
check_client_first_message(Bin, _Cache, #{iteration_count := IterationCount} = State) ->
RetrieveFun = fun(Username) ->
retrieve(Username, State)
end,
case
esasl_scram:check_client_first_message(
Bin,
#{
iteration_count => IterationCount,
retrieve => RetrieveFun
}
)
of
{continue, ServerFirstMessage, Cache} ->
{continue, ServerFirstMessage, Cache};
ignore ->
ignore;
{error, Reason} ->
?TRACE_AUTHN_PROVIDER("check_client_first_message_error", #{
reason => Reason
}),
{error, not_authorized}
end.
check_client_final_message(Bin, #{is_superuser := IsSuperuser} = Cache, #{algorithm := Alg}) ->
case
esasl_scram:check_client_final_message(
Bin,
Cache#{algorithm => Alg}
)
of
{ok, ServerFinalMessage} ->
{ok, #{is_superuser => IsSuperuser}, ServerFinalMessage};
{error, Reason} ->
?TRACE_AUTHN_PROVIDER("check_client_final_message_error", #{
reason => Reason
}),
{error, not_authorized}
end.
user_info_record(
#{
user_id := UserID,

7
apps/emqx_auth_mnesia/src/emqx_authn_scram_mnesia_schema.erl
View File

@ -29,6 +29,8 @@
select_union_member/1
]).
-export([algorithm/1, iteration_count/1]).
namespace() -> "authn".
refs() ->
@ -38,11 +40,6 @@ select_union_member(#{
<<"mechanism">> := ?AUTHN_MECHANISM_SCRAM_BIN, <<"backend">> := ?AUTHN_BACKEND_BIN
}) ->
refs();
select_union_member(#{<<"mechanism">> := ?AUTHN_MECHANISM_SCRAM_BIN}) ->
throw(#{
reason => "unknown_backend",
expected => ?AUTHN_BACKEND
});
select_union_member(_) ->
undefined.

20
apps/emqx_auth_mysql/src/emqx_authz_mysql.erl
View File

@ -101,19 +101,9 @@ authorize(
do_authorize(_Client, _Action, _Topic, _ColumnNames, []) ->
nomatch;
do_authorize(Client, Action, Topic, ColumnNames, [Row | Tail]) ->
try
emqx_authz_rule:match(
Client, Action, Topic, emqx_authz_utils:parse_rule_from_row(ColumnNames, Row)
)
of
{matched, Permission} -> {matched, Permission};
nomatch -> do_authorize(Client, Action, Topic, ColumnNames, Tail)
catch
error:Reason ->
?SLOG(error, #{
msg => "match_rule_error",
reason => Reason,
rule => Row
}),
do_authorize(Client, Action, Topic, ColumnNames, Tail)
case emqx_authz_utils:do_authorize(mysql, Client, Action, Topic, ColumnNames, Row) of
nomatch ->
do_authorize(Client, Action, Topic, ColumnNames, Tail);
{matched, Permission} ->
{matched, Permission}
end.

21
apps/emqx_auth_postgresql/src/emqx_authz_postgresql.erl
View File

@ -107,22 +107,11 @@ authorize(
do_authorize(_Client, _Action, _Topic, _ColumnNames, []) ->
nomatch;
do_authorize(Client, Action, Topic, ColumnNames, [Row | Tail]) ->
try
emqx_authz_rule:match(
Client, Action, Topic, emqx_authz_utils:parse_rule_from_row(ColumnNames, Row)
)
of
{matched, Permission} -> {matched, Permission};
nomatch -> do_authorize(Client, Action, Topic, ColumnNames, Tail)
catch
error:Reason:Stack ->
?SLOG(error, #{
msg => "match_rule_error",
reason => Reason,
rule => Row,
stack => Stack
}),
do_authorize(Client, Action, Topic, ColumnNames, Tail)
case emqx_authz_utils:do_authorize(postgresql, Client, Action, Topic, ColumnNames, Row) of
nomatch ->
do_authorize(Client, Action, Topic, ColumnNames, Tail);
{matched, Permission} ->
{matched, Permission}
end.
column_names(Columns) ->

4
apps/emqx_auth_postgresql/test/emqx_authn_postgresql_SUITE.erl
View File

@ -198,9 +198,9 @@ test_user_auth(#{
t_authenticate_disabled_prepared_statements(_Config) ->
ResConfig = maps:merge(pgsql_config(), #{disable_prepared_statements => true}),
{ok, _} = emqx_resource:recreate_local(?PGSQL_RESOURCE, emqx_postgresql, ResConfig),
{ok, _} = emqx_resource:recreate_local(?PGSQL_RESOURCE, emqx_postgresql, ResConfig, #{}),
on_exit(fun() ->
emqx_resource:recreate_local(?PGSQL_RESOURCE, emqx_postgresql, pgsql_config())
emqx_resource:recreate_local(?PGSQL_RESOURCE, emqx_postgresql, pgsql_config(), #{})
end),
ok = lists:foreach(
fun(Sample0) ->

58
apps/emqx_auth_redis/src/emqx_authz_redis.erl
View File

@ -92,42 +92,28 @@ authorize(
do_authorize(_Client, _Action, _Topic, []) ->
nomatch;
do_authorize(Client, Action, Topic, [TopicFilterRaw, RuleEncoded | Tail]) ->
try
emqx_authz_rule:match(
Client,
Action,
Topic,
compile_rule(RuleEncoded, TopicFilterRaw)
)
of
{matched, Permission} -> {matched, Permission};
nomatch -> do_authorize(Client, Action, Topic, Tail)
catch
error:Reason:Stack ->
?SLOG(error, #{
msg => "match_rule_error",
reason => Reason,
rule_encoded => RuleEncoded,
topic_filter_raw => TopicFilterRaw,
stacktrace => Stack
}),
do_authorize(Client, Action, Topic, Tail)
end.
compile_rule(RuleBin, TopicFilterRaw) ->
RuleRaw =
case parse_rule(RuleEncoded) of
{ok, RuleMap0} ->
RuleMap =
maps:merge(
#{
<<"permission">> => <<"allow">>,
<<"topic">> => TopicFilterRaw
},
parse_rule(RuleBin)
RuleMap0
),
case emqx_authz_rule_raw:parse_rule(RuleRaw) of
{ok, {Permission, Action, Topics}} ->
emqx_authz_rule:compile({Permission, all, Action, Topics});
case emqx_authz_utils:do_authorize(redis, Client, Action, Topic, undefined, RuleMap) of
nomatch ->
do_authorize(Client, Action, Topic, Tail);
{matched, Permission} ->
{matched, Permission}
end;
{error, Reason} ->
error(Reason)
?SLOG(error, Reason#{
msg => "parse_rule_error",
rule => RuleEncoded
}),
do_authorize(Client, Action, Topic, Tail)
end.
parse_cmd(Query) ->
@ -154,17 +140,17 @@ validate_cmd(Cmd) ->
end.
parse_rule(<<"publish">>) ->
#{<<"action">> => <<"publish">>};
{ok, #{<<"action">> => <<"publish">>}};
parse_rule(<<"subscribe">>) ->
#{<<"action">> => <<"subscribe">>};
{ok, #{<<"action">> => <<"subscribe">>}};
parse_rule(<<"all">>) ->
#{<<"action">> => <<"all">>};
{ok, #{<<"action">> => <<"all">>}};
parse_rule(Bin) when is_binary(Bin) ->
case emqx_utils_json:safe_decode(Bin, [return_maps]) of
{ok, Map} when is_map(Map) ->
maps:with([<<"qos">>, <<"action">>, <<"retain">>], Map);
{ok, maps:with([<<"qos">>, <<"action">>, <<"retain">>], Map)};
{ok, _} ->
error({invalid_topic_rule, Bin, notamap});
{error, Error} ->
error({invalid_topic_rule, Bin, Error})
{error, #{reason => invalid_topic_rule_not_map, value => Bin}};
{error, _Error} ->
{error, #{reason => invalid_topic_rule_not_json, value => Bin}}
end.

2
apps/emqx_bridge/src/emqx_bridge_resource.erl
View File

@ -198,7 +198,7 @@ create(Type, Name, Conf0, Opts) ->
Conf = Conf0#{bridge_type => TypeBin, bridge_name => Name},
{ok, _Data} = emqx_resource:create_local(
resource_id(Type, Name),
<<"emqx_bridge">>,
<<"bridge">>,
bridge_to_resource_type(Type),
parse_confs(TypeBin, Name, Conf),
parse_opts(Conf, Opts)

1
apps/emqx_bridge/src/schema/emqx_bridge_schema.erl
View File

@ -123,6 +123,7 @@ common_bridge_fields() ->
boolean(),
#{
desc => ?DESC("desc_enable"),
importance => ?IMPORTANCE_NO_DOC,
default => true
}
)},

17
apps/emqx_bridge/src/schema/emqx_bridge_v2_schema.erl
View File

@ -65,6 +65,7 @@
-export([
make_producer_action_schema/1, make_producer_action_schema/2,
make_consumer_action_schema/1, make_consumer_action_schema/2,
common_fields/0,
top_level_common_action_keys/0,
top_level_common_source_keys/0,
project_to_actions_resource_opts/1,
@ -507,16 +508,26 @@ make_consumer_action_schema(ParametersRef, Opts) ->
})}
].
common_schema(ParametersRef, _Opts) ->
common_fields() ->
[
{enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
{enable,
mk(boolean(), #{
desc => ?DESC("config_enable"),
importance => ?IMPORTANCE_NO_DOC,
default => true
})},
{connector,
mk(binary(), #{
desc => ?DESC(emqx_connector_schema, "connector_field"), required => true
})},
{tags, emqx_schema:tags_schema()},
{description, emqx_schema:description_schema()},
{description, emqx_schema:description_schema()}
].
common_schema(ParametersRef, _Opts) ->
[
{parameters, ParametersRef}
| common_fields()
].
project_to_actions_resource_opts(OldResourceOpts) ->

1
apps/emqx_bridge/test/emqx_bridge_v2_SUITE.erl
View File

@ -1110,6 +1110,7 @@ t_query_uses_action_query_mode(_Config) ->
%% ... now we use a quite different query mode for the action
meck:expect(con_mod(), query_mode, 1, simple_async_internal_buffer),
meck:expect(con_mod(), resource_type, 0, dummy),
meck:expect(con_mod(), callback_mode, 0, async_if_possible),
{ok, _} = emqx_bridge_v2:create(bridge_type(), ActionName, ActionConfig),

1
apps/emqx_bridge/test/emqx_bridge_v2_api_SUITE.erl
View File

@ -302,6 +302,7 @@ init_mocks() ->
meck:new(emqx_connector_resource, [passthrough, no_link]),
meck:expect(emqx_connector_resource, connector_to_resource_type, 1, ?CONNECTOR_IMPL),
meck:new(?CONNECTOR_IMPL, [non_strict, no_link]),
meck:expect(?CONNECTOR_IMPL, resource_type, 0, dummy),
meck:expect(?CONNECTOR_IMPL, callback_mode, 0, async_if_possible),
meck:expect(
?CONNECTOR_IMPL,

4
apps/emqx_bridge/test/emqx_bridge_v2_dummy_connector.erl
View File

@ -15,15 +15,17 @@
%% this module is only intended to be mocked
-module(emqx_bridge_v2_dummy_connector).
-behavior(emqx_resource).
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
on_add_channel/4,
on_get_channel_status/3
]).
resource_type() -> dummy.
callback_mode() -> error(unexpected).
on_start(_, _) -> error(unexpected).
on_stop(_, _) -> error(unexpected).

3
apps/emqx_bridge/test/emqx_bridge_v2_test_connector.erl
View File

@ -19,6 +19,7 @@
-export([
query_mode/1,
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -34,6 +35,8 @@
query_mode(_Config) ->
sync.
resource_type() -> test_connector.
callback_mode() ->
always_sync.

5
apps/emqx_bridge_azure_blob_storage/src/emqx_bridge_azure_blob_storage_connector.erl
View File

@ -18,6 +18,7 @@
%% `emqx_resource' API
-export([
callback_mode/0,
resource_type/0,
on_start/2,
on_stop/2,
@ -148,6 +149,10 @@
callback_mode() ->
always_sync.
-spec resource_type() -> atom().
resource_type() ->
azure_blob_storage.
-spec on_start(connector_resource_id(), connector_config()) ->
{ok, connector_state()} | {error, _Reason}.
on_start(_ConnResId, ConnConfig) ->

2
apps/emqx_bridge_azure_event_hub/mix.exs
View File

@ -23,7 +23,7 @@ defmodule EMQXBridgeAzureEventHub.MixProject do
def deps() do
[
{:wolff, github: "kafka4beam/wolff", tag: "2.0.0"},
{:wolff, github: "kafka4beam/wolff", tag: "3.0.2"},
{:kafka_protocol, github: "kafka4beam/kafka_protocol", tag: "4.1.5", override: true},
{:brod_gssapi, github: "kafka4beam/brod_gssapi", tag: "v0.1.1"},
{:brod, github: "kafka4beam/brod", tag: "3.18.0"},

2
apps/emqx_bridge_azure_event_hub/rebar.config
View File

@ -2,7 +2,7 @@
{erl_opts, [debug_info]}.
{deps, [
{wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "2.0.0"}}},
{wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "3.0.2"}}},
{kafka_protocol, {git, "https://github.com/kafka4beam/kafka_protocol.git", {tag, "4.1.5"}}},
{brod_gssapi, {git, "https://github.com/kafka4beam/brod_gssapi.git", {tag, "v0.1.1"}}},
{brod, {git, "https://github.com/kafka4beam/brod.git", {tag, "3.18.0"}}},

2
apps/emqx_bridge_azure_event_hub/src/emqx_bridge_azure_event_hub.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_azure_event_hub, [
{description, "EMQX Enterprise Azure Event Hub Bridge"},
{vsn, "0.1.7"},
{vsn, "0.1.8"},
{registered, []},
{applications, [
kernel,

11
apps/emqx_bridge_azure_event_hub/src/emqx_bridge_azure_event_hub.erl
View File

@ -129,16 +129,7 @@ fields(actions) ->
override(
emqx_bridge_kafka:producer_opts(action),
bridge_v2_overrides()
) ++
[
{enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
{connector,
mk(binary(), #{
desc => ?DESC(emqx_connector_schema, "connector_field"), required => true
})},
{tags, emqx_schema:tags_schema()},
{description, emqx_schema:description_schema()}
],
) ++ emqx_bridge_v2_schema:common_fields(),
override_documentations(Fields);
fields(Method) ->
Fields = emqx_bridge_kafka:fields(Method),

21
apps/emqx_bridge_azure_event_hub/test/emqx_bridge_azure_event_hub_v2_SUITE.erl
View File

@ -382,7 +382,26 @@ t_multiple_actions_sharing_topic(Config) ->
ActionConfig0,
#{<<"parameters">> => #{<<"query_mode">> => <<"sync">>}}
),
ok = emqx_bridge_v2_kafka_producer_SUITE:t_multiple_actions_sharing_topic(
ok =
emqx_bridge_v2_kafka_producer_SUITE:?FUNCTION_NAME(
[
{type, ?BRIDGE_TYPE_BIN},
{connector_name, ?config(connector_name, Config)},
{connector_config, ?config(connector_config, Config)},
{action_config, ActionConfig}
]
),
ok.
t_dynamic_topics(Config) ->
ActionConfig0 = ?config(action_config, Config),
ActionConfig =
emqx_utils_maps:deep_merge(
ActionConfig0,
#{<<"parameters">> => #{<<"query_mode">> => <<"sync">>}}
),
ok =
emqx_bridge_v2_kafka_producer_SUITE:?FUNCTION_NAME(
[
{type, ?BRIDGE_TYPE_BIN},
{connector_name, ?config(connector_name, Config)},

2
apps/emqx_bridge_cassandra/src/emqx_bridge_cassandra.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_cassandra, [
{description, "EMQX Enterprise Cassandra Bridge"},
{vsn, "0.3.1"},
{vsn, "0.3.2"},
{registered, []},
{applications, [
kernel,

2
apps/emqx_bridge_cassandra/src/emqx_bridge_cassandra_connector.erl
View File

@ -19,6 +19,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -94,6 +95,7 @@ desc("connector") ->
%%--------------------------------------------------------------------
%% callbacks for emqx_resource
resource_type() -> cassandra.
callback_mode() -> async_if_possible.

2
apps/emqx_bridge_clickhouse/src/emqx_bridge_clickhouse.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_clickhouse, [
{description, "EMQX Enterprise ClickHouse Bridge"},
{vsn, "0.4.1"},
{vsn, "0.4.2"},
{registered, []},
{applications, [
kernel,

2
apps/emqx_bridge_clickhouse/src/emqx_bridge_clickhouse_connector.erl
View File

@ -29,6 +29,7 @@
%% callbacks for behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -128,6 +129,7 @@ values(_) ->
%% ===================================================================
%% Callbacks defined in emqx_resource
%% ===================================================================
resource_type() -> clickhouse.
callback_mode() -> always_sync.

2
apps/emqx_bridge_confluent/mix.exs
View File

@ -23,7 +23,7 @@ defmodule EMQXBridgeConfluent.MixProject do
def deps() do
[
{:wolff, github: "kafka4beam/wolff", tag: "2.0.0"},
{:wolff, github: "kafka4beam/wolff", tag: "3.0.2"},
{:kafka_protocol, github: "kafka4beam/kafka_protocol", tag: "4.1.5", override: true},
{:brod_gssapi, github: "kafka4beam/brod_gssapi", tag: "v0.1.1"},
{:brod, github: "kafka4beam/brod", tag: "3.18.0"},

2
apps/emqx_bridge_confluent/rebar.config
View File

@ -2,7 +2,7 @@
{erl_opts, [debug_info]}.
{deps, [
{wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "2.0.0"}}},
{wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "3.0.2"}}},
{kafka_protocol, {git, "https://github.com/kafka4beam/kafka_protocol.git", {tag, "4.1.5"}}},
{brod_gssapi, {git, "https://github.com/kafka4beam/brod_gssapi.git", {tag, "v0.1.1"}}},
{brod, {git, "https://github.com/kafka4beam/brod.git", {tag, "3.18.0"}}},

2
apps/emqx_bridge_confluent/src/emqx_bridge_confluent.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_confluent, [
{description, "EMQX Enterprise Confluent Connector and Action"},
{vsn, "0.1.2"},
{vsn, "0.1.3"},
{registered, []},
{applications, [
kernel,

11
apps/emqx_bridge_confluent/src/emqx_bridge_confluent_producer.erl
View File

@ -116,16 +116,7 @@ fields(actions) ->
override(
emqx_bridge_kafka:producer_opts(action),
bridge_v2_overrides()
) ++
[
{enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
{connector,
mk(binary(), #{
desc => ?DESC(emqx_connector_schema, "connector_field"), required => true
})},
{tags, emqx_schema:tags_schema()},
{description, emqx_schema:description_schema()}
],
) ++ emqx_bridge_v2_schema:common_fields(),
override_documentations(Fields);
fields(Method) ->
Fields = emqx_bridge_kafka:fields(Method),

21
apps/emqx_bridge_confluent/test/emqx_bridge_confluent_producer_SUITE.erl
View File

@ -391,7 +391,26 @@ t_multiple_actions_sharing_topic(Config) ->
ActionConfig0,
#{<<"parameters">> => #{<<"query_mode">> => <<"sync">>}}
),
ok = emqx_bridge_v2_kafka_producer_SUITE:t_multiple_actions_sharing_topic(
ok =
emqx_bridge_v2_kafka_producer_SUITE:?FUNCTION_NAME(
[
{type, ?ACTION_TYPE_BIN},
{connector_name, ?config(connector_name, Config)},
{connector_config, ?config(connector_config, Config)},
{action_config, ActionConfig}
]
),
ok.
t_dynamic_topics(Config) ->
ActionConfig0 = ?config(action_config, Config),
ActionConfig =
emqx_utils_maps:deep_merge(
ActionConfig0,
#{<<"parameters">> => #{<<"query_mode">> => <<"sync">>}}
),
ok =
emqx_bridge_v2_kafka_producer_SUITE:?FUNCTION_NAME(
[
{type, ?ACTION_TYPE_BIN},
{connector_name, ?config(connector_name, Config)},

5
apps/emqx_bridge_couchbase/src/emqx_bridge_couchbase_connector.erl
View File

@ -15,6 +15,7 @@
%% `emqx_resource' API
-export([
callback_mode/0,
resource_type/0,
on_start/2,
on_stop/2,
@ -84,6 +85,10 @@
callback_mode() ->
always_sync.
-spec resource_type() -> atom().
resource_type() ->
couchbase.
-spec on_start(connector_resource_id(), connector_config()) ->
{ok, connector_state()} | {error, _Reason}.
on_start(ConnResId, ConnConfig) ->

2
apps/emqx_bridge_dynamo/src/emqx_bridge_dynamo.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_dynamo, [
{description, "EMQX Enterprise Dynamo Bridge"},
{vsn, "0.2.2"},
{vsn, "0.2.3"},
{registered, []},
{applications, [
kernel,

2
apps/emqx_bridge_dynamo/src/emqx_bridge_dynamo_connector.erl
View File

@ -17,6 +17,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -68,6 +69,7 @@ fields(config) ->
%%========================================================================================
%% `emqx_resource' API
%%========================================================================================
resource_type() -> dynamo.
callback_mode() -> always_sync.

2
apps/emqx_bridge_es/src/emqx_bridge_es.app.src
View File

@ -1,7 +1,7 @@
%% -*- mode: erlang -*-
{application, emqx_bridge_es, [
{description, "EMQX Enterprise Elastic Search Bridge"},
{vsn, "0.1.3"},
{vsn, "0.1.4"},
{modules, [
emqx_bridge_es,
emqx_bridge_es_connector

3
apps/emqx_bridge_es/src/emqx_bridge_es_connector.erl
View File

@ -14,6 +14,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -207,6 +208,8 @@ base_url(#{server := Server}) -> "http://" ++ Server.
%%-------------------------------------------------------------------------------------
%% `emqx_resource' API
%%-------------------------------------------------------------------------------------
resource_type() -> elastic_search.
callback_mode() -> async_if_possible.
-spec on_start(manager_id(), config()) -> {ok, state()} | no_return().

1
apps/emqx_bridge_gcp_pubsub/mix.exs
View File

@ -23,6 +23,7 @@ defmodule EMQXBridgeGcpPubsub.MixProject do
def deps() do
[
{:emqx_connector_jwt, in_umbrella: true},
{:emqx_connector, in_umbrella: true, runtime: false},
{:emqx_resource, in_umbrella: true},
{:emqx_bridge, in_umbrella: true, runtime: false},

1
apps/emqx_bridge_gcp_pubsub/rebar.config
View File

@ -9,6 +9,7 @@
debug_info
]}.
{deps, [
{emqx_connector_jwt, {path, "../../apps/emqx_connector_jwt"}},
{emqx_connector, {path, "../../apps/emqx_connector"}},
{emqx_resource, {path, "../../apps/emqx_resource"}},
{emqx_bridge, {path, "../../apps/emqx_bridge"}},

3
apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub.app.src
View File

@ -6,7 +6,8 @@
kernel,
stdlib,
emqx_resource,
ehttpc
ehttpc,
emqx_connector_jwt
]},
{env, [
{emqx_action_info_modules, [

2
apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub_client.erl
View File

@ -5,7 +5,7 @@
-module(emqx_bridge_gcp_pubsub_client).
-include_lib("jose/include/jose_jwk.hrl").
-include_lib("emqx_connector/include/emqx_connector_tables.hrl").
-include_lib("emqx_connector_jwt/include/emqx_connector_jwt_tables.hrl").
-include_lib("emqx_resource/include/emqx_resource.hrl").
-include_lib("typerefl/include/types.hrl").
-include_lib("emqx/include/logger.hrl").

3
apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub_impl_consumer.erl
View File

@ -8,6 +8,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
query_mode/1,
on_start/2,
@ -84,6 +85,8 @@
%%-------------------------------------------------------------------------------------------------
%% `emqx_resource' API
%%-------------------------------------------------------------------------------------------------
-spec resource_type() -> resource_type().
resource_type() -> gcp_pubsub_consumer.
-spec callback_mode() -> callback_mode().
callback_mode() -> async_if_possible.

2
apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub_impl_producer.erl
View File

@ -41,6 +41,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
query_mode/1,
on_start/2,
@ -62,6 +63,7 @@
%%-------------------------------------------------------------------------------------------------
%% `emqx_resource' API
%%-------------------------------------------------------------------------------------------------
resource_type() -> gcp_pubsub.
callback_mode() -> async_if_possible.

2
apps/emqx_bridge_gcp_pubsub/test/emqx_bridge_gcp_pubsub_consumer_SUITE.erl
View File

@ -594,7 +594,7 @@ cluster(Config) ->
Cluster = emqx_common_test_helpers:emqx_cluster(
[core, core],
[
{apps, [emqx_conf, emqx_rule_engine, emqx_bridge]},
{apps, [emqx_conf, emqx_rule_engine, emqx_bridge_gcp_pubsub, emqx_bridge]},
{listener_ports, []},
{priv_data_dir, PrivDataDir},
{load_schema, true},

3
apps/emqx_bridge_greptimedb/src/emqx_bridge_greptimedb_connector.erl
View File

@ -16,6 +16,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -67,6 +68,8 @@
%% -------------------------------------------------------------------------------------------------
%% resource callback
resource_type() -> greptimedb.
callback_mode() -> async_if_possible.
on_add_channel(

2
apps/emqx_bridge_hstreamdb/src/emqx_bridge_hstreamdb.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_hstreamdb, [
{description, "EMQX Enterprise HStreamDB Bridge"},
{vsn, "0.2.1"},
{vsn, "0.2.2"},
{registered, []},
{applications, [
kernel,

3
apps/emqx_bridge_hstreamdb/src/emqx_bridge_hstreamdb_connector.erl
View File

@ -16,6 +16,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -44,6 +45,8 @@
%% -------------------------------------------------------------------------------------------------
%% resource callback
resource_type() -> hstreamdb.
callback_mode() -> always_sync.
on_start(InstId, Config) ->

2
apps/emqx_bridge_http/src/emqx_bridge_http_connector.erl
View File

@ -26,6 +26,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -183,6 +184,7 @@ sc(Type, Meta) -> hoconsc:mk(Type, Meta).
ref(Field) -> hoconsc:ref(?MODULE, Field).
%% ===================================================================
resource_type() -> webhook.
callback_mode() -> async_if_possible.

8
apps/emqx_bridge_http/src/emqx_bridge_http_schema.erl
View File

@ -72,14 +72,8 @@ fields(action) ->
}
)};
fields("http_action") ->
emqx_bridge_v2_schema:common_fields() ++
[
{enable, mk(boolean(), #{desc => ?DESC("config_enable_bridge"), default => true})},
{connector,
mk(binary(), #{
desc => ?DESC(emqx_connector_schema, "connector_field"), required => true
})},
{tags, emqx_schema:tags_schema()},
{description, emqx_schema:description_schema()},
%% Note: there's an implicit convention in `emqx_bridge' that,
%% for egress bridges with this config, the published messages
%% will be forwarded to such bridges.

2
apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_bridge_influxdb, [
{description, "EMQX Enterprise InfluxDB Bridge"},
{vsn, "0.2.3"},
{vsn, "0.2.4"},
{registered, []},
{applications, [
kernel,

3
apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb_connector.erl
View File

@ -16,6 +16,7 @@
%% callbacks of behaviour emqx_resource
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -70,6 +71,8 @@
%% -------------------------------------------------------------------------------------------------
%% resource callback
resource_type() -> influxdb.
callback_mode() -> async_if_possible.
on_add_channel(

2
apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb.app.src
View File

@ -1,7 +1,7 @@
%% -*- mode: erlang -*-
{application, emqx_bridge_iotdb, [
{description, "EMQX Enterprise Apache IoTDB Bridge"},
{vsn, "0.2.2"},
{vsn, "0.2.3"},
{modules, [
emqx_bridge_iotdb,
emqx_bridge_iotdb_connector

3
apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb_connector.erl
View File

@ -15,6 +15,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
on_start/2,
on_stop/2,
@ -206,6 +207,8 @@ proplists_without(Keys, List) ->
%%-------------------------------------------------------------------------------------
%% `emqx_resource' API
%%-------------------------------------------------------------------------------------
resource_type() -> iotdb.
callback_mode() -> async_if_possible.
-spec on_start(manager_id(), config()) -> {ok, state()} | no_return().

2
apps/emqx_bridge_kafka/mix.exs
View File

@ -23,7 +23,7 @@ defmodule EMQXBridgeKafka.MixProject do
def deps() do
[
{:wolff, github: "kafka4beam/wolff", tag: "2.0.0"},
{:wolff, github: "kafka4beam/wolff", tag: "3.0.2"},
{:kafka_protocol, github: "kafka4beam/kafka_protocol", tag: "4.1.5", override: true},
{:brod_gssapi, github: "kafka4beam/brod_gssapi", tag: "v0.1.1"},
{:brod, github: "kafka4beam/brod", tag: "3.18.0"},

2
apps/emqx_bridge_kafka/rebar.config
View File

@ -2,7 +2,7 @@
{erl_opts, [debug_info]}.
{deps, [
{wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "2.0.0"}}},
{wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "3.0.2"}}},
{kafka_protocol, {git, "https://github.com/kafka4beam/kafka_protocol.git", {tag, "4.1.5"}}},
{brod_gssapi, {git, "https://github.com/kafka4beam/brod_gssapi.git", {tag, "v0.1.1"}}},
{brod, {git, "https://github.com/kafka4beam/brod.git", {tag, "3.18.0"}}},

45
apps/emqx_bridge_kafka/src/emqx_bridge_kafka.erl
View File

@ -295,17 +295,10 @@ fields("config_producer") ->
fields("config_consumer") ->
fields(kafka_consumer);
fields(kafka_producer) ->
%% Schema used by bridges V1.
connector_config_fields() ++ producer_opts(v1);
fields(kafka_producer_action) ->
[
{enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
{connector,
mk(binary(), #{
desc => ?DESC(emqx_connector_schema, "connector_field"), required => true
})},
{tags, emqx_schema:tags_schema()},
{description, emqx_schema:description_schema()}
] ++ producer_opts(action);
emqx_bridge_v2_schema:common_fields() ++ producer_opts(action);
fields(kafka_consumer) ->
connector_config_fields() ++ fields(consumer_opts);
fields(ssl_client_opts) ->
@ -364,9 +357,33 @@ fields(socket_opts) ->
validator => fun emqx_schema:validate_tcp_keepalive/1
})}
];
fields(v1_producer_kafka_opts) ->
OldSchemaFields =
[
topic,
message,
max_batch_bytes,
compression,
partition_strategy,
required_acks,
kafka_headers,
kafka_ext_headers,
kafka_header_value_encode_mode,
partition_count_refresh_interval,
partitions_limit,
max_inflight,
buffer,
query_mode,
sync_query_timeout
],
Fields = fields(producer_kafka_opts),
lists:filter(
fun({K, _V}) -> lists:member(K, OldSchemaFields) end,
Fields
);
fields(producer_kafka_opts) ->
[
{topic, mk(string(), #{required => true, desc => ?DESC(kafka_topic)})},
{topic, mk(emqx_schema:template(), #{required => true, desc => ?DESC(kafka_topic)})},
{message, mk(ref(kafka_message), #{required => false, desc => ?DESC(kafka_message)})},
{max_batch_bytes,
mk(emqx_schema:bytesize(), #{default => <<"896KB">>, desc => ?DESC(max_batch_bytes)})},
@ -680,15 +697,15 @@ resource_opts() ->
%% However we need to keep it backward compatible for generated schema json (version 0.1.0)
%% since schema is data for the 'schemas' API.
parameters_field(ActionOrBridgeV1) ->
{Name, Alias} =
{Name, Alias, Ref} =
case ActionOrBridgeV1 of
v1 ->
{kafka, parameters};
{kafka, parameters, v1_producer_kafka_opts};
action ->
{parameters, kafka}
{parameters, kafka, producer_kafka_opts}
end,
{Name,
mk(ref(producer_kafka_opts), #{
mk(ref(Ref), #{
required => true,
aliases => [Alias],
desc => ?DESC(producer_kafka_opts),

14
apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_consumer.erl
View File

@ -7,6 +7,7 @@
%% `emqx_resource' API
-export([
resource_type/0,
callback_mode/0,
query_mode/1,
on_start/2,
@ -126,6 +127,7 @@
%%-------------------------------------------------------------------------------------
%% `emqx_resource' API
%%-------------------------------------------------------------------------------------
resource_type() -> kafka_consumer.
callback_mode() ->
async_if_possible.
@ -631,16 +633,6 @@ consumer_group_id(_ConsumerParams, BridgeName0) ->
BridgeName = to_bin(BridgeName0),
<<"emqx-kafka-consumer-", BridgeName/binary>>.
-spec is_dry_run(connector_resource_id()) -> boolean().
is_dry_run(ConnectorResId) ->
TestIdStart = string:find(ConnectorResId, ?TEST_ID_PREFIX),
case TestIdStart of
nomatch ->
false;
_ ->
string:equal(TestIdStart, ConnectorResId)
end.
-spec check_client_connectivity(pid()) ->
?status_connected
| ?status_disconnected
@ -676,7 +668,7 @@ maybe_clean_error(Reason) ->
-spec make_client_id(connector_resource_id(), binary(), atom() | binary()) -> atom().
make_client_id(ConnectorResId, BridgeType, BridgeName) ->
case is_dry_run(ConnectorResId) of
case emqx_resource:is_dry_run(ConnectorResId) of
false ->
ClientID0 = emqx_bridge_kafka_impl:make_client_id(BridgeType, BridgeName),
binary_to_atom(ClientID0);

Some files were not shown because too many files have changed in this diff Show More