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

Merge remote-tracking branch 'origin/main-v4.3' into sync-v4.3-to-v4.4

This commit is contained in:
Zaiming Shi 2021-11-08 21:01:14 +01:00
parent 14474a2739 6b06142562
commit 6222e1b0eb
28 changed files with 1004 additions and 347 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
.github/workflows/elvis_lint.yaml → .github/workflows/code_style_check.yaml vendored
View File

@ -1,4 +1,4 @@
name: Elvis Linter
name: Code style check
on: [pull_request]
@ -7,10 +7,18 @@ jobs:
runs-on: ubuntu-20.04
steps:
- uses: actions/checkout@v2
with:
fetch-depth: 1000
- name: Set git token
if: endsWith(github.repository, 'enterprise')
run: |
echo "https://ci%40emqx.io:${{ secrets.CI_GIT_TOKEN }}@github.com" > $HOME/.git-credentials
git config --global credential.helper store
- run: |
./scripts/elvis-check.sh $GITHUB_BASE_REF
- name: Run elvis check
run: |
set -e
if [ -f EMQX_ENTERPRISE ]; then
./scripts/elvis-check.sh $GITHUB_BASE_REF emqx-enterprise
else
./scripts/elvis-check.sh $GITHUB_BASE_REF emqx
fi

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

@ -24,6 +24,7 @@ jobs:
git config --global credential.helper store
echo "${{ secrets.CI_GIT_TOKEN }}" >> scripts/git-token
make deps-emqx-ee
make clean
fi
make docker
echo "::set-output name=version::$(./pkg-vsn.sh)"
@ -95,7 +96,7 @@ jobs:
- uses: actions/checkout@v2
with:
repository: emqx/emqx-fvt
ref: integration_test_suites
ref: v1.4.0
path: scripts
- uses: actions/setup-java@v1
with:
@ -188,7 +189,7 @@ jobs:
- uses: actions/checkout@v2
with:
repository: emqx/emqx-fvt
ref: integration_test_suites
ref: v1.4.0
path: scripts
- uses: actions/setup-java@v1
with:
@ -292,7 +293,7 @@ jobs:
- uses: actions/checkout@v2
with:
repository: emqx/emqx-fvt
ref: integration_test_suites
ref: v1.4.0
path: scripts
- uses: actions/setup-java@v1
with:
@ -390,7 +391,7 @@ jobs:
- uses: actions/checkout@v2
with:
repository: emqx/emqx-fvt
ref: integration_test_suites
ref: v1.4.0
path: scripts
- uses: actions/setup-java@v1
with:

3
.github/workflows/run_fvt_tests.yaml vendored
View File

@ -223,7 +223,7 @@ jobs:
- name: Generate matrix
id: generate-matrix
run: |
matrix=$(echo -n "$OLD_VSNS" | jq -R -s -c 'split(" ")')
matrix=$(echo -n "$OLD_VSNS" | sed 's/ $//g' | jq -R -s -c 'split(" ")')
echo "::set-output name=matrix::$matrix"
relup_test_build:
@ -275,6 +275,7 @@ jobs:
runs-on: ubuntu-20.04
container: emqx/relup-test-env:erl23.2.7.2-emqx-2-ubuntu20.04
strategy:
fail-fast: false
matrix:
old_vsn: ${{ fromJson(needs.relup_test_plan.outputs.matrix) }}
env:

2
apps/emqx_auth_pgsql/rebar.config
View File

@ -1,5 +1,5 @@
{deps,
[{epgsql, {git, "https://github.com/epgsql/epgsql", {tag, "4.4.0"}}}
[{epgsql, {git, "https://github.com/epgsql/epgsql.git", {tag, "4.4.0"}}}
]}.
{erl_opts, [warn_unused_vars,

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

@ -1,6 +1,6 @@
{application, emqx_exproto,
[{description, "EMQ X Extension for Protocol"},
{vsn, "4.3.2"}, %% strict semver
{vsn, "4.3.4"}, %% 4.3.3 is used by ee
{modules, []},
{registered, []},
{mod, {emqx_exproto_app, []}},

55
apps/emqx_exproto/src/emqx_exproto.appup.src
View File

@ -1,31 +1,26 @@
%% -*-: erlang -*-
%% -*- mode: erlang -*-
{VSN,
[
{"4.3.1", [
{load_module, emqx_exproto_gsvr, brutal_purge, soft_purge, []},
{load_module, emqx_exproto_gcli, brutal_purge, soft_purge, []},
{load_module, emqx_exproto_channel, brutal_purge, soft_purge, []}
]},
{"4.3.0", [
{load_module, emqx_exproto_gsvr, brutal_purge, soft_purge, []},
{load_module, emqx_exproto_gcli, brutal_purge, soft_purge, []},
{load_module, emqx_exproto_conn, brutal_purge, soft_purge, []},
{load_module, emqx_exproto_channel, brutal_purge, soft_purge, []}
]},
{<<".*">>, []}
],
[
{"4.3.1", [
{load_module, emqx_exproto_gsvr, brutal_purge, soft_purge, []},
{load_module, emqx_exproto_gcli, brutal_purge, soft_purge, []},
{load_module, emqx_exproto_channel, brutal_purge, soft_purge, []}
]},
{"4.3.0", [
{load_module, emqx_exproto_gsvr, brutal_purge, soft_purge, []},
{load_module, emqx_exproto_gcli, brutal_purge, soft_purge, []},
{load_module, emqx_exproto_conn, brutal_purge, soft_purge, []},
{load_module, emqx_exproto_channel, brutal_purge, soft_purge, []}
]},
{<<".*">>, []}
]
}.
[{"4.3.3",
[{load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
{load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]},
{"4.3.2",
[{load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
{load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]},
{<<"4.3.[0-1]">>,
[{load_module,emqx_exproto_gsvr,brutal_purge,soft_purge,[]},
{load_module,emqx_exproto_gcli,brutal_purge,soft_purge,[]},
{load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
{load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]},
{<<".*">>,[]}],
[{"4.3.3",
[{load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
{load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]},
{"4.3.2",
[{load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
{load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]},
{<<"4.3.[0-1]">>,
[{load_module,emqx_exproto_gsvr,brutal_purge,soft_purge,[]},
{load_module,emqx_exproto_gcli,brutal_purge,soft_purge,[]},
{load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
{load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]},
{<<".*">>,[]}]}.

31
apps/emqx_exproto/src/emqx_exproto_channel.erl
View File

@ -94,6 +94,9 @@
awaiting_rel_max
]).
-define(CHANMOCK(P), {exproto_anonymous_client, P}).
-define(CHAN_CONN_TAB, emqx_channel_conn).
%%--------------------------------------------------------------------
%% Info, Attrs and Caps
%%--------------------------------------------------------------------
@ -103,7 +106,7 @@
info(Channel) ->
maps:from_list(info(?INFO_KEYS, Channel)).
-spec(info(list(atom())|atom(), channel()) -> term()).
-spec(info(list(atom()) | atom(), channel()) -> term()).
info(Keys, Channel) when is_list(Keys) ->
[{Key, info(Key, Channel)} || Key <- Keys];
info(conninfo, #channel{conninfo = ConnInfo}) ->
@ -155,13 +158,19 @@ init(ConnInfo = #{socktype := Socktype,
conn_state = connecting,
timers = #{}
},
Req = #{conninfo =>
peercert(Peercert,
#{socktype => socktype(Socktype),
peername => address(Peername),
sockname => address(Sockname)})},
try_dispatch(on_socket_created, wrap(Req), Channel).
case emqx_hooks:run_fold('client.connect', [NConnInfo], #{}) of
{error, _Reason} ->
throw(nopermission);
_ ->
ConnMod = maps:get(conn_mod, NConnInfo),
true = ets:insert(?CHAN_CONN_TAB, {?CHANMOCK(self()), ConnMod}),
Req = #{conninfo =>
peercert(Peercert,
#{socktype => socktype(Socktype),
peername => address(Peername),
sockname => address(Sockname)})},
try_dispatch(on_socket_created, wrap(Req), Channel)
end.
%% @private
peercert(NoSsl, ConnInfo) when NoSsl == nossl;
@ -283,6 +292,7 @@ handle_call({auth, ClientInfo0, Password},
emqx_metrics:inc('client.auth.anonymous'),
NClientInfo = maps:merge(ClientInfo1, AuthResult),
NChannel = Channel1#channel{clientinfo = NClientInfo},
clean_anonymous_clients(),
case emqx_cm:open_session(true, NClientInfo, NConnInfo) of
{ok, _Session} ->
?LOG(debug, "Client ~s (Username: '~s') authorized successfully!",
@ -399,12 +409,16 @@ handle_info(Info, Channel) ->
-spec(terminate(any(), channel()) -> channel()).
terminate(Reason, Channel) ->
clean_anonymous_clients(),
Req = #{reason => stringfy(Reason)},
try_dispatch(on_socket_closed, wrap(Req), Channel).
is_anonymous(#{anonymous := true}) -> true;
is_anonymous(_AuthResult) -> false.
clean_anonymous_clients() ->
ets:delete(?CHAN_CONN_TAB, ?CHANMOCK(self())).
%%--------------------------------------------------------------------
%% Sub/UnSub
%%--------------------------------------------------------------------
@ -577,7 +591,6 @@ default_conninfo(ConnInfo) ->
ConnInfo#{clean_start => true,
clientid => undefined,
username => undefined,
conn_mod => undefined,
conn_props => #{},
connected => true,
connected_at => erlang:system_time(millisecond),

20
apps/emqx_exproto/src/emqx_exproto_conn.erl
View File

@ -115,7 +115,7 @@ start_link(esockd_transport, Sock, Options) ->
%%--------------------------------------------------------------------
%% @doc Get infos of the connection/channel.
-spec(info(pid()|state()) -> emqx_types:infos()).
-spec(info(pid() | state()) -> emqx_types:infos()).
info(CPid) when is_pid(CPid) ->
call(CPid, info);
info(State = #state{channel = Channel}) ->
@ -137,7 +137,7 @@ info(sockstate, #state{sockstate = SockSt}) ->
info(active_n, #state{active_n = ActiveN}) ->
ActiveN.
-spec(stats(pid()|state()) -> emqx_types:stats()).
-spec(stats(pid() | state()) -> emqx_types:stats()).
stats(CPid) when is_pid(CPid) ->
call(CPid, stats);
stats(#state{socket = Socket,
@ -233,7 +233,11 @@ init(Parent, WrappedSock, Peername0, Options) ->
case esockd_wait(WrappedSock) of
{ok, NWrappedSock} ->
Peername = esockd_peername(NWrappedSock, Peername0),
run_loop(Parent, init_state(NWrappedSock, Peername, Options));
try
run_loop(Parent, init_state(NWrappedSock, Peername, Options))
catch
throw : nopermission -> erlang:exit(normal)
end;
{error, Reason} ->
ok = esockd_close(WrappedSock),
exit_on_sock_error(Reason)
@ -337,7 +341,7 @@ cancel_stats_timer(State) -> State.
process_msg([], Parent, State) -> recvloop(Parent, State);
process_msg([Msg|More], Parent, State) ->
process_msg([Msg | More], Parent, State) ->
case catch handle_msg(Msg, State) of
ok ->
process_msg(More, Parent, State);
@ -413,7 +417,7 @@ handle_msg({Passive, _Sock}, State)
handle_msg(Deliver = {deliver, _Topic, _Msg},
State = #state{active_n = ActiveN}) ->
Delivers = [Deliver|emqx_misc:drain_deliver(ActiveN)],
Delivers = [Deliver | emqx_misc:drain_deliver(ActiveN)],
with_channel(handle_deliver, [Delivers], State);
%% Something sent
@ -601,9 +605,9 @@ handle_outgoing(IoData, State = #state{socket = Socket}) ->
handle_info(activate_socket, State = #state{sockstate = OldSst}) ->
case activate_socket(State) of
{ok, NState = #state{sockstate = NewSst}} ->
if OldSst =/= NewSst ->
{ok, {event, NewSst}, NState};
true -> {ok, NState}
case OldSst =/= NewSst of
true -> {ok, {event, NewSst}, NState};
false -> {ok, NState}
end;
{error, Reason} ->
handle_info({sock_error, Reason}, State)

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

@ -1,6 +1,6 @@
{application, emqx_rule_engine,
[{description, "EMQ X Rule Engine"},
{vsn, "4.3.5"}, % strict semver, bump manually!
{vsn, "4.3.6"}, % strict semver, bump manually!
{modules, []},
{registered, [emqx_rule_engine_sup, emqx_rule_registry]},
{applications, [kernel,stdlib,rulesql,getopt]},

106
apps/emqx_rule_engine/src/emqx_rule_engine.appup.src
View File

@ -1,64 +1,52 @@
%% -*-: erlang -*-
{"4.3.5",
[ {"4.3.0",
[ {load_module, emqx_rule_funcs, brutal_purge, soft_purge, []}
, {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
, {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
, {apply, {emqx_stats, cancel_update, [rule_registery_stats]}}
, {load_module, emqx_rule_actions, brutal_purge, soft_purge, []}
]},
{"4.3.1",
[ {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
, {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
, {apply, {emqx_stats, cancel_update, [rule_registery_stats]}}
, {load_module, emqx_rule_actions, brutal_purge, soft_purge, []}
]},
{"4.3.2",
[ {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
, {apply, {emqx_stats, cancel_update, [rule_registery_stats]}}
, {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
, {load_module, emqx_rule_actions, brutal_purge, soft_purge, []}
]},
{"4.3.3",
[ {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
, {load_module, emqx_rule_actions, brutal_purge, soft_purge, []}
, {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
]},
{"4.3.4",
[ {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
, {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
]},
{<<".*">>, []}
],
[
%% -*- mode: erlang -*-
{VSN,
[{"4.3.5",[{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
{"4.3.0",
[ {load_module, emqx_rule_funcs, brutal_purge, soft_purge, []}
, {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
, {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
, {apply, {emqx_stats, cancel_update, [rule_registery_stats]}}
, {load_module, emqx_rule_actions, brutal_purge, soft_purge, []}
]},
[{load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
{load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
{apply,{emqx_stats,cancel_update,[rule_registery_stats]}},
{load_module,emqx_rule_actions,brutal_purge,soft_purge,[]}]},
{"4.3.1",
[ {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
, {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
, {apply, {emqx_stats, cancel_update, [rule_registery_stats]}}
, {load_module, emqx_rule_actions, brutal_purge, soft_purge, []}
]},
[{load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
{apply,{emqx_stats,cancel_update,[rule_registery_stats]}},
{load_module,emqx_rule_actions,brutal_purge,soft_purge,[]}]},
{"4.3.2",
[ {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
, {apply, {emqx_stats, cancel_update, [rule_registery_stats]}}
, {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
, {load_module, emqx_rule_actions, brutal_purge, soft_purge, []}
]},
[{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
{apply,{emqx_stats,cancel_update,[rule_registery_stats]}},
{load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
{load_module,emqx_rule_actions,brutal_purge,soft_purge,[]}]},
{"4.3.3",
[ {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
, {load_module, emqx_rule_actions, brutal_purge, soft_purge, []}
, {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
]},
[{load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
{load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
{"4.3.4",
[ {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
, {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
]},
{<<".*">>, []}
]
}.
[{load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
{<<".*">>,[]}],
[{"4.3.5",[{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
{"4.3.0",
[{load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
{load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
{apply,{emqx_stats,cancel_update,[rule_registery_stats]}},
{load_module,emqx_rule_actions,brutal_purge,soft_purge,[]}]},
{"4.3.1",
[{load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
{apply,{emqx_stats,cancel_update,[rule_registery_stats]}},
{load_module,emqx_rule_actions,brutal_purge,soft_purge,[]}]},
{"4.3.2",
[{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
{apply,{emqx_stats,cancel_update,[rule_registery_stats]}},
{load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
{load_module,emqx_rule_actions,brutal_purge,soft_purge,[]}]},
{"4.3.3",
[{load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
{load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
{"4.3.4",
[{load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
{<<".*">>,[]}]}.

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

@ -1,6 +1,6 @@
{application, emqx_stomp,
[{description, "EMQ X Stomp Protocol Plugin"},
{vsn, "4.3.0"}, % strict semver, bump manually!
{vsn, "4.3.2"}, % strict semver, bump manually!
{modules, []},
{registered, [emqx_stomp_sup]},
{applications, [kernel,stdlib]},

10
apps/emqx_stomp/src/emqx_stomp.appup.src Normal file
View File

@ -0,0 +1,10 @@
%% -*- mode: erlang -*-
{VSN,
[{"4.3.1",[{load_module,emqx_stomp_connection,brutal_purge,soft_purge,[]}]},
{"4.3.0",
[{restart_application,emqx_stomp}]},
{<<".*">>,[]}],
[{"4.3.1",[{load_module,emqx_stomp_connection,brutal_purge,soft_purge,[]}]},
{"4.3.0",
[{restart_application,emqx_stomp}]},
{<<".*">>,[]}]}.

434
apps/emqx_stomp/src/emqx_stomp_connection.erl
View File

@ -20,9 +20,15 @@
-include("emqx_stomp.hrl").
-include_lib("emqx/include/logger.hrl").
-include_lib("emqx/include/types.hrl").
-include_lib("snabbkaffe/include/snabbkaffe.hrl").
-logger_header("[Stomp-Conn]").
-import(emqx_misc,
[ start_timer/2
]).
-export([ start_link/3
, info/1
]).
@ -37,56 +43,177 @@
]).
%% for protocol
-export([send/4, heartbeat/2]).
-export([send/4, heartbeat/2, statfun/3]).
-record(state, {transport, socket, peername, conn_name, conn_state,
await_recv, rate_limit, parser, pstate,
proto_env, heartbeat}).
%% for mgmt
-export([call/2, call/3]).
-define(INFO_KEYS, [peername, await_recv, conn_state]).
-define(SOCK_STATS, [recv_oct, recv_cnt, send_oct, send_cnt]).
-record(state, {
%% TCP/TLS Transport
transport :: esockd:transport(),
%% TCP/TLS Socket
socket :: esockd:socket(),
%% Peername of the connection
peername :: emqx_types:peername(),
%% Sockname of the connection
sockname :: emqx_types:peername(),
%% Sock State
sockstate :: emqx_types:sockstate(),
%% The {active, N} option
active_n :: pos_integer(),
%% Limiter
limiter :: maybe(emqx_limiter:limiter()),
%% Limit Timer
limit_timer :: maybe(reference()),
%% GC State
gc_state :: maybe(emqx_gc:gc_state()),
%% Stats Timer
stats_timer :: disabled | maybe(reference()),
%% Parser State
parser :: emqx_stomp_frame:parser(),
%% Protocol State
pstate :: emqx_stomp_protocol:pstate(),
%% XXX: some common confs
proto_env :: list()
}).
-type(state() :: #state{}).
-define(DEFAULT_GC_POLICY, #{bytes => 16777216, count => 16000}).
-define(DEFAULT_OOM_POLICY, #{ max_heap_size => 8388608,
message_queue_len => 10000}).
-define(ACTIVE_N, 100).
-define(IDLE_TIMEOUT, 30000).
-define(INFO_KEYS, [socktype, peername, sockname, sockstate, active_n]).
-define(CONN_STATS, [recv_pkt, recv_msg, send_pkt, send_msg]).
-define(SOCK_STATS, [recv_oct, recv_cnt, send_oct, send_cnt, send_pend]).
-define(ENABLED(X), (X =/= undefined)).
-elvis([{elvis_style, invalid_dynamic_call, #{ignore => [emqx_stomp_connection]}}]).
-dialyzer({nowarn_function, [ ensure_stats_timer/2
]}).
-dialyzer({no_return, [ init/1
, init_state/3
]}).
start_link(Transport, Sock, ProtoEnv) ->
{ok, proc_lib:spawn_link(?MODULE, init, [[Transport, Sock, ProtoEnv]])}.
info(CPid) ->
gen_server:call(CPid, info, infinity).
-spec info(pid() | state()) -> emqx_types:infos().
info(CPid) when is_pid(CPid) ->
call(CPid, info);
info(State = #state{pstate = PState}) ->
ChanInfo = emqx_stomp_protocol:info(PState),
SockInfo = maps:from_list(
info(?INFO_KEYS, State)),
ChanInfo#{sockinfo => SockInfo}.
init([Transport, Sock, ProtoEnv]) ->
process_flag(trap_exit, true),
case Transport:wait(Sock) of
{ok, NewSock} ->
{ok, Peername} = Transport:ensure_ok_or_exit(peername, [NewSock]),
ConnName = esockd:format(Peername),
SendFun = {fun ?MODULE:send/4, [Transport, Sock, self()]},
HrtBtFun = {fun ?MODULE:heartbeat/2, [Transport, Sock]},
Parser = emqx_stomp_frame:init_parer_state(ProtoEnv),
PState = emqx_stomp_protocol:init(#{peername => Peername,
sendfun => SendFun,
heartfun => HrtBtFun}, ProtoEnv),
RateLimit = init_rate_limit(proplists:get_value(rate_limit, ProtoEnv)),
State = run_socket(#state{transport = Transport,
socket = NewSock,
peername = Peername,
conn_name = ConnName,
conn_state = running,
await_recv = false,
rate_limit = RateLimit,
parser = Parser,
proto_env = ProtoEnv,
pstate = PState}),
emqx_logger:set_metadata_peername(esockd:format(Peername)),
gen_server:enter_loop(?MODULE, [{hibernate_after, 5000}], State, 20000);
info(Keys, State) when is_list(Keys) ->
[{Key, info(Key, State)} || Key <- Keys];
info(socktype, #state{transport = Transport, socket = Socket}) ->
Transport:type(Socket);
info(peername, #state{peername = Peername}) ->
Peername;
info(sockname, #state{sockname = Sockname}) ->
Sockname;
info(sockstate, #state{sockstate = SockSt}) ->
SockSt;
info(active_n, #state{active_n = ActiveN}) ->
ActiveN.
-spec stats(pid() | state()) -> emqx_types:stats().
stats(CPid) when is_pid(CPid) ->
call(CPid, stats);
stats(#state{transport = Transport,
socket = Socket,
pstate = PState}) ->
SockStats = case Transport:getstat(Socket, ?SOCK_STATS) of
{ok, Ss} -> Ss;
{error, _} -> []
end,
ConnStats = emqx_pd:get_counters(?CONN_STATS),
ChanStats = emqx_stomp_protocol:stats(PState),
ProcStats = emqx_misc:proc_stats(),
lists:append([SockStats, ConnStats, ChanStats, ProcStats]).
call(Pid, Req) ->
call(Pid, Req, infinity).
call(Pid, Req, Timeout) ->
gen_server:call(Pid, Req, Timeout).
init([Transport, RawSocket, ProtoEnv]) ->
case Transport:wait(RawSocket) of
{ok, Socket} ->
init_state(Transport, Socket, ProtoEnv);
{error, Reason} ->
{stop, Reason}
ok = Transport:fast_close(RawSocket),
exit_on_sock_error(Reason)
end.
init_rate_limit(undefined) ->
undefined;
init_rate_limit({Rate, Burst}) ->
esockd_rate_limit:new(Rate, Burst).
init_state(Transport, Socket, ProtoEnv) ->
{ok, Peername} = Transport:ensure_ok_or_exit(peername, [Socket]),
{ok, Sockname} = Transport:ensure_ok_or_exit(sockname, [Socket]),
send(Data, Transport, Sock, ConnPid) ->
SendFun = {fun ?MODULE:send/4, [Transport, Socket, self()]},
StatFun = {fun ?MODULE:statfun/3, [Transport, Socket]},
HrtBtFun = {fun ?MODULE:heartbeat/2, [Transport, Socket]},
Parser = emqx_stomp_frame:init_parer_state(ProtoEnv),
ActiveN = proplists:get_value(active_n, ProtoEnv, ?ACTIVE_N),
GcState = emqx_gc:init(?DEFAULT_GC_POLICY),
Peercert = Transport:ensure_ok_or_exit(peercert, [Socket]),
ConnInfo = #{socktype => Transport:type(Socket),
peername => Peername,
sockname => Sockname,
peercert => Peercert,
statfun => StatFun,
sendfun => SendFun,
heartfun => HrtBtFun,
conn_mod => ?MODULE
},
PState = emqx_stomp_protocol:init(ConnInfo, ProtoEnv),
State = #state{transport = Transport,
socket = Socket,
peername = Peername,
sockname = Sockname,
sockstate = idle,
active_n = ActiveN,
limiter = undefined,
parser = Parser,
proto_env = ProtoEnv,
gc_state = GcState,
pstate = PState},
case activate_socket(State) of
{ok, NState} ->
emqx_logger:set_metadata_peername(
esockd:format(Peername)),
gen_server:enter_loop(
?MODULE, [{hibernate_after, 5000}], NState, 20000);
{error, Reason} ->
ok = Transport:fast_close(Socket),
exit_on_sock_error(Reason)
end.
-spec exit_on_sock_error(any()) -> no_return().
exit_on_sock_error(Reason) when Reason =:= einval;
Reason =:= enotconn;
Reason =:= closed ->
erlang:exit(normal);
exit_on_sock_error(timeout) ->
erlang:exit({shutdown, ssl_upgrade_timeout});
exit_on_sock_error(Reason) ->
erlang:exit({shutdown, Reason}).
send(Frame, Transport, Sock, ConnPid) ->
?LOG(info, "SEND Frame: ~s", [emqx_stomp_frame:format(Frame)]),
ok = inc_outgoing_stats(Frame),
Data = emqx_stomp_frame:serialize(Frame),
?LOG(debug, "SEND ~p", [Data]),
try Transport:async_send(Sock, Data) of
ok -> ok;
{error, Reason} -> ConnPid ! {shutdown, Reason}
@ -95,23 +222,27 @@ send(Data, Transport, Sock, ConnPid) ->
end.
heartbeat(Transport, Sock) ->
?LOG(debug, "SEND heartbeat: \\n"),
Transport:send(Sock, <<$\n>>).
handle_call(info, _From, State = #state{transport = Transport,
socket = Sock,
peername = Peername,
await_recv = AwaitRecv,
conn_state = ConnState,
pstate = PState}) ->
ClientInfo = [{peername, Peername}, {await_recv, AwaitRecv},
{conn_state, ConnState}],
ProtoInfo = emqx_stomp_protocol:info(PState),
case Transport:getstat(Sock, ?SOCK_STATS) of
{ok, SockStats} ->
{reply, lists:append([ClientInfo, ProtoInfo, SockStats]), State};
{error, Reason} ->
{stop, Reason, lists:append([ClientInfo, ProtoInfo]), State}
end;
statfun(Stat, Transport, Sock) ->
case Transport:getstat(Sock, [Stat]) of
{ok, [{Stat, Val}]} -> {ok, Val};
{error, Error} -> {error, Error}
end.
handle_call(info, _From, State) ->
{reply, info(State), State};
handle_call(stats, _From, State) ->
{reply, stats(State), State};
handle_call(discard, _From, State) ->
%% TODO: send the DISCONNECT packet?
shutdown_and_reply(discared, ok, State);
handle_call(kick, _From, State) ->
shutdown_and_reply(kicked, ok, State);
handle_call(Req, _From, State) ->
?LOG(error, "unexpected request: ~p", [Req]),
@ -121,6 +252,13 @@ handle_cast(Msg, State) ->
?LOG(error, "unexpected msg: ~p", [Msg]),
noreply(State).
handle_info({event, Name}, State = #state{pstate = PState})
when Name == connected;
Name == updated ->
ClientId = emqx_stomp_protocol:info(clientid, PState),
emqx_cm:insert_channel_info(ClientId, info(State), stats(State)),
noreply(State);
handle_info(timeout, State) ->
shutdown(idle_timeout, State);
@ -141,26 +279,70 @@ handle_info({timeout, TRef, TMsg}, State) when TMsg =:= incoming;
shutdown({sock_error, Reason}, State)
end;
handle_info({timeout, _TRef, limit_timeout}, State) ->
NState = State#state{sockstate = idle,
limit_timer = undefined
},
handle_info(activate_socket, NState);
handle_info({timeout, _TRef, emit_stats},
State = #state{pstate = PState}) ->
ClientId = emqx_stomp_protocol:info(clientid, PState),
emqx_cm:set_chan_stats(ClientId, stats(State)),
noreply(State#state{stats_timer = undefined});
handle_info({timeout, TRef, TMsg}, State) ->
with_proto(timeout, [TRef, TMsg], State);
handle_info({'EXIT', HbProc, Error}, State = #state{heartbeat = HbProc}) ->
stop(Error, State);
handle_info(activate_sock, State) ->
noreply(run_socket(State#state{conn_state = running}));
handle_info({inet_async, _Sock, _Ref, {ok, Bytes}}, State) ->
?LOG(debug, "RECV ~p", [Bytes]),
received(Bytes, rate_limit(size(Bytes), State#state{await_recv = false}));
handle_info({inet_async, _Sock, _Ref, {error, Reason}}, State) ->
shutdown(Reason, State);
handle_info(activate_socket, State) ->
case activate_socket(State) of
{ok, NState} ->
noreply(NState);
{error, Reason} ->
handle_info({sock_error, Reason}, State)
end;
handle_info({inet_reply, _Ref, ok}, State) ->
noreply(State);
handle_info({Inet, _Sock, Data}, State) when Inet == tcp; Inet == ssl ->
?LOG(debug, "RECV ~0p", [Data]),
Oct = iolist_size(Data),
inc_counter(incoming_bytes, Oct),
ok = emqx_metrics:inc('bytes.received', Oct),
received(Data, ensure_stats_timer(?IDLE_TIMEOUT, State));
handle_info({Passive, _Sock}, State)
when Passive == tcp_passive; Passive == ssl_passive ->
%% In Stats
Pubs = emqx_pd:reset_counter(incoming_pubs),
Bytes = emqx_pd:reset_counter(incoming_bytes),
InStats = #{cnt => Pubs, oct => Bytes},
%% Ensure Rate Limit
NState = ensure_rate_limit(InStats, State),
%% Run GC and Check OOM
NState1 = check_oom(run_gc(InStats, NState)),
handle_info(activate_socket, NState1);
handle_info({Error, _Sock, Reason}, State)
when Error == tcp_error; Error == ssl_error ->
handle_info({sock_error, Reason}, State);
handle_info({Closed, _Sock}, State)
when Closed == tcp_closed; Closed == ssl_closed ->
handle_info({sock_closed, Closed}, close_socket(State));
handle_info({inet_reply, _Sock, {error, Reason}}, State) ->
handle_info({sock_error, Reason}, State);
handle_info({sock_error, Reason}, State) ->
case Reason =/= closed andalso Reason =/= einval of
true -> ?LOG(warning, "socket_error: ~p", [Reason]);
false -> ok
end,
handle_info({sock_closed, Reason}, close_socket(State));
handle_info({sock_closed, Reason}, State) ->
shutdown(Reason, State);
handle_info({deliver, _Topic, Msg}, State = #state{pstate = PState}) ->
@ -172,8 +354,7 @@ handle_info({deliver, _Topic, Msg}, State = #state{pstate = PState}) ->
end});
handle_info(Info, State) ->
?LOG(error, "Unexpected info: ~p", [Info]),
noreply(State).
with_proto(handle_info, [Info], State).
terminate(Reason, #state{transport = Transport,
socket = Sock,
@ -197,6 +378,8 @@ code_change(_OldVsn, State, _Extra) ->
with_proto(Fun, Args, State = #state{pstate = PState}) ->
case erlang:apply(emqx_stomp_protocol, Fun, Args ++ [PState]) of
ok ->
noreply(State);
{ok, NPState} ->
noreply(State#state{pstate = NPState});
{F, Reason, NPState} when F == stop;
@ -208,13 +391,14 @@ with_proto(Fun, Args, State = #state{pstate = PState}) ->
received(<<>>, State) ->
noreply(State);
received(Bytes, State = #state{parser = Parser,
received(Bytes, State = #state{parser = Parser,
pstate = PState}) ->
try emqx_stomp_frame:parse(Bytes, Parser) of
{more, NewParser} ->
noreply(State#state{parser = NewParser});
{ok, Frame, Rest} ->
?LOG(info, "RECV Frame: ~s", [emqx_stomp_frame:format(Frame)]),
ok = inc_incoming_stats(Frame),
case emqx_stomp_protocol:received(Frame, PState) of
{ok, PState1} ->
received(Rest, reset_parser(State#state{pstate = PState1}));
@ -237,25 +421,97 @@ received(Bytes, State = #state{parser = Parser,
reset_parser(State = #state{proto_env = ProtoEnv}) ->
State#state{parser = emqx_stomp_frame:init_parer_state(ProtoEnv)}.
rate_limit(_Size, State = #state{rate_limit = undefined}) ->
run_socket(State);
rate_limit(Size, State = #state{rate_limit = Rl}) ->
case esockd_rate_limit:check(Size, Rl) of
{0, Rl1} ->
run_socket(State#state{conn_state = running, rate_limit = Rl1});
{Pause, Rl1} ->
?LOG(error, "Rate limiter pause for ~p", [Pause]),
erlang:send_after(Pause, self(), activate_sock),
State#state{conn_state = blocked, rate_limit = Rl1}
activate_socket(State = #state{sockstate = closed}) ->
{ok, State};
activate_socket(State = #state{sockstate = blocked}) ->
{ok, State};
activate_socket(State = #state{transport = Transport,
socket = Socket,
active_n = N}) ->
case Transport:setopts(Socket, [{active, N}]) of
ok -> {ok, State#state{sockstate = running}};
Error -> Error
end.
run_socket(State = #state{conn_state = blocked}) ->
State;
run_socket(State = #state{await_recv = true}) ->
State;
run_socket(State = #state{transport = Transport, socket = Sock}) ->
Transport:async_recv(Sock, 0, infinity),
State#state{await_recv = true}.
close_socket(State = #state{sockstate = closed}) -> State;
close_socket(State = #state{transport = Transport, socket = Socket}) ->
ok = Transport:fast_close(Socket),
State#state{sockstate = closed}.
%%--------------------------------------------------------------------
%% Inc incoming/outgoing stats
inc_incoming_stats(#stomp_frame{command = Cmd}) ->
inc_counter(recv_pkt, 1),
case Cmd of
<<"SEND">> ->
inc_counter(recv_msg, 1),
inc_counter(incoming_pubs, 1),
emqx_metrics:inc('messages.received'),
emqx_metrics:inc('messages.qos1.received');
_ ->
ok
end,
emqx_metrics:inc('packets.received').
inc_outgoing_stats(#stomp_frame{command = Cmd}) ->
inc_counter(send_pkt, 1),
case Cmd of
<<"MESSAGE">> ->
inc_counter(send_msg, 1),
inc_counter(outgoing_pubs, 1),
emqx_metrics:inc('messages.sent'),
emqx_metrics:inc('messages.qos1.sent');
_ ->
ok
end,
emqx_metrics:inc('packets.sent').
%%--------------------------------------------------------------------
%% Ensure rate limit
ensure_rate_limit(Stats, State = #state{limiter = Limiter}) ->
case ?ENABLED(Limiter) andalso emqx_limiter:check(Stats, Limiter) of
false -> State;
{ok, Limiter1} ->
State#state{limiter = Limiter1};
{pause, Time, Limiter1} ->
?LOG(warning, "Pause ~pms due to rate limit", [Time]),
TRef = start_timer(Time, limit_timeout),
State#state{sockstate = blocked,
limiter = Limiter1,
limit_timer = TRef
}
end.
%%--------------------------------------------------------------------
%% Run GC and Check OOM
run_gc(Stats, State = #state{gc_state = GcSt}) ->
case ?ENABLED(GcSt) andalso emqx_gc:run(Stats, GcSt) of
false -> State;
{_IsGC, GcSt1} ->
State#state{gc_state = GcSt1}
end.
check_oom(State) ->
OomPolicy = ?DEFAULT_OOM_POLICY,
?tp(debug, check_oom, #{policy => OomPolicy}),
case ?ENABLED(OomPolicy) andalso emqx_misc:check_oom(OomPolicy) of
{shutdown, Reason} ->
%% triggers terminate/2 callback immediately
erlang:exit({shutdown, Reason});
_Other ->
ok
end,
State.
%%--------------------------------------------------------------------
%% Ensure/cancel stats timer
ensure_stats_timer(Timeout, State = #state{stats_timer = undefined}) ->
State#state{stats_timer = start_timer(Timeout, emit_stats)};
ensure_stats_timer(_Timeout, State) -> State.
getstat(Stat, #state{transport = Transport, socket = Sock}) ->
case Transport:getstat(Sock, [Stat]) of
@ -272,3 +528,9 @@ stop(Reason, State) ->
shutdown(Reason, State) ->
stop({shutdown, Reason}, State).
shutdown_and_reply(Reason, Reply, State) ->
{stop, {shutdown, Reason}, Reply, State}.
inc_counter(Key, Inc) ->
_ = emqx_pd:inc_counter(Key, Inc),
ok.

7
apps/emqx_stomp/src/emqx_stomp_frame.erl
View File

@ -126,6 +126,13 @@ parse(Bytes, #{phase := body, len := Len, state := State}) ->
parse(Bytes, Parser = #{pre := Pre}) ->
parse(<<Pre/binary, Bytes/binary>>, maps:without([pre], Parser));
parse(<<?CR, Rest/binary>>, Parser = #{phase := none}) ->
parse(Rest, Parser);
parse(<<?LF, Rest/binary>>, Parser = #{phase := none}) ->
case byte_size(Rest) of
0 -> {more, Parser};
_ -> parse(Rest, Parser)
end;
parse(<<?CR, ?LF, Rest/binary>>, #{phase := Phase, state := State}) ->
parse(Phase, <<?LF, Rest/binary>>, State);
parse(<<?CR>>, Parser) ->

30
apps/emqx_stomp/src/emqx_stomp_heartbeat.erl
View File

@ -23,9 +23,10 @@
, check/3
, info/1
, interval/2
, reset/3
]).
-record(heartbeater, {interval, statval, repeat}).
-record(heartbeater, {interval, statval, repeat, repeat_max}).
-type name() :: incoming | outgoing.
@ -33,7 +34,6 @@
outgoing => #heartbeater{}
}.
%%--------------------------------------------------------------------
%% APIs
%%--------------------------------------------------------------------
@ -43,19 +43,23 @@ init({0, 0}) ->
#{};
init({Cx, Cy}) ->
maps:filter(fun(_, V) -> V /= undefined end,
#{incoming => heartbeater(Cx),
outgoing => heartbeater(Cy)
#{incoming => heartbeater(incoming, Cx),
outgoing => heartbeater(outgoing, Cy)
}).
heartbeater(0) ->
heartbeater(_, 0) ->
undefined;
heartbeater(I) ->
heartbeater(N, I) ->
#heartbeater{
interval = I,
statval = 0,
repeat = 0
repeat = 0,
repeat_max = repeat_max(N)
}.
repeat_max(incoming) -> 1;
repeat_max(outgoing) -> 0.
-spec check(name(), pos_integer(), heartbeat())
-> {ok, heartbeat()}
| {error, timeout}.
@ -68,11 +72,12 @@ check(Name, NewVal, HrtBt) ->
end.
check(NewVal, HrtBter = #heartbeater{statval = OldVal,
repeat = Repeat}) ->
repeat = Repeat,
repeat_max = Max}) ->
if
NewVal =/= OldVal ->
{ok, HrtBter#heartbeater{statval = NewVal, repeat = 0}};
Repeat < 1 ->
Repeat < Max ->
{ok, HrtBter#heartbeater{repeat = Repeat + 1}};
true -> {error, timeout}
end.
@ -90,3 +95,10 @@ interval(Type, HrtBt) ->
undefined -> undefined;
#heartbeater{interval = Intv} -> Intv
end.
reset(Type, StatVal, HrtBt) ->
case maps:get(Type, HrtBt, undefined) of
undefined -> HrtBt;
HrtBter ->
HrtBt#{Type => HrtBter#heartbeater{statval = StatVal, repeat = 0}}
end.

491
apps/emqx_stomp/src/emqx_stomp_protocol.erl
View File

@ -20,6 +20,7 @@
-include("emqx_stomp.hrl").
-include_lib("emqx/include/emqx.hrl").
-include_lib("emqx/include/types.hrl").
-include_lib("emqx/include/logger.hrl").
-include_lib("emqx/include/emqx_mqtt.hrl").
@ -30,6 +31,8 @@
%% API
-export([ init/2
, info/1
, info/2
, stats/1
]).
-export([ received/2
@ -38,6 +41,9 @@
, timeout/3
]).
-export([ handle_info/2
]).
%% for trans callback
-export([ handle_recv_send_frame/2
, handle_recv_ack_frame/2
@ -45,21 +51,37 @@
]).
-record(pstate, {
peername,
heartfun,
sendfun,
%% Stomp ConnInfo
conninfo :: emqx_types:conninfo(),
%% Stomp ClientInfo
clientinfo :: emqx_types:clientinfo(),
%% Stomp Heartbeats
heart_beats :: maybe(emqx_stomp_hearbeat:heartbeat()),
%% Stomp Connection State
connected = false,
proto_ver,
proto_name,
heart_beats,
login,
allow_anonymous,
default_user,
subscriptions = [],
%% Timers
timers :: #{atom() => disable | undefined | reference()},
transaction :: #{binary() => list()}
%% Transaction
transaction :: #{binary() => list()},
%% Subscriptions
subscriptions = #{},
%% Send function
sendfun :: {function(), list()},
%% Heartbeat function
heartfun :: {function(), list()},
%% Get Socket stat function
statfun :: {function(), list()},
%% The confs for the connection
%% TODO: put these configs into a public mem?
allow_anonymous :: maybe(boolean()),
default_user :: maybe(list())
}).
-define(DEFAULT_SUB_ACK, <<"auto">>).
-define(INCOMING_TIMER_BACKOFF, 1.25).
-define(OUTCOMING_TIMER_BACKOFF, 0.75).
-define(TIMER_TABLE, #{
incoming_timer => incoming,
outgoing_timer => outgoing,
@ -68,34 +90,135 @@
-define(TRANS_TIMEOUT, 60000).
-define(INFO_KEYS, [conninfo, conn_state, clientinfo, session, will_msg]).
-define(STATS_KEYS, [subscriptions_cnt,
subscriptions_max,
inflight_cnt,
inflight_max,
mqueue_len,
mqueue_max,
mqueue_dropped,
next_pkt_id,
awaiting_rel_cnt,
awaiting_rel_max
]).
-dialyzer({nowarn_function, [ check_acl/3
, init/2
]}).
-type(pstate() :: #pstate{}).
%% @doc Init protocol
init(#{peername := Peername,
sendfun := SendFun,
heartfun := HeartFun}, Env) ->
AllowAnonymous = get_value(allow_anonymous, Env, false),
DefaultUser = get_value(default_user, Env),
#pstate{peername = Peername,
heartfun = HeartFun,
sendfun = SendFun,
timers = #{},
transaction = #{},
allow_anonymous = AllowAnonymous,
default_user = DefaultUser}.
init(ConnInfo = #{peername := {PeerHost, _Port},
sockname := {_Host, SockPort},
statfun := StatFun,
sendfun := SendFun,
heartfun := HeartFun}, Opts) ->
info(#pstate{connected = Connected,
proto_ver = ProtoVer,
proto_name = ProtoName,
heart_beats = Heartbeats,
login = Login,
subscriptions = Subscriptions}) ->
[{connected, Connected},
{proto_ver, ProtoVer},
{proto_name, ProtoName},
{heart_beats, Heartbeats},
{login, Login},
{subscriptions, Subscriptions}].
NConnInfo = default_conninfo(ConnInfo),
ClientInfo = #{zone => undefined,
protocol => stomp,
peerhost => PeerHost,
sockport => SockPort,
clientid => undefined,
username => undefined,
mountpoint => undefined, %% XXX: not supported now
is_bridge => false,
is_superuser => false
},
AllowAnonymous = get_value(allow_anonymous, Opts, false),
DefaultUser = get_value(default_user, Opts),
#pstate{
conninfo = NConnInfo,
clientinfo = ClientInfo,
heartfun = HeartFun,
sendfun = SendFun,
statfun = StatFun,
timers = #{},
transaction = #{},
allow_anonymous = AllowAnonymous,
default_user = DefaultUser
}.
default_conninfo(ConnInfo) ->
NConnInfo = maps:without([sendfun, heartfun], ConnInfo),
NConnInfo#{
proto_name => <<"STOMP">>,
proto_ver => <<"1.2">>,
clean_start => true,
clientid => undefined,
username => undefined,
conn_props => [],
connected => false,
connected_at => undefined,
keepalive => undefined,
receive_maximum => 0,
expiry_interval => 0
}.
-spec info(pstate()) -> emqx_types:infos().
info(State) ->
maps:from_list(info(?INFO_KEYS, State)).
-spec info(list(atom())|atom(), pstate()) -> term().
info(Keys, State) when is_list(Keys) ->
[{Key, info(Key, State)} || Key <- Keys];
info(conninfo, #pstate{conninfo = ConnInfo}) ->
ConnInfo;
info(socktype, #pstate{conninfo = ConnInfo}) ->
maps:get(socktype, ConnInfo, undefined);
info(peername, #pstate{conninfo = ConnInfo}) ->
maps:get(peername, ConnInfo, undefined);
info(sockname, #pstate{conninfo = ConnInfo}) ->
maps:get(sockname, ConnInfo, undefined);
info(proto_name, #pstate{conninfo = ConnInfo}) ->
maps:get(proto_name, ConnInfo, undefined);
info(proto_ver, #pstate{conninfo = ConnInfo}) ->
maps:get(proto_ver, ConnInfo, undefined);
info(connected_at, #pstate{conninfo = ConnInfo}) ->
maps:get(connected_at, ConnInfo, undefined);
info(clientinfo, #pstate{clientinfo = ClientInfo}) ->
ClientInfo;
info(zone, _) ->
undefined;
info(clientid, #pstate{clientinfo = ClientInfo}) ->
maps:get(clientid, ClientInfo, undefined);
info(username, #pstate{clientinfo = ClientInfo}) ->
maps:get(username, ClientInfo, undefined);
info(session, State) ->
session_info(State);
info(conn_state, #pstate{connected = true}) ->
connected;
info(conn_state, _) ->
disconnected;
info(will_msg, _) ->
undefined.
session_info(#pstate{conninfo = ConnInfo, subscriptions = Subs}) ->
#{subscriptions => Subs,
upgrade_qos => false,
retry_interval => 0,
await_rel_timeout => 0,
created_at => maps:get(connected_at, ConnInfo, 0)
}.
-spec stats(pstate()) -> emqx_types:stats().
stats(#pstate{subscriptions = Subs}) ->
[{subscriptions_cnt, maps:size(Subs)},
{subscriptions_max, 0},
{inflight_cnt, 0},
{inflight_max, 0},
{mqueue_len, 0},
{mqueue_max, 0},
{mqueue_dropped, 0},
{next_pkt_id, 0},
{awaiting_rel_cnt, 0},
{awaiting_rel_max, 0}].
-spec(received(stomp_frame(), pstate())
-> {ok, pstate()}
@ -105,20 +228,49 @@ received(Frame = #stomp_frame{command = <<"STOMP">>}, State) ->
received(Frame#stomp_frame{command = <<"CONNECT">>}, State);
received(#stomp_frame{command = <<"CONNECT">>, headers = Headers},
State = #pstate{connected = false, allow_anonymous = AllowAnonymous, default_user = DefaultUser}) ->
State = #pstate{connected = false}) ->
case negotiate_version(header(<<"accept-version">>, Headers)) of
{ok, Version} ->
Login = header(<<"login">>, Headers),
Passc = header(<<"passcode">>, Headers),
case check_login(Login, Passc, AllowAnonymous, DefaultUser) of
case check_login(Login, Passc,
allow_anonymous(State),
default_user(State)
) of
true ->
emqx_logger:set_metadata_clientid(Login),
Heartbeats = parse_heartbeats(
header(<<"heart-beat">>, Headers, <<"0,0">>)),
ClientId = emqx_guid:to_base62(emqx_guid:gen()),
emqx_logger:set_metadata_clientid(ClientId),
ConnInfo = State#pstate.conninfo,
ClitInfo = State#pstate.clientinfo,
NConnInfo = ConnInfo#{
proto_ver => Version,
clientid => ClientId,
keepalive => element(1, Heartbeats) div 1000,
username => Login
},
NClitInfo = ClitInfo#{
clientid => ClientId,
username => Login
},
Heartbeats = parse_heartbeats(header(<<"heart-beat">>, Headers, <<"0,0">>)),
NState = start_heartbeart_timer(Heartbeats, State#pstate{connected = true,
proto_ver = Version, login = Login}),
send(connected_frame([{<<"version">>, Version},
{<<"heart-beat">>, reverse_heartbeats(Heartbeats)}]), NState);
ConnPid = self(),
_ = emqx_cm_locker:trans(ClientId, fun(_) ->
emqx_cm:discard_session(ClientId),
emqx_cm:register_channel(ClientId, ConnPid, NConnInfo)
end),
NState = start_heartbeart_timer(
Heartbeats,
State#pstate{
conninfo = NConnInfo,
clientinfo = NClitInfo}
),
ConnectedFrame = connected_frame(
[{<<"version">>, Version},
{<<"heart-beat">>, reverse_heartbeats(Heartbeats)}
]),
send(ConnectedFrame, ensure_connected(NState));
false ->
_ = send(error_frame(undefined, <<"Login or passcode error!">>), State),
{error, login_or_passcode_error, State}
@ -130,6 +282,7 @@ received(#stomp_frame{command = <<"CONNECT">>, headers = Headers},
end;
received(#stomp_frame{command = <<"CONNECT">>}, State = #pstate{connected = true}) ->
?LOG(error, "Received CONNECT frame on connected=true state"),
{error, unexpected_connect, State};
received(Frame = #stomp_frame{command = <<"SEND">>, headers = Headers}, State) ->
@ -139,31 +292,51 @@ received(Frame = #stomp_frame{command = <<"SEND">>, headers = Headers}, State) -
end;
received(#stomp_frame{command = <<"SUBSCRIBE">>, headers = Headers},
State = #pstate{subscriptions = Subscriptions}) ->
State = #pstate{subscriptions = Subs}) ->
Id = header(<<"id">>, Headers),
Topic = header(<<"destination">>, Headers),
Ack = header(<<"ack">>, Headers, <<"auto">>),
{ok, State1} = case lists:keyfind(Id, 1, Subscriptions) of
{Id, Topic, Ack} ->
{ok, State};
false ->
emqx_broker:subscribe(Topic),
{ok, State#pstate{subscriptions = [{Id, Topic, Ack}|Subscriptions]}}
end,
maybe_send_receipt(receipt_id(Headers), State1);
Ack = header(<<"ack">>, Headers, ?DEFAULT_SUB_ACK),
case find_sub_by_id(Id, Subs) of
{Topic, #{sub_props := #{id := Id}}} ->
?LOG(info, "Subscription has established: ~s", [Topic]),
maybe_send_receipt(receipt_id(Headers), State);
{InuseTopic, #{sub_props := #{id := InuseId}}} ->
?LOG(info, "Subscription id ~p inused by topic: ~s, "
"request topic: ~s", [InuseId, InuseTopic, Topic]),
send(error_frame(receipt_id(Headers),
["Request sub-id ", Id, " inused "]), State);
undefined ->
case check_acl(subscribe, Topic, State) of
allow ->
ClientInfo = State#pstate.clientinfo,
[{TopicFilter, SubOpts}] = parse_topic_filters(
[{Topic, ?DEFAULT_SUBOPTS}
]),
NSubOpts = SubOpts#{sub_props => #{id => Id, ack => Ack}},
_ = run_hooks('client.subscribe',
[ClientInfo, _SubProps = #{}],
[{TopicFilter, NSubOpts}]),
NState = do_subscribe(TopicFilter, NSubOpts, State),
maybe_send_receipt(receipt_id(Headers), NState)
end
end;
received(#stomp_frame{command = <<"UNSUBSCRIBE">>, headers = Headers},
State = #pstate{subscriptions = Subscriptions}) ->
State = #pstate{subscriptions = Subs, clientinfo = ClientInfo}) ->
Id = header(<<"id">>, Headers),
{ok, State1} = case lists:keyfind(Id, 1, Subscriptions) of
{Id, Topic, _Ack} ->
ok = emqx_broker:unsubscribe(Topic),
{ok, State#pstate{subscriptions = lists:keydelete(Id, 1, Subscriptions)}};
false ->
{ok, State}
end,
maybe_send_receipt(receipt_id(Headers), State1);
{ok, NState} = case find_sub_by_id(Id, Subs) of
{Topic, #{sub_props := #{id := Id}}} ->
_ = run_hooks('client.unsubscribe',
[ClientInfo, #{}],
[{Topic, #{}}]),
State1 = do_unsubscribe(Topic, ?DEFAULT_SUBOPTS, State),
{ok, State1};
undefined ->
{ok, State}
end,
maybe_send_receipt(receipt_id(Headers), NState);
%% ACK
%% id:12345
@ -239,10 +412,15 @@ received(#stomp_frame{command = <<"DISCONNECT">>, headers = Headers}, State) ->
_ = maybe_send_receipt(receipt_id(Headers), State),
{stop, normal, State}.
send(Msg = #message{topic = Topic, headers = Headers, payload = Payload},
State = #pstate{subscriptions = Subscriptions}) ->
case lists:keyfind(Topic, 2, Subscriptions) of
{Id, Topic, Ack} ->
send(Msg0 = #message{},
State = #pstate{clientinfo = ClientInfo, subscriptions = Subs}) ->
ok = emqx_metrics:inc('messages.delivered'),
Msg = emqx_hooks:run_fold('message.delivered', [ClientInfo], Msg0),
#message{topic = Topic,
headers = Headers,
payload = Payload} = Msg,
case find_sub_by_topic(Topic, Subs) of
{Topic, #{sub_props := #{id := Id, ack := Ack}}} ->
Headers0 = [{<<"subscription">>, Id},
{<<"message-id">>, next_msgid()},
{<<"destination">>, Topic},
@ -256,19 +434,21 @@ send(Msg = #message{topic = Topic, headers = Headers, payload = Payload},
Frame = #stomp_frame{command = <<"MESSAGE">>,
headers = Headers1 ++ maps:get(stomp_headers, Headers, []),
body = Payload},
send(Frame, State);
false ->
undefined ->
?LOG(error, "Stomp dropped: ~p", [Msg]),
{error, dropped, State}
end;
send(Frame, State = #pstate{sendfun = {Fun, Args}}) ->
?LOG(info, "SEND Frame: ~s", [emqx_stomp_frame:format(Frame)]),
Data = emqx_stomp_frame:serialize(Frame),
?LOG(debug, "SEND ~p", [Data]),
erlang:apply(Fun, [Data] ++ Args),
send(Frame, State = #pstate{sendfun = {Fun, Args}}) when is_record(Frame, stomp_frame) ->
erlang:apply(Fun, [Frame] ++ Args),
{ok, State}.
shutdown(Reason, State = #pstate{connected = true}) ->
_ = ensure_disconnected(Reason, State),
ok;
shutdown(_Reason, _State) ->
ok.
@ -283,11 +463,18 @@ timeout(_TRef, {incoming, NewVal},
timeout(_TRef, {outgoing, NewVal},
State = #pstate{heart_beats = HrtBt,
heartfun = {Fun, Args}}) ->
statfun = {StatFun, StatArgs},
heartfun = {Fun, Args}}) ->
case emqx_stomp_heartbeat:check(outgoing, NewVal, HrtBt) of
{error, timeout} ->
_ = erlang:apply(Fun, Args),
{ok, State};
case erlang:apply(StatFun, [send_oct] ++ StatArgs) of
{ok, NewVal2} ->
NHrtBt = emqx_stomp_heartbeat:reset(outgoing, NewVal2, HrtBt),
{ok, reset_timer(outgoing_timer, State#pstate{heart_beats = NHrtBt})};
{error, Reason} ->
{shutdown, {error, {get_stats_error, Reason}}, State}
end;
{ok, NHrtBt} ->
{ok, reset_timer(outgoing_timer, State#pstate{heart_beats = NHrtBt})}
end;
@ -297,6 +484,28 @@ timeout(_TRef, clean_trans, State = #pstate{transaction = Trans}) ->
NTrans = maps:filter(fun(_, {Ts, _}) -> Ts + ?TRANS_TIMEOUT < Now end, Trans),
{ok, ensure_clean_trans_timer(State#pstate{transaction = NTrans})}.
-spec(handle_info(Info :: term(), pstate())
-> ok | {ok, pstate()} | {shutdown, Reason :: term(), pstate()}).
handle_info({subscribe, TopicFilters}, State) ->
NState = lists:foldl(
fun({TopicFilter, SubOpts}, StateAcc = #pstate{subscriptions = Subs}) ->
NSubOpts = enrich_sub_opts(SubOpts, Subs),
do_subscribe(TopicFilter, NSubOpts, StateAcc)
end, State, parse_topic_filters(TopicFilters)),
{ok, NState};
handle_info({unsubscribe, TopicFilters}, State) ->
NState = lists:foldl(fun({TopicFilter, SubOpts}, StateAcc) ->
do_unsubscribe(TopicFilter, SubOpts, StateAcc)
end, State, parse_topic_filters(TopicFilters)),
{ok, NState};
handle_info(Info, State) ->
?LOG(warning, "Unexpected info ~p", [Info]),
{ok, State}.
negotiate_version(undefined) ->
{ok, <<"1.0">>};
negotiate_version(Accepts) ->
@ -312,13 +521,15 @@ negotiate_version(Ver, [AcceptVer|_]) when Ver >= AcceptVer ->
negotiate_version(Ver, [_|T]) ->
negotiate_version(Ver, T).
check_login(undefined, _, AllowAnonymous, _) ->
check_login(Login, _, AllowAnonymous, _)
when Login == <<>>;
Login == undefined ->
AllowAnonymous;
check_login(_, _, _, undefined) ->
false;
check_login(Login, Passcode, _, DefaultUser) ->
case {list_to_binary(get_value(login, DefaultUser)),
list_to_binary(get_value(passcode, DefaultUser))} of
case {iolist_to_binary(get_value(login, DefaultUser)),
iolist_to_binary(get_value(passcode, DefaultUser))} of
{Login, Passcode} -> true;
{_, _ } -> false
end.
@ -396,11 +607,18 @@ receipt_id(Headers) ->
handle_recv_send_frame(#stomp_frame{command = <<"SEND">>, headers = Headers, body = Body}, State) ->
Topic = header(<<"destination">>, Headers),
_ = maybe_send_receipt(receipt_id(Headers), State),
_ = emqx_broker:publish(
make_mqtt_message(Topic, Headers, iolist_to_binary(Body))
),
State.
case check_acl(publish, Topic, State) of
allow ->
_ = maybe_send_receipt(receipt_id(Headers), State),
_ = emqx_broker:publish(
make_mqtt_message(Topic, Headers, iolist_to_binary(Body))
),
State;
deny ->
ErrFrame = error_frame(receipt_id(Headers), <<"Not Authorized">>),
{ok, NState} = send(ErrFrame, State),
NState
end.
handle_recv_ack_frame(#stomp_frame{command = <<"ACK">>, headers = Headers}, State) ->
Id = header(<<"id">>, Headers),
@ -431,7 +649,111 @@ reverse_heartbeats({Cx, Cy}) ->
start_heartbeart_timer(Heartbeats, State) ->
ensure_timer(
[incoming_timer, outgoing_timer],
State#pstate{heart_beats = emqx_stomp_heartbeat:init(Heartbeats)}).
State#pstate{heart_beats = emqx_stomp_heartbeat:init(backoff(Heartbeats))}).
backoff({Cx, Cy}) ->
{erlang:ceil(Cx * ?INCOMING_TIMER_BACKOFF),
erlang:ceil(Cy * ?OUTCOMING_TIMER_BACKOFF)}.
%%--------------------------------------------------------------------
%% pub & sub helpers
parse_topic_filters(TopicFilters) ->
lists:map(fun emqx_topic:parse/1, TopicFilters).
check_acl(PubSub, Topic, State = #pstate{clientinfo = ClientInfo}) ->
case is_acl_enabled(State) andalso
emqx_access_control:check_acl(ClientInfo, PubSub, Topic) of
false -> allow;
Res -> Res
end.
do_subscribe(TopicFilter, SubOpts,
State = #pstate{clientinfo = ClientInfo, subscriptions = Subs}) ->
ClientId = maps:get(clientid, ClientInfo),
_ = emqx_broker:subscribe(TopicFilter, ClientId),
NSubOpts = SubOpts#{is_new => true},
_ = run_hooks('session.subscribed',
[ClientInfo, TopicFilter, NSubOpts]),
send_event_to_self(updated),
State#pstate{subscriptions = maps:put(TopicFilter, SubOpts, Subs)}.
do_unsubscribe(TopicFilter, SubOpts,
State = #pstate{clientinfo = ClientInfo, subscriptions = Subs}) ->
ok = emqx_broker:unsubscribe(TopicFilter),
_ = run_hooks('session.unsubscribe',
[ClientInfo, TopicFilter, SubOpts]),
send_event_to_self(updated),
State#pstate{subscriptions = maps:remove(TopicFilter, Subs)}.
find_sub_by_topic(Topic, Subs) ->
case maps:get(Topic, Subs, undefined) of
undefined -> undefined;
SubOpts -> {Topic, SubOpts}
end.
find_sub_by_id(Id, Subs) ->
Found = maps:filter(fun(_, SubOpts) ->
%% FIXME: datatype??
maps:get(id, maps:get(sub_props, SubOpts, #{}), -1) == Id
end, Subs),
case maps:to_list(Found) of
[] -> undefined;
[Sub|_] -> Sub
end.
is_acl_enabled(_) ->
%% TODO: configs from somewhere
true.
%% automaticly fill the next sub-id and ack if sub-id is absent
enrich_sub_opts(SubOpts0, Subs) ->
SubOpts = maps:merge(?DEFAULT_SUBOPTS, SubOpts0),
SubProps = maps:get(sub_props, SubOpts, #{}),
SubOpts#{sub_props =>
maps:merge(#{id => next_sub_id(Subs),
ack => ?DEFAULT_SUB_ACK}, SubProps)}.
next_sub_id(Subs) ->
Ids = maps:fold(fun(_, SubOpts, Acc) ->
[binary_to_integer(
maps:get(id, maps:get(sub_props, SubOpts, #{}), <<"0">>)) | Acc]
end, [], Subs),
integer_to_binary(lists:max(Ids) + 1).
%%--------------------------------------------------------------------
%% helpers
default_user(#pstate{default_user = DefaultUser}) ->
DefaultUser.
allow_anonymous(#pstate{allow_anonymous = AllowAnonymous}) ->
AllowAnonymous.
ensure_connected(State = #pstate{conninfo = ConnInfo,
clientinfo = ClientInfo}) ->
NConnInfo = ConnInfo#{
connected => true,
connected_at => erlang:system_time(millisecond)
},
send_event_to_self(connected),
ok = run_hooks('client.connected', [ClientInfo, NConnInfo]),
State#pstate{conninfo = NConnInfo,
connected = true
}.
ensure_disconnected(Reason, State = #pstate{conninfo = ConnInfo, clientinfo = ClientInfo}) ->
NConnInfo = ConnInfo#{disconnected_at => erlang:system_time(millisecond)},
ok = run_hooks('client.disconnected', [ClientInfo, Reason, NConnInfo]),
State#pstate{conninfo = NConnInfo, connected = false}.
send_event_to_self(Name) ->
self() ! {event, Name}, ok.
run_hooks(Name, Args) ->
emqx_hooks:run(Name, Args).
run_hooks(Name, Args, Acc) ->
emqx_hooks:run_fold(Name, Args, Acc).
%%--------------------------------------------------------------------
%% Timer
@ -466,3 +788,4 @@ interval(outgoing_timer, #pstate{heart_beats = HrtBt}) ->
emqx_stomp_heartbeat:interval(outgoing, HrtBt);
interval(clean_trans_timer, _) ->
?TRANS_TIMEOUT.

2
apps/emqx_stomp/test/emqx_stomp_SUITE.erl
View File

@ -100,7 +100,7 @@ t_heartbeat(_) ->
{<<"host">>, <<"127.0.0.1:61613">>},
{<<"login">>, <<"guest">>},
{<<"passcode">>, <<"guest">>},
{<<"heart-beat">>, <<"1000,800">>}])),
{<<"heart-beat">>, <<"1000,2000">>}])),
{ok, Data} = gen_tcp:recv(Sock, 0),
{ok, #stomp_frame{command = <<"CONNECTED">>,
headers = _,

3
apps/emqx_stomp/test/emqx_stomp_heartbeat_SUITE.erl
View File

@ -35,8 +35,7 @@ t_check_1(_) ->
{ok, HrtBt1} = emqx_stomp_heartbeat:check(incoming, 0, HrtBt),
{error, timeout} = emqx_stomp_heartbeat:check(incoming, 0, HrtBt1),
{ok, HrtBt2} = emqx_stomp_heartbeat:check(outgoing, 0, HrtBt1),
{error, timeout} = emqx_stomp_heartbeat:check(outgoing, 0, HrtBt2),
{error, timeout} = emqx_stomp_heartbeat:check(outgoing, 0, HrtBt1),
ok.
t_check_2(_) ->

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

@ -1,6 +1,6 @@
{application, emqx_web_hook,
[{description, "EMQ X WebHook Plugin"},
{vsn, "4.3.6"}, % strict semver, bump manually!
{vsn, "4.3.7"}, % strict semver, bump manually!
{modules, []},
{registered, [emqx_web_hook_sup]},
{applications, [kernel,stdlib,ehttpc]},

2
bin/emqx
View File

@ -27,7 +27,6 @@ export EMU="beam"
export PROGNAME="erl"
DYNLIBS_DIR="$RUNNER_ROOT_DIR/dynlibs"
ERTS_LIB_DIR="$ERTS_DIR/../lib"
MNESIA_DATA_DIR="$RUNNER_DATA_DIR/mnesia/$NAME"
# Echo to stderr on errors
echoerr() { echo "$*" 1>&2; }
@ -409,6 +408,7 @@ case $NAME in
NAME=$NAME@$(relx_get_nodename)
;;
esac
MNESIA_DATA_DIR="$RUNNER_DATA_DIR/mnesia/$NAME"
# Check the first argument for instructions
case "$1" in

2
lib-ce/emqx_dashboard/src/emqx_dashboard.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_dashboard,
[{description, "EMQ X Web Dashboard"},
{vsn, "4.3.5"}, % strict semver, bump manually!
{vsn, "4.3.6"}, % strict semver, bump manually!
{modules, []},
{registered, [emqx_dashboard_sup]},
{applications, [kernel,stdlib,mnesia,minirest]},

2
rebar.config
View File

@ -43,7 +43,7 @@
, {jiffy, {git, "https://github.com/emqx/jiffy", {tag, "1.0.5"}}}
, {cowboy, {git, "https://github.com/emqx/cowboy", {tag, "2.8.2"}}}
, {esockd, {git, "https://github.com/emqx/esockd", {tag, "5.8.0"}}}
, {ekka, {git, "https://github.com/emqx/ekka", {tag, "0.8.1.3"}}}
, {ekka, {git, "https://github.com/emqx/ekka", {tag, "0.8.1.4"}}}
, {gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "2.6.0"}}}
, {cuttlefish, {git, "https://github.com/emqx/cuttlefish", {tag, "v3.3.6"}}}
, {minirest, {git, "https://github.com/emqx/minirest", {tag, "0.3.7"}}}

40
scripts/elvis-check.sh
View File

@ -5,16 +5,16 @@
set -euo pipefail
ELVIS_VERSION='1.0.0-emqx-2'
elvis_version='1.0.0-emqx-2'
base="${1:-}"
repo="${2:-emqx/emqx}"
REPO="${GITHUB_REPOSITORY:-${repo}}"
if [ "${base}" = "" ]; then
echo "Usage $0 <git-compare-base-ref>"
exit 1
fi
elvis_version="${2:-$ELVIS_VERSION}"
echo "elvis -v: $elvis_version"
echo "git diff base: $base"
@ -27,11 +27,7 @@ if [[ "$base" =~ [0-9a-f]{8,40} ]]; then
# base is a commit sha1
compare_base="$base"
else
if [[ $CI == true ]];then
remote="$(git remote -v | grep -E "github\.com(.|/)$GITHUB_REPOSITORY" | grep fetch | awk '{print $1}')"
else
remote="$(git remote -v | grep -E 'github\.com(.|/)emqx' | grep fetch | awk '{print $1}')"
fi
remote="$(git remote -v | grep -E "github\.com(:|/)$REPO((\.git)|(\s))" | grep fetch | awk '{print $1}')"
git fetch "$remote" "$base"
compare_base="$remote/$base"
fi
@ -58,3 +54,31 @@ if [ $bad_file_count -gt 0 ]; then
echo "elvis: $bad_file_count errors"
exit 1
fi
### now check new-line at EOF for changed files
nl_at_eof() {
local file="$1"
if ! [ -f "$file" ]; then
return
fi
case "$file" in
*.png|*rebar3)
return
;;
esac
local lastbyte
lastbyte="$(tail -c 1 "$file" 2>&1)"
if [ "$lastbyte" != '' ]; then
echo "$file"
return 1
fi
}
for file in $(git_diff); do
if ! nl_at_eof "$file"; then
bad_file_count=$(( bad_file_count + 1 ))
fi
done
exit $bad_file_count

23
scripts/update_appup.escript
View File

@ -21,12 +21,11 @@ support other repos too.
Usage:
update_appup.escript [--check] [--repo URL] [--remote NAME] [--skip-build] [--make-commad SCRIPT] [--release-dir DIR] <current_release_tag>
update_appup.escript [--check] [--repo URL] [--remote NAME] [--skip-build] [--make-commad SCRIPT] [--release-dir DIR] <previous_release_tag>
Options:
--check Don't update the appfile, just check that they are complete
--prev-tag Specify the previous release tag. Otherwise the previous patch version is used
--repo Upsteam git repo URL
--remote Get upstream repo URL from the specified git remote
--skip-build Don't rebuild the releases. May produce wrong results
@ -34,7 +33,7 @@ Options:
--release-dir Release directory
--src-dirs Directories where source code is found. Defaults to '{src,apps,lib-*}/**/'
--binary-rel-url Binary release URL pattern. %TAG% variable is substituted with the release tag.
E.g. \"https://github.com/emqx/emqx/releases/download/v4.3.8/emqx-centos7-%TAG%-amd64.zip\"
E.g. \"https://github.com/emqx/emqx/releases/download/v%TAG%/emqx-centos7-%TAG%-amd64.zip\"
".
-record(app,
@ -60,18 +59,12 @@ ignored_apps() ->
[emqx_dashboard, emqx_management] ++ otp_standard_apps().
main(Args) ->
#{current_release := CurrentRelease} = Options = parse_args(Args, default_options()),
#{prev_tag := Baseline} = Options = parse_args(Args, default_options()),
init_globals(Options),
case find_prev_tag(CurrentRelease) of
{ok, Baseline} ->
main(Options, Baseline);
undefined ->
log("No appup update is needed for this release, nothing to be done~n", []),
ok
end.
main(Options, Baseline).
parse_args([CurrentRelease = [A|_]], State) when A =/= $- ->
State#{current_release => CurrentRelease};
parse_args([PrevTag = [A|_]], State) when A =/= $- ->
State#{prev_tag => PrevTag};
parse_args(["--check"|Rest], State) ->
parse_args(Rest, State#{check => true});
parse_args(["--skip-build"|Rest], State) ->
@ -86,8 +79,6 @@ parse_args(["--release-dir", Dir|Rest], State) ->
parse_args(Rest, State#{beams_dir => Dir});
parse_args(["--src-dirs", Pattern|Rest], State) ->
parse_args(Rest, State#{src_dirs => Pattern});
parse_args(["--prev-tag", Tag|Rest], State) ->
parse_args(Rest, State#{prev_tag => Tag});
parse_args(["--binary-rel-url", URL|Rest], State) ->
parse_args(Rest, State#{binary_rel_url => {ok, URL}});
parse_args(_, _) ->
@ -163,7 +154,7 @@ download_prev_release(Tag, #{binary_rel_url := {ok, URL0}, clone_url := Repo}) -
Dir = filename:basename(Repo, ".git") ++ [$-|Tag],
Filename = filename:join(BaseDir, Dir),
Script = "mkdir -p ${OUTFILE} &&
{ [ -f ${OUTFILE}.zip ] || wget -O ${OUTFILE}.zip ${URL}; } &&
wget -O ${OUTFILE}.zip ${URL} &&
unzip -n -d ${OUTFILE} ${OUTFILE}.zip",
Env = [{"TAG", Tag}, {"OUTFILE", Filename}, {"URL", URL}],
bash(Script, Env),

2
src/emqx.app.src
View File

@ -1,7 +1,7 @@
{application, emqx,
[{id, "emqx"},
{description, "EMQ X"},
{vsn, "4.3.10"}, % strict semver, bump manually!
{vsn, "4.3.11"}, % strict semver, bump manually!
{modules, []},
{registered, []},
{applications, [kernel,stdlib,gproc,gen_rpc,esockd,cowboy,sasl,os_mon]},

48
src/emqx.appup.src
View File

@ -1,7 +1,9 @@
%% -*- mode: erlang -*-
{VSN,
[{"4.3.9",
[{load_module,emqx_channel,brutal_purge,soft_purge,[]},
[{"4.3.10",[{load_module,emqx_connection,brutal_purge,soft_purge,[]}]},
{"4.3.9",
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_channel,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_pqueue,brutal_purge,soft_purge,[]},
@ -10,7 +12,8 @@
{load_module,emqx_rpc,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.3.8",
[{load_module,emqx_channel,brutal_purge,soft_purge,[]},
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_channel,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_pqueue,brutal_purge,soft_purge,[]},
@ -19,7 +22,8 @@
{load_module,emqx_rpc,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.3.7",
[{load_module,emqx_channel,brutal_purge,soft_purge,[]},
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_channel,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]},
@ -30,7 +34,8 @@
{load_module,emqx_rpc,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.3.6",
[{load_module,emqx_channel,brutal_purge,soft_purge,[]},
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_channel,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]},
@ -42,7 +47,8 @@
{load_module,emqx_rpc,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.3.5",
[{load_module,emqx_channel,brutal_purge,soft_purge,[]},
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_channel,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]},
{load_module,emqx_misc,brutal_purge,soft_purge,[]},
@ -55,7 +61,8 @@
{load_module,emqx_rpc,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.3.4",
[{load_module,emqx_channel,brutal_purge,soft_purge,[]},
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_channel,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]},
{load_module,emqx_misc,brutal_purge,soft_purge,[]},
@ -69,7 +76,8 @@
{load_module,emqx_rpc,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.3.3",
[{load_module,emqx_channel,brutal_purge,soft_purge,[]},
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_channel,brutal_purge,soft_purge,[]},
{load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]},
{load_module,emqx_misc,brutal_purge,soft_purge,[]},
{load_module,emqx_packet,brutal_purge,soft_purge,[]},
@ -147,8 +155,10 @@
{load_module,emqx_rpc,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{<<".*">>,[]}],
[{"4.3.9",
[{load_module,emqx_channel,brutal_purge,soft_purge,[]},
[{"4.3.10",[{load_module,emqx_connection,brutal_purge,soft_purge,[]}]},
{"4.3.9",
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_channel,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_pqueue,brutal_purge,soft_purge,[]},
@ -157,7 +167,8 @@
{load_module,emqx_rpc,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.3.8",
[{load_module,emqx_channel,brutal_purge,soft_purge,[]},
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_channel,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_pqueue,brutal_purge,soft_purge,[]},
@ -166,7 +177,8 @@
{load_module,emqx_rpc,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.3.7",
[{load_module,emqx_channel,brutal_purge,soft_purge,[]},
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_channel,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_pqueue,brutal_purge,soft_purge,[]},
@ -177,7 +189,8 @@
{load_module,emqx_rpc,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.3.6",
[{load_module,emqx_channel,brutal_purge,soft_purge,[]},
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_channel,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_pqueue,brutal_purge,soft_purge,[]},
@ -189,7 +202,8 @@
{load_module,emqx_rpc,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.3.5",
[{load_module,emqx_channel,brutal_purge,soft_purge,[]},
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_channel,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_pqueue,brutal_purge,soft_purge,[]},
{load_module,emqx_mqueue,brutal_purge,soft_purge,[]},
@ -202,7 +216,8 @@
{load_module,emqx_rpc,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.3.4",
[{load_module,emqx_channel,brutal_purge,soft_purge,[]},
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_channel,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_pqueue,brutal_purge,soft_purge,[]},
{load_module,emqx_mqueue,brutal_purge,soft_purge,[]},
@ -216,7 +231,8 @@
{load_module,emqx_rpc,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.3.3",
[{load_module,emqx_channel,brutal_purge,soft_purge,[]},
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_channel,brutal_purge,soft_purge,[]},
{load_module,emqx_pqueue,brutal_purge,soft_purge,[]},
{load_module,emqx_mqueue,brutal_purge,soft_purge,[]},
{load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]},

2
src/emqx_connection.erl
View File

@ -702,7 +702,7 @@ send(IoData, #state{transport = Transport, socket = Socket, channel = Channel})
ok = emqx_metrics:inc('bytes.sent', Oct),
inc_counter(outgoing_bytes, Oct),
emqx_congestion:maybe_alarm_conn_congestion(Socket, Transport, Channel),
case Transport:async_send(Socket, IoData, [nosuspend]) of
case Transport:async_send(Socket, IoData, []) of
ok -> ok;
Error = {error, _Reason} ->
%% Send an inet_reply to postpone handling the error

5
src/emqx_types.erl
View File

@ -94,7 +94,10 @@
-type(ver() :: ?MQTT_PROTO_V3
| ?MQTT_PROTO_V4
| ?MQTT_PROTO_V5
| non_neg_integer()).
| non_neg_integer()
%% Some non-MQTT versions of protocol may be a binary type
| binary()
).
-type(qos() :: ?QOS_0 | ?QOS_1 | ?QOS_2).
-type(qos_name() :: qos0 | at_most_once |