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

Merge pull request #4088 from emqx/dev/v4.3.0 

Auto-pull-request-on-2021-01-27
This commit is contained in:
Zaiming Shi 2021-01-27 22:02:52 +01:00 committed by GitHub
parent fa7d96dd4a e6235d3bc9
commit c422b200be
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 204 additions and 256 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -55,10 +55,10 @@ translate_env(EnvName) ->
URL = proplists:get_value(url, Req),
#{host := Host0,
path := Path0,
scheme := Scheme} = URIMap = uri_string:parse(add_default_scheme(URL)),
scheme := Scheme} = URIMap = uri_string:parse(add_default_scheme(uri_string:normalize(URL))),
Port = maps:get(port, URIMap, case Scheme of
"https" -> 443;
_ -> 80
"http" -> 80
end),
Path = path(Path0),
{Inet, Host} = parse_host(Host0),
@ -66,7 +66,7 @@ translate_env(EnvName) ->
"http" ->
[{transport_opts, [Inet]}];
"https" ->
CACertFile = application:get_env(?APP, cafile, undefined),
CACertFile = application:get_env(?APP, cacertfile, undefined),
CertFile = application:get_env(?APP, certfile, undefined),
KeyFile = application:get_env(?APP, keyfile, undefined),
TLSOpts = lists:filter(fun({_K, V}) when V =:= <<>> ->

1
apps/emqx_auth_http/src/emqx_auth_http_cli.erl
View File

@ -80,7 +80,6 @@ feedvar(Params, ClientInfo = #{clientid := ClientId,
({Param, "%A"}) -> {Param, maps:get(access, ClientInfo, null)};
({Param, "%t"}) -> {Param, maps:get(topic, ClientInfo, null)};
({Param, "%m"}) -> {Param, maps:get(mountpoint, ClientInfo, null)};
({Param, "%k"}) -> {Param, emqx_json:encode(maps:get(ws_cookie, ClientInfo, null))};
({Param, Var}) -> {Param, Var}
end, Params).

12
apps/emqx_auth_http/test/emqx_auth_http_SUITE.erl
View File

@ -66,7 +66,7 @@ set_special_configs(emqx, _Schmea, _Inet) ->
set_special_configs(emqx_auth_http, Schema, Inet) ->
ServerAddr = http_server(Schema, Inet),
AuthReq = #{method => post,
AuthReq = #{method => get,
url => ServerAddr ++ "/mqtt/auth",
headers => [{"content-type", "application/json"}],
params => [{"clientid", "%c"}, {"username", "%u"}, {"password", "%P"}]},
@ -88,7 +88,7 @@ set_special_configs(emqx_auth_http, Schema, Inet) ->
%% @private
set_https_client_opts() ->
SSLOpt = emqx_ct_helpers:client_ssl_twoway(),
application:set_env(emqx_auth_http, cafile, proplists:get_value(cacertfile, SSLOpt, undefined)),
application:set_env(emqx_auth_http, cacertfile, proplists:get_value(cacertfile, SSLOpt, undefined)),
application:set_env(emqx_auth_http, certfile, proplists:get_value(certfile, SSLOpt, undefined)),
application:set_env(emqx_auth_http, keyfile, proplists:get_value(keyfile, SSLOpt, undefined)).
@ -171,11 +171,3 @@ t_comment_config(_) ->
?assertEqual([], emqx_hooks:lookup('client.authenticate')),
?assertEqual(AuthCount - 1, length(emqx_hooks:lookup('client.authenticate'))),
?assertEqual(AclCount - 1, length(emqx_hooks:lookup('client.check_acl'))).
t_feedvar(_) ->
Params = [{"cookie", "%k"}],
User0 = ?USER(<<"client1">>, <<"testuser">>, mqtt, {127,0,0,1}, external),
?assertEqual([{"cookie", <<"null">>}], emqx_auth_http_cli:feedvar(Params, User0)),
User1 = User0#{ws_cookie => [{<<"k">>, <<"v">>}]},
?assertEqual([{"cookie", <<"{\"k\":\"v\"}">>}], emqx_auth_http_cli:feedvar(Params, User1)).

26
apps/emqx_auth_ldap/.ci/docker-compose.yml
View File

@ -1,26 +0,0 @@
version: '3'
services:
erlang:
image: erlang:22.1
volumes:
- ../:/emqx_auth_ldap
networks:
- emqx_bridge
depends_on:
- ldap_server
tty: true
ldap_server:
build: ./emqx-ldap
image: emqx-ldap:1.0
restart: always
ports:
- 389:389
- 636:636
networks:
- emqx_bridge
networks:
emqx_bridge:
driver: bridge

26
apps/emqx_auth_ldap/.ci/emqx-ldap/Dockerfile
View File

@ -1,26 +0,0 @@
FROM buildpack-deps:stretch
ENV VERSION=2.4.50
RUN apt-get update && apt-get install -y groff groff-base
RUN wget ftp://ftp.openldap.org/pub/OpenLDAP/openldap-release/openldap-${VERSION}.tgz \
&& gunzip -c openldap-${VERSION}.tgz | tar xvfB - \
&& cd openldap-${VERSION} \
&& ./configure && make depend && make && make install \
&& cd .. && rm -rf openldap-${VERSION}
COPY ./slapd.conf /usr/local/etc/openldap/slapd.conf
COPY ./emqx.io.ldif /usr/local/etc/openldap/schema/emqx.io.ldif
COPY ./emqx.schema /usr/local/etc/openldap/schema/emqx.schema
COPY ./*.pem /usr/local/etc/openldap/
RUN mkdir -p /usr/local/etc/openldap/data \
&& slapadd -l /usr/local/etc/openldap/schema/emqx.io.ldif -f /usr/local/etc/openldap/slapd.conf
WORKDIR /usr/local/etc/openldap
EXPOSE 389 636
ENTRYPOINT ["/usr/local/libexec/slapd", "-h", "ldap:/// ldaps:///", "-d", "3", "-f", "/usr/local/etc/openldap/slapd.conf"]
CMD []

16
apps/emqx_auth_ldap/.ci/emqx-ldap/slapd.conf
View File

@ -1,16 +0,0 @@
include /usr/local/etc/openldap/schema/core.schema
include /usr/local/etc/openldap/schema/cosine.schema
include /usr/local/etc/openldap/schema/inetorgperson.schema
include /usr/local/etc/openldap/schema/ppolicy.schema
include /usr/local/etc/openldap/schema/emqx.schema
TLSCACertificateFile /usr/local/etc/openldap/cacert.pem
TLSCertificateFile /usr/local/etc/openldap/cert.pem
TLSCertificateKeyFile /usr/local/etc/openldap/key.pem
database bdb
suffix "dc=emqx,dc=io"
rootdn "cn=root,dc=emqx,dc=io"
rootpw {SSHA}eoF7NhNrejVYYyGHqnt+MdKNBh4r1w3W
directory /usr/local/etc/openldap/data

2
apps/emqx_coap/src/emqx_coap_mqtt_adapter.erl
View File

@ -50,8 +50,6 @@
-record(state, {peername, clientid, username, password, sub_topics = [], connected_at}).
-type(state() :: #state{}).
-define(ALIVE_INTERVAL, 20000).
-define(CONN_STATS, [recv_pkt, recv_msg, send_pkt, send_msg]).

65
apps/emqx_rule_engine/src/emqx_rule_engine.erl
View File

@ -249,13 +249,13 @@ create_resource(#{type := Type, config := Config0} = Params) ->
update_resource(ResId, NewParams) ->
try
lists:foreach(fun(#rule{id = RuleId, enabled = Enabled, actions = Actions}) ->
lists:foreach(
fun (#action_instance{args = #{<<"$resource">> := ResId1}})
when ResId =:= ResId1, Enabled == true ->
throw({dependency_exists, RuleId});
(_) -> ok
end, Actions)
end, ets:tab2list(?RULE_TAB)),
lists:foreach(
fun (#action_instance{args = #{<<"$resource">> := ResId1}})
when ResId =:= ResId1, Enabled =:= true ->
throw({dependency_exists, RuleId});
(_) -> ok
end, Actions)
end, ets:tab2list(?RULE_TAB)),
do_update_resource_check(ResId, NewParams)
catch _ : Reason ->
{error, Reason}
@ -267,37 +267,42 @@ do_update_resource_check(Id, NewParams) ->
type = Type,
config = OldConfig,
description = OldDescription} = _OldResource} ->
try
do_update_resource(#{id => Id,
config => case maps:find(<<"config">>, NewParams) of
{ok, NewConfig} -> NewConfig;
error -> OldConfig
end,
type => Type,
description => case maps:find(<<"description">>, NewParams) of
{ok, NewDescription} -> NewDescription;
error -> OldDescription
end}),
ok
catch _ : Reason ->
{error, Reason}
end;
try
do_update_resource(#{id => Id,
config => case maps:find(<<"config">>, NewParams) of
{ok, NewConfig} -> NewConfig;
error -> OldConfig
end,
type => Type,
description => case maps:find(<<"description">>, NewParams) of
{ok, NewDescription} -> NewDescription;
error -> OldDescription
end}),
ok
catch _ : Reason ->
{error, Reason}
end;
_Other ->
{error, not_found}
end.
do_update_resource(#{id := Id, type := Type, description:= NewDescription, config:= NewConfig}) ->
do_update_resource(#{id := Id, type := Type, description := NewDescription, config := NewConfig}) ->
case emqx_rule_registry:find_resource_type(Type) of
{ok, #resource_type{on_create = {Module, Create},
on_destroy = {Module, Destroy},
params_spec = ParamSpec}} ->
Config = emqx_rule_validator:validate_params(NewConfig, ParamSpec),
cluster_call(init_resource, [Module, Create, Id, Config]),
emqx_rule_registry:add_resource(#resource{id = Id,
type = Type,
config = Config,
description = NewDescription,
created_at = erlang:system_time(millisecond)})
case test_resource(#{type => Type, config => NewConfig}) of
ok ->
Resource = #resource{id = Id,
type = Type,
config = Config,
description = NewDescription,
created_at = erlang:system_time(millisecond)},
cluster_call(init_resource, [Module, Create, Id, Config]),
emqx_rule_registry:add_resource(Resource);
{error, Reason} ->
{error, Reason}
end
end.
-spec(start_resource(resource_id()) -> ok | {error, Reason :: term()}).

10
apps/emqx_rule_engine/src/emqx_rule_engine_api.erl
View File

@ -329,13 +329,13 @@ start_resource(#{id := Id}, _Params) ->
update_resource(#{id := Id}, NewParams) ->
P1 = case proplists:get_value(<<"description">>, NewParams) of
undefined -> #{};
Value -> #{<<"description">> => Value}
undefined -> #{};
Value -> #{<<"description">> => Value}
end,
P2 = case proplists:get_value(<<"config">>, NewParams) of
undefined -> #{};
<<"{}">> -> #{};
Map -> #{<<"config">> => ?RAISE(maps:from_list(Map), {invalid_config, Map})}
undefined -> #{};
<<"{}">> -> #{};
Map -> #{<<"config">> => ?RAISE(maps:from_list(Map), {invalid_config, Map})}
end,
case emqx_rule_engine:update_resource(Id, maps:merge(P1, P2)) of
ok ->

10
apps/emqx_rule_engine/src/emqx_rule_engine_cli.erl
View File

@ -167,7 +167,7 @@ resources(["create" | Params]) ->
resources(["update" | Params]) ->
with_opts(fun({Opts, _}) ->
Id = maps:get(id, maps:from_list(Opts)),
Id = proplists:get_value(id, Opts),
Maps = make_updated_resource(Opts),
case emqx_rule_engine:update_resource(Id, Maps) of
ok ->
@ -324,12 +324,12 @@ make_resource(Opts) ->
make_updated_resource(Opts) ->
P1 = case proplists:get_value(description, Opts) of
undefined -> #{};
Value -> #{<<"description">> => Value}
undefined -> #{};
Value -> #{<<"description">> => Value}
end,
P2 = case proplists:get_value(config, Opts) of
undefined -> #{};
Map -> #{<<"config">> => ?RAISE((emqx_json:decode(Map, [return_maps])), {invalid_config, Map})}
undefined -> #{};
Map -> #{<<"config">> => ?RAISE((emqx_json:decode(Map, [return_maps])), {invalid_config, Map})}
end,
maps:merge(P1, P2).

8
apps/emqx_web_hook/src/emqx_web_hook_app.erl
View File

@ -72,11 +72,9 @@ translate_env() ->
true -> verify_peer;
false -> verify_none
end,
TLSOpts = lists:filter(fun({_K, V}) when V =:= <<>> ->
false;
(_) ->
true
end, [{keyfile, KeyFile}, {certfile, CertFile}, {cacertfile, CACertFile}]),
TLSOpts = lists:filter(fun({_K, V}) ->
V /= <<>> andalso V /= undefined andalso V /= "" andalso true
end, [{keyfile, KeyFile}, {certfile, CertFile}, {cacertfile, CACertFile}]),
TlsVers = ['tlsv1.2','tlsv1.1',tlsv1],
NTLSOpts = [{verify, VerifyType},
{versions, TlsVers},

97
apps/emqx_web_hook/test/emqx_web_hook_SUITE.erl
View File

@ -47,49 +47,38 @@ init_per_group(Name, Config) ->
set_special_cfgs(),
case Name of
http ->
http_server:start_http(),
emqx_ct_helpers:start_apps([emqx_web_hook], fun set_special_configs_http/1);
https ->
http_server:start_https(),
emqx_ct_helpers:start_apps([emqx_web_hook], fun set_special_configs_https/1);
ipv6http ->
http_server:start_http(),
emqx_ct_helpers:start_apps([emqx_web_hook], fun set_special_configs_ipv6_http/1);
ipv6https ->
http_server:start_https(),
emqx_ct_helpers:start_apps([emqx_web_hook], fun set_special_configs_ipv6_https/1)
end,
Config.
end_per_group(Name, Config) ->
case lists:member(Name,[http, ipv6http]) of
true ->
http_server:stop_http();
_ ->
http_server:stop_https()
end,
end_per_group(_Name, Config) ->
emqx_ct_helpers:stop_apps([emqx_web_hook]),
Config.
set_special_configs_http(_) ->
ok.
application:set_env(emqx_web_hook, url, "http://127.0.0.1:9999").
set_special_configs_https(_) ->
Path = emqx_ct_helpers:deps_path(emqx_web_hook, "test/emqx_web_hook_SUITE_data/"),
SslOpts = [{keyfile, Path ++ "/client-key.pem"},
{certfile, Path ++ "/client-cert.pem"},
{cacertfile, Path ++ "/ca.pem"}],
{cafile, Path ++ "/ca.pem"}],
application:set_env(emqx_web_hook, ssl, true),
application:set_env(emqx_web_hook, ssloptions, SslOpts),
application:set_env(emqx_web_hook, url, "https://127.0.0.1:8081").
application:set_env(emqx_web_hook, url, "https://127.0.0.1:8888").
set_special_configs_ipv6_http(N) ->
set_special_configs_http(N),
application:set_env(emqx_web_hook, url, "http://[::1]:8080").
set_special_configs_ipv6_http(_) ->
application:set_env(emqx_web_hook, url, "http://[::1]:9999").
set_special_configs_ipv6_https(N) ->
set_special_configs_https(N),
application:set_env(emqx_web_hook, url, "https://[::1]:8081").
application:set_env(emqx_web_hook, url, "https://[::1]:8888").
set_special_cfgs() ->
AllRules = [{"message.acked", "{\"action\": \"on_message_acked\"}"},
@ -109,6 +98,27 @@ set_special_cfgs() ->
%% Test cases
%%--------------------------------------------------------------------
t_valid(Config) ->
{ok, ServerPid} = http_server:start_link(),
application:set_env(emqx_web_hook, headers, [{"k1","K1"}, {"k2", "K2"}]),
{ok, C} = emqtt:start_link([ {clientid, <<"simpleClient">>}
, {proto_ver, v5}
, {keepalive, 60}
]),
{ok, _} = emqtt:connect(C),
emqtt:subscribe(C, <<"TopicA">>, qos2),
emqtt:publish(C, <<"TopicA">>, <<"Payload...">>, qos2),
emqtt:unsubscribe(C, <<"TopicA">>),
emqtt:disconnect(C),
[begin
Maps = emqx_json:decode(P, [return_maps]),
validate_hook_resp(Maps),
validate_hook_headers(Headers)
end
|| {{P, _Bool}, Headers} <- http_server:get_received_data()],
http_server:stop(ServerPid),
Config.
t_check_hooked(_) ->
{ok, Rules} = application:get_env(emqx_web_hook, rules),
lists:foreach(fun({HookName, _Action}) ->
@ -127,90 +137,69 @@ t_change_config(_) ->
application:set_env(emqx_web_hook, rules, Rules),
emqx_web_hook:load().
t_valid() ->
application:set_env(emqx_web_hook, headers, [{"k1","K1"}, {"k2", "K2"}]),
{ok, C} = emqtt:start_link([ {clientid, <<"simpleClient">>}
, {proto_ver, v5}
, {keepalive, 60}
]),
{ok, _} = emqtt:connect(C),
emqtt:subscribe(C, <<"TopicA">>, qos2),
emqtt:publish(C, <<"TopicA">>, <<"Payload...">>, qos2),
emqtt:unsubscribe(C, <<"TopicA">>),
emqtt:disconnect(C),
{Params, Headers} = get_http_message(),
[validate_hook_resp(A) || A <- Params],
?assertEqual(<<"K1">>, maps:get(<<"k1">>, Headers)),
?assertEqual(<<"K2">>, maps:get(<<"k2">>, Headers)).
%%--------------------------------------------------------------------
%% Utils
%%--------------------------------------------------------------------
get_http_message() ->
receive
{Params, Headers} ->
L = [B || {B, _} <- Params],
{lists:reverse([emqx_json:decode(E, [return_maps]) || E <- L]), Headers}
after 500 ->
{null, null}
end.
validate_hook_headers(Headers) ->
?assertEqual(<<"K1">>, maps:get(<<"k1">>, Headers)),
?assertEqual(<<"K2">>, maps:get(<<"k2">>, Headers)).
validate_hook_resp(Body = ?ACTION(<<"client_connect">>)) ->
?assertEqual(5, maps:get(<<"proto_ver">>, Body)),
?assertEqual(60, maps:get(<<"keepalive">>, Body)),
?assertEqual(<<"127.0.0.1">>, maps:get(<<"ipaddress">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)),
?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"client_connack">>)) ->
?assertEqual(5, maps:get(<<"proto_ver">>, Body)),
?assertEqual(60, maps:get(<<"keepalive">>, Body)),
?assertEqual(<<"success">>, maps:get(<<"conn_ack">>, Body)),
?assertEqual(<<"127.0.0.1">>, maps:get(<<"ipaddress">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)),
?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"client_connected">>)) ->
_ = maps:get(<<"connected_at">>, Body),
?assertEqual(5, maps:get(<<"proto_ver">>, Body)),
?assertEqual(60, maps:get(<<"keepalive">>, Body)),
?assertEqual(<<"127.0.0.1">>, maps:get(<<"ipaddress">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)),
?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"client_disconnected">>)) ->
?assertEqual(<<"normal">>, maps:get(<<"reason">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)),
?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"client_subscribe">>)) ->
_ = maps:get(<<"opts">>, Body),
?assertEqual(<<"TopicA">>, maps:get(<<"topic">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)),
?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"client_unsubscribe">>)) ->
_ = maps:get(<<"opts">>, Body),
?assertEqual(<<"TopicA">>, maps:get(<<"topic">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)),
?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"session_subscribed">>)) ->
_ = maps:get(<<"opts">>, Body),
?assertEqual(<<"TopicA">>, maps:get(<<"topic">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)),
?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"session_unsubscribed">>)) ->
?assertEqual(<<"TopicA">>, maps:get(<<"topic">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)),
?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"session_terminated">>)) ->
?assertEqual(<<"normal">>, maps:get(<<"reason">>, Body)),
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)),
?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_username_clientid(Body);
validate_hook_resp(Body = ?ACTION(<<"message_publish">>)) ->
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)),
?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_messages_attrs(Body);
validate_hook_resp(Body = ?ACTION(<<"message_delivered">>)) ->
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)),
?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_messages_attrs(Body);
validate_hook_resp(Body = ?ACTION(<<"message_acked">>)) ->
?assertEqual(<<"emqx@127.0.0.1">>, maps:get(<<"node">>, Body)),
?assertEqual(<<"test@127.0.0.1">>, maps:get(<<"node">>, Body)),
assert_messages_attrs(Body).
assert_username_clientid(#{<<"clientid">> := ClientId, <<"username">> := Username}) ->

124
apps/emqx_web_hook/test/http_server.erl
View File

@ -3,73 +3,103 @@
%%
%% It will deliver the http-request params to initialer process
%%--------------------------------------------------------------------
%%
%% Author:wwhai
%%
-module(http_server).
-behaviour(gen_server).
-compile(export_all).
-compile(nowarn_export_all).
-export([start_link/0]).
-export([get_received_data/0]).
-export([stop/1]).
-export([code_change/3, handle_call/3, handle_cast/2, handle_info/2, init/1, terminate/2]).
-define(HTTP_PORT, 9999).
-define(HTTPS_PORT, 8888).
-record(state, {}).
%%--------------------------------------------------------------------
%% APIs
%%--------------------------------------------------------------------
start() ->
{ok, _} = application:ensure_all_started(cowboy),
cowboy_router:compile([
{'_', [
{"/", ?MODULE, self()}
]}
]).
start_link() ->
gen_server:start_link(?MODULE, [], []).
start_http() ->
{ok, _Pid1} = cowboy:start_clear(http, [{port, 8080}], #{
env => #{dispatch => start()}
}),
io:format("Start http server on 8080 successfully!~n").
init([]) ->
EtsOptions = [named_table, public, set, {write_concurrency, true},
{read_concurrency, true}],
emqx_web_hook_http_test = ets:new(emqx_web_hook_http_test, EtsOptions),
ok = start_http(?HTTP_PORT),
ok = start_https(?HTTPS_PORT),
{ok, #state{}}.
start_https() ->
Path = emqx_ct_helpers:deps_path(emqx_web_hook, "test/emqx_web_hook_SUITE_data/"),
SslOpts = [{keyfile, Path ++ "/server-key.pem"},
{cacertfile, Path ++ "/ca.pem"},
{certfile, Path ++ "/server-cert.pem"}],
handle_call(_Request, _From, State) ->
{reply, ignored, State}.
{ok, _Pid2} = cowboy:start_tls(https, [{port, 8081}] ++ SslOpts,
#{env => #{dispatch => start()}}),
io:format(standard_error, "Start https server on 8081 successfully!~n", []).
handle_cast(_Msg, State) ->
{noreply, State}.
stop_http() ->
ok = cowboy:stop_listener(http),
io:format("Stopped http server on 8080").
handle_info(_Info, State) ->
{noreply, State}.
stop_https() ->
ok = cowboy:stop_listener(https),
io:format("Stopped https server on 8081").
terminate(_Reason, _State) ->
stop_http(),
stop_https().
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
get_received_data() ->
ets:tab2list(emqx_web_hook_http_test).
stop(Pid) ->
ok = gen_server:stop(Pid).
%%--------------------------------------------------------------------
%% Callbacks
%%--------------------------------------------------------------------
init(Req, ReceiverPid) ->
Req1 = handle_request(Req, ReceiverPid),
{ok, Req1, ReceiverPid}.
start_http(Port) ->
{ok, _Pid1} = cowboy:start_clear(http, [{port, Port}], #{
env => #{dispatch => compile_router()}
}),
io:format(standard_error, "[TEST LOG] Start http server on 9999 successfully!~n", []).
%% @private
handle_request(Req, ReceiverPid) ->
start_https(Port) ->
Path = emqx_ct_helpers:deps_path(emqx_web_hook, "test/emqx_web_hook_SUITE_data/"),
SslOpts = [{keyfile, Path ++ "/server-key.pem"},
{cacertfile, Path ++ "/ca.pem"},
{certfile, Path ++ "/server-cert.pem"}],
{ok, _Pid2} = cowboy:start_tls(https, [{port, Port}] ++ SslOpts,
#{env => #{dispatch => compile_router()}}),
io:format(standard_error, "[TEST LOG] Start https server on 8888 successfully!~n", []).
stop_http() ->
ok = cowboy:stop_listener(http),
io:format("[TEST LOG] Stopped http server on 9999").
stop_https() ->
ok = cowboy:stop_listener(https),
io:format("[TEST LOG] Stopped https server on 8888").
compile_router() ->
{ok, _} = application:ensure_all_started(cowboy),
cowboy_router:compile([
{'_', [{"/", ?MODULE, #{}}]}
]).
init(Req, State) ->
Method = cowboy_req:method(Req),
Headers = cowboy_req:headers(Req),
Params =
case Method of
<<"GET">> -> cowboy_req:parse_qs(Req);
<<"POST">> ->
{ok, PostVals, _Req2} = cowboy_req:read_urlencoded_body(Req),
PostVals
end,
io:format("Request Data:~p~nHeaders :~p~n", [Params, Headers]),
erlang:send(ReceiverPid, {Params, Headers}),
reply(Req, ok).
[Params] = case Method of
<<"GET">> -> cowboy_req:parse_qs(Req);
<<"POST">> ->
{ok, PostVals, _} = cowboy_req:read_urlencoded_body(Req),
PostVals
end,
ets:insert(emqx_web_hook_http_test, {Params, Headers}),
{ok, reply(Req, ok), State}.
%% @private
reply(Req, ok) ->
cowboy_req:reply(200, #{<<"content-type">> => <<"text/plain">>}, <<"hello">>, Req);
cowboy_req:reply(200, #{<<"content-type">> => <<"text/plain">>}, <<"ok">>, Req);
reply(Req, error) ->
cowboy_req:reply(404, #{<<"content-type">> => <<"text/plain">>}, <<"deny">>, Req).
cowboy_req:reply(404, #{<<"content-type">> => <<"text/plain">>}, <<"deny">>, Req).

2
deploy/charts/emqx/values.yaml
View File

@ -40,7 +40,7 @@ initContainers: {}
# image: alpine
# command: ["sh", "-c", "for i in $(seq 1 300); do nc -zvw1 mysql 3306 && exit 0 || sleep 3; done; exit 1"]
## EMQX configuration item, see the documentation (https://github.com/emqx/emqx-docker#emq-x-configuration)
## EMQX configuration item, see the documentation (https://hub.docker.com/r/emqx/emqx)
emqxConfig:
EMQX_CLUSTER__K8S__APISERVER: \"https://kubernetes.default.svc:443\"
## The address type is used to extract host from k8s service.

38
deploy/docker/docker-entrypoint.sh
View File

@ -62,32 +62,36 @@ if [[ -z "$EMQX_NODE__MAX_ETS_TABLES" ]]; then
export EMQX_NODE__MAX_ETS_TABLES=2097152
fi
if [[ -z "$EMQX__LOG_CONSOLE" ]]; then
export EMQX__LOG_CONSOLE='console'
fi
if [[ -z "$EMQX_LISTENER__TCP__EXTERNAL__ACCEPTORS" ]]; then
export EMQX_LISTENER__TCP__EXTERNAL__ACCEPTORS=64
fi
if [[ -z "$EMQX_LISTENER__TCP__EXTERNAL__MAX_CLIENTS" ]]; then
export EMQX_LISTENER__TCP__EXTERNAL__MAX_CLIENTS=1000000
if [[ -z "$EMQX_LISTENER__TCP__EXTERNAL__MAX_CONNECTIONS" ]]; then
export EMQX_LISTENER__TCP__EXTERNAL__MAX_CONNECTIONS=1024000
fi
if [[ -z "$EMQX_LISTENER__SSL__EXTERNAL__ACCEPTORS" ]]; then
export EMQX_LISTENER__SSL__EXTERNAL__ACCEPTORS=32
fi
if [[ -z "$EMQX_LISTENER__SSL__EXTERNAL__MAX_CLIENTS" ]]; then
export EMQX_LISTENER__SSL__EXTERNAL__MAX_CLIENTS=500000
if [[ -z "$EMQX_LISTENER__SSL__EXTERNAL__MAX_CONNECTIONS" ]]; then
export EMQX_LISTENER__SSL__EXTERNAL__MAX_CONNECTIONS=102400
fi
if [[ -z "$EMQX_LISTENER__WS__EXTERNAL__ACCEPTORS" ]]; then
export EMQX_LISTENER__WS__EXTERNAL__ACCEPTORS=16
fi
if [[ -z "$EMQX_LISTENER__WS__EXTERNAL__MAX_CLIENTS" ]]; then
export EMQX_LISTENER__WS__EXTERNAL__MAX_CLIENTS=250000
if [[ -z "$EMQX_LISTENER__WS__EXTERNAL__MAX_CONNECTIONS" ]]; then
export EMQX_LISTENER__WS__EXTERNAL__MAX_CONNECTIONS=102400
fi
if [[ -z "$EMQX_LISTENER__WSS__EXTERNAL__ACCEPTORS" ]]; then
export EMQX_LISTENER__WSS__EXTERNAL__ACCEPTORS=16
fi
if [[ -z "$EMQX_LISTENER__WSS__EXTERNAL__MAX_CONNECTIONS" ]]; then
export EMQX_LISTENER__WSS__EXTERNAL__MAX_CONNECTIONS=102400
fi
# Fix issue #42 - export env EMQX_DASHBOARD__DEFAULT_USER__PASSWORD to configure
@ -128,9 +132,9 @@ try_fill_config() {
if grep -qE "^[#[:space:]]*$escaped_key\s*=" "$file"; then
echo_value "$key" "$value"
if [[ -z "$value" ]]; then
sed -i -r "s/^[#[:space:]]*($escaped_key)\s*=\s*(.*)/# \1 = \2/" "$file"
echo "$(sed -r "s/^[#[:space:]]*($escaped_key)\s*=\s*(.*)/# \1 = \2/" "$file")" > "$file"
else
sed -i -r "s/^[#[:space:]]*($escaped_key)\s*=\s*(.*)/\1 = $escaped_value/" "$file"
echo "$(sed -r "s/^[#[:space:]]*($escaped_key)\s*=\s*(.*)/\1 = $escaped_value/" "$file")" > "$file"
fi
# Check if config has a numbering system, but no existing configuration line in file
elif echo "$key" | grep -qE '\.\d+|\d+\.'; then
@ -139,7 +143,7 @@ try_fill_config() {
template="$(echo "$escaped_key" | sed -r -e 's/\\\.[0-9]+/\\.[0-9]+/g' -e 's/[0-9]+\\\./[0-9]+\\./g')"
if grep -qE "^[#[:space:]]*$template\s*=" "$file"; then
echo_value "$key" "$value"
sed -i '$a'\\ "$file"
echo "$(sed '$a'\\ "$file")" > "$file"
echo "$key = $value" >> "$file"
fi
fi
@ -171,13 +175,13 @@ fill_tuples() {
local elements=${*:2}
for var in $elements; do
if grep -qE "\{\s*$var\s*,\s*(true|false)\s*\}\s*\." "$file"; then
sed -i -r "s/\{\s*($var)\s*,\s*(true|false)\s*\}\s*\./{\1, true}./1" "$file"
echo "$(sed -r "s/\{\s*($var)\s*,\s*(true|false)\s*\}\s*\./{\1, true}./1" "$file")" > "$file"
elif grep -q "$var\s*\." "$file"; then
# backward compatible.
sed -i -r "s/($var)\s*\./{\1, true}./1" "$file"
echo "$(sed -r "s/($var)\s*\./{\1, true}./1" "$file")" > "$file"
else
sed -i '$a'\\ "$file"
echo "{$var, true}." >>"$file"
echo "$(sed '$a'\\ "$file")" > "$file"
echo "{$var, true}." >> "$file"
fi
done
}

4
deploy/docker/start.sh
View File

@ -1,6 +1,8 @@
#!/bin/sh
set -e -u
EMQX_WAIT_TIME=${EMQX_WAIT_TIME:-5}
emqx_exit(){
# At least erlang.log.1 exists
if [ -f /opt/emqx/log/erlang.log.1 ]; then
@ -53,7 +55,7 @@ while [ $IDLE_TIME -lt 5 ]; do
else
echo "['$(date -u +"%Y-%m-%dT%H:%M:%SZ")']:emqx not running, waiting for recovery in $((25-IDLE_TIME*5)) seconds"
fi
sleep 5
sleep $EMQX_WAIT_TIME
done
# If running to here (the result 5 times not is running, thus in 25s emqx is not running), exit docker image

2
rebar.config
View File

@ -17,7 +17,7 @@
{extra_src_dirs, [{"etc", [{recursive,true}]}]}.
{xref_checks,[undefined_function_calls,undefined_functions,locals_not_used,
deprecated_function_calls,warnings_as_errors, deprecated_functions]}.
deprecated_function_calls,warnings_as_errors,deprecated_functions]}.
{dialyzer, [
{warnings, [unmatched_returns, error_handling, race_conditions]},

10
rebar.config.erl
View File

@ -34,19 +34,19 @@ test_deps() ->
].
profiles() ->
[ {'emqx', [ {erl_opts, [no_debug_info, {parse_transform, mod_vsn}]}
[ {'emqx', [ {erl_opts, [no_debug_info, warnings_as_errors, {parse_transform, mod_vsn}]}
, {relx, relx('emqx')}
]}
, {'emqx-pkg', [ {erl_opts, [no_debug_info, {parse_transform, mod_vsn}]}
, {'emqx-pkg', [ {erl_opts, [no_debug_info, warnings_as_errors, {parse_transform, mod_vsn}]}
, {relx, relx('emqx-pkg')}
]}
, {'emqx-edge', [ {erl_opts, [no_debug_info, {parse_transform, mod_vsn}]}
, {'emqx-edge', [ {erl_opts, [no_debug_info, warnings_as_errors, {parse_transform, mod_vsn}]}
, {relx, relx('emqx-edge')}
]}
, {'emqx-edge-pkg', [ {erl_opts, [no_debug_info, {parse_transform, mod_vsn}]}
, {'emqx-edge-pkg', [ {erl_opts, [no_debug_info, warnings_as_errors, {parse_transform, mod_vsn}]}
, {relx, relx('emqx-edge-pkg')}
]}
, {check, [ {erl_opts, [debug_info, {parse_transform, mod_vsn}]}
, {check, [ {erl_opts, [debug_info, warnings_as_errors, {parse_transform, mod_vsn}]}
]}
, {test, [ {deps, test_deps()}
, {erl_opts, [debug_info, {parse_transform, mod_vsn}] ++ erl_opts_i()}

1
src/emqx_alarm.erl
View File

@ -83,7 +83,6 @@
timer = undefined :: undefined | reference()
}).
-type state() :: #state{}.
-type action() :: log | publish | event.
-define(ACTIVATED_ALARM, emqx_activated_alarm).