diff --git a/apps/emqx_gateway/etc/emqx_gateway.conf.example b/apps/emqx_gateway/etc/emqx_gateway.conf.example index 03a9a7868..49977c473 100644 --- a/apps/emqx_gateway/etc/emqx_gateway.conf.example +++ b/apps/emqx_gateway/etc/emqx_gateway.conf.example @@ -70,9 +70,9 @@ gateway.stomp { ## SSL options ## See ${example_common_ssl_options} for more information ssl.versions = ["tlsv1.3", "tlsv1.2", "tlsv1.1", "tlsv1"] - ssl.keyfile = "{{ platform_etc_dir }}/certs/key.pem" - ssl.certfile = "{{ platform_etc_dir }}/certs/cert.pem" ssl.cacertfile = "{{ platform_etc_dir }}/certs/cacert.pem" + ssl.certfile = "{{ platform_etc_dir }}/certs/cert.pem" + ssl.keyfile = "{{ platform_etc_dir }}/certs/key.pem" #ssl.verify = verify_none #ssl.fail_if_no_peer_cert = false #ssl.server_name_indication = disable diff --git a/apps/emqx_gateway/include/emqx_gateway_http.hrl b/apps/emqx_gateway/include/emqx_gateway_http.hrl new file mode 100644 index 000000000..d48c89859 --- /dev/null +++ b/apps/emqx_gateway/include/emqx_gateway_http.hrl @@ -0,0 +1,26 @@ +%%-------------------------------------------------------------------- +%% Copyright (c) 2021 EMQ Technologies Co., Ltd. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%%-------------------------------------------------------------------- + +-define(BAD_REQUEST, 'BAD_REQUEST'). +-define(NOT_FOUND, 'NOT_FOUND'). +-define(INTERNAL_ERROR, 'INTERNAL_SERVER_ERROR'). + +-define(STANDARD_RESP(R), + R#{ 400 => emqx_dashboard_swagger:error_codes( + [?BAD_REQUEST], <<"Bad request">>) + , 404 => emqx_dashboard_swagger:error_codes( + [?NOT_FOUND], <<"Not Found">>) + }). diff --git a/apps/emqx_gateway/src/coap/emqx_coap_api.erl b/apps/emqx_gateway/src/coap/emqx_coap_api.erl index aa448a1a2..0efb8043e 100644 --- a/apps/emqx_gateway/src/coap/emqx_coap_api.erl +++ b/apps/emqx_gateway/src/coap/emqx_coap_api.erl @@ -25,7 +25,7 @@ -export([request/2]). --define(PREFIX, "/gateway/coap/:clientid"). +-define(PREFIX, "/gateway/coap/clients/:clientid"). -define(DEF_WAIT_TIME, 10). -import(emqx_mgmt_util, [ schema/1 diff --git a/apps/emqx_gateway/src/emqx_gateway_api.erl b/apps/emqx_gateway/src/emqx_gateway_api.erl index 925b7cc0e..596d42c9a 100644 --- a/apps/emqx_gateway/src/emqx_gateway_api.erl +++ b/apps/emqx_gateway/src/emqx_gateway_api.erl @@ -16,22 +16,30 @@ %% -module(emqx_gateway_api). +-include("emqx_gateway_http.hrl"). +-include_lib("typerefl/include/types.hrl"). +-include_lib("hocon/include/hoconsc.hrl"). -include_lib("emqx/include/emqx_placeholder.hrl"). -include_lib("emqx/include/emqx_authentication.hrl"). -behaviour(minirest_api). +-import(hoconsc, [mk/2, ref/1, ref/2]). + -import(emqx_gateway_http, [ return_http_error/2 , with_gateway/2 - , schema_bad_request/0 - , schema_not_found/0 - , schema_internal_error/0 - , schema_no_content/0 ]). -%% minirest behaviour callbacks --export([api_spec/0]). +%% minirest/dashbaord_swagger behaviour callbacks +-export([ api_spec/0 + , paths/0 + , schema/1 + ]). + +-export([ roots/0 + , fields/1 + ]). %% http handlers -export([ gateway/2 @@ -44,12 +52,12 @@ %%-------------------------------------------------------------------- api_spec() -> - {metadata(apis()), []}. + emqx_dashboard_swagger:spec(?MODULE, #{check_schema => true}). -apis() -> - [ {"/gateway", gateway} - , {"/gateway/:name", gateway_insta} - , {"/gateway/:name/stats", gateway_insta_stats} +paths() -> + [ "/gateway" + , "/gateway/:name" + , "/gateway/:name/stats" ]. %%-------------------------------------------------------------------- @@ -137,270 +145,209 @@ gateway_insta_stats(get, _Req) -> %% Swagger defines %%-------------------------------------------------------------------- -metadata(APIs) -> - metadata(APIs, []). -metadata([], APIAcc) -> - lists:reverse(APIAcc); -metadata([{Path, Fun}|More], APIAcc) -> - Methods = [get, post, put, delete, patch], - Mds = lists:foldl(fun(M, Acc) -> - try - Acc#{M => swagger(Path, M)} - catch - error : function_clause -> - Acc - end - end, #{}, Methods), - metadata(More, [{Path, Mds, Fun} | APIAcc]). - -swagger("/gateway", get) -> - #{ description => <<"Get gateway list">> - , parameters => params_gateway_status_in_qs() - , responses => - #{ <<"200">> => schema_gateway_overview_list() } +schema("/gateway") -> + #{ 'operationId' => gateway, + get => + #{ description => <<"Get gateway list">> + , parameters => params_gateway_status_in_qs() + , responses => + ?STANDARD_RESP(#{200 => ref(gateway_overview)}) + }, + post => + #{ description => <<"Load a gateway">> + , 'requestBody' => schema_gateways_conf() + , responses => + ?STANDARD_RESP(#{201 => schema_gateways_conf()}) + } }; -swagger("/gateway", post) -> - #{ description => <<"Load a gateway">> - , requestBody => schema_gateway_conf() - , responses => - #{ <<"400">> => schema_bad_request() - , <<"404">> => schema_not_found() - , <<"500">> => schema_internal_error() - , <<"204">> => schema_no_content() - } +schema("/gateway/:name") -> + #{ 'operationId' => gateway_insta, + get => + #{ description => <<"Get the gateway configurations">> + , parameters => params_gateway_name_in_path() + , responses => + ?STANDARD_RESP(#{200 => schema_gateways_conf()}) + }, + delete => + #{ description => <<"Delete/Unload the gateway">> + , parameters => params_gateway_name_in_path() + , responses => + ?STANDARD_RESP(#{204 => <<"Deleted">>}) + }, + put => + #{ description => <<"Update the gateway configurations/status">> + , parameters => params_gateway_name_in_path() + , 'requestBody' => schema_gateways_conf() + , responses => + ?STANDARD_RESP(#{200 => schema_gateways_conf()}) + } }; -swagger("/gateway/:name", get) -> - #{ description => <<"Get the gateway configurations">> - , parameters => params_gateway_name_in_path() - , responses => - #{ <<"400">> => schema_bad_request() - , <<"404">> => schema_not_found() - , <<"500">> => schema_internal_error() - , <<"200">> => schema_gateway_conf() - } - }; -swagger("/gateway/:name", delete) -> - #{ description => <<"Delete/Unload the gateway">> - , parameters => params_gateway_name_in_path() - , responses => - #{ <<"400">> => schema_bad_request() - , <<"404">> => schema_not_found() - , <<"500">> => schema_internal_error() - , <<"204">> => schema_no_content() - } - }; -swagger("/gateway/:name", put) -> - #{ description => <<"Update the gateway configurations/status">> - , parameters => params_gateway_name_in_path() - , requestBody => schema_gateway_conf() - , responses => - #{ <<"400">> => schema_bad_request() - , <<"404">> => schema_not_found() - , <<"500">> => schema_internal_error() - , <<"200">> => schema_no_content() - } - }; -swagger("/gateway/:name/stats", get) -> - #{ description => <<"Get gateway Statistic">> - , parameters => params_gateway_name_in_path() - , responses => - #{ <<"400">> => schema_bad_request() - , <<"404">> => schema_not_found() - , <<"500">> => schema_internal_error() - , <<"200">> => schema_gateway_stats() - } +schema("/gateway/:name/stats") -> + #{ 'operationId' => gateway_insta_stats, + get => + #{ description => <<"Get gateway Statistic">> + , parameters => params_gateway_name_in_path() + , responses => + ?STANDARD_RESP( + #{200 => emqx_dashboard_swagger:schema_with_examples( + ref(gateway_stats), + examples_gateway_stats()) + }) + } }. %%-------------------------------------------------------------------- %% params defines params_gateway_name_in_path() -> - [#{ name => name - , in => path - , schema => #{type => string} - , required => true - }]. + [{name, + mk(binary(), + #{ in => path + , desc => <<"Gateway Name">> + })} + ]. params_gateway_status_in_qs() -> - [#{ name => status - , in => query - , schema => #{type => string} - , required => false - }]. + [{status, + mk(binary(), + #{ in => query + , nullable => true + , desc => <<"Gateway Status">> + })} + ]. %%-------------------------------------------------------------------- %% schemas -schema_gateway_overview_list() -> - emqx_mgmt_util:array_schema( - #{ type => object - , properties => properties_gateway_overview() - }, - <<"Gateway list">> +roots() -> + [ gateway_overview + , gateway_stats + ]. + +fields(gateway_overview) -> + [ {name, + mk(string(), + #{ desc => <<"Gateway Name">>})} + , {status, + mk(hoconsc:enum([running, stopped, unloaded]), + #{ desc => <<"The Gateway status">>})} + , {created_at, + mk(string(), + #{desc => <<"The Gateway created datetime">>})} + , {started_at, + mk(string(), + #{ nullable => true + , desc => <<"The Gateway started datetime">>})} + , {stopped_at, + mk(string(), + #{ nullable => true + , desc => <<"The Gateway stopped datetime">>})} + , {max_connections, + mk(integer(), + #{ desc => <<"The Gateway allowed maximum connections/clients">>})} + , {current_connections, + mk(integer(), + #{ desc => <<"The Gateway current connected connections/clients">> + })} + , {listeners, + mk(hoconsc:array(ref(gateway_listener_overview)), + #{ nullable => {true, recursively} + , desc => <<"The Gateway listeners overview">>})} + ]; +fields(gateway_listener_overview) -> + [ {id, + mk(string(), + #{ desc => <<"Listener ID">>})} + , {running, + mk(boolean(), + #{ desc => <<"Listener Running status">>})} + , {type, + mk(hoconsc:enum([tcp, ssl, udp, dtls]), + #{ desc => <<"Listener Type">>})} + ]; + +fields(Gw) when Gw == stomp; Gw == mqttsn; + Gw == coap; Gw == lwm2m; + Gw == exproto -> + [{name, + mk(string(), #{ desc => <<"Gateway Name">>})} + ] ++ convert_listener_struct(emqx_gateway_schema:fields(Gw)); +fields(Listener) when Listener == tcp_listener; + Listener == ssl_listener; + Listener == udp_listener; + Listener == dtls_listener -> + [ {id, + mk(string(), + #{ nullable => true + , desc => <<"Listener ID">>})} + , {type, + mk(hoconsc:union([tcp, ssl, udp, dtls]), + #{ desc => <<"Listener type">>})} + , {name, + mk(string(), + #{ desc => <<"Listener Name">>})} + , {running, + mk(boolean(), + #{ nullable => true + , desc => <<"Listener running status">>})} + ] ++ emqx_gateway_schema:fields(Listener); + +fields(gateway_stats) -> + [{key, mk(string(), #{})}]. + +schema_gateways_conf() -> + %% XXX: We need convert the emqx_gateway_schema's listener map + %% structure to array + emqx_dashboard_swagger:schema_with_examples( + hoconsc:union([ref(?MODULE, stomp), ref(?MODULE, mqttsn), + ref(?MODULE, coap), ref(?MODULE, lwm2m), + ref(?MODULE, exproto)]), + examples_gateway_confs() ). -%% XXX: This is whole confs for all type gateways. It is used to fill the -%% default configurations and generate the swagger-schema -%% -%% NOTE: It is a temporary measure to generate swagger-schema --define(COAP_GATEWAY_CONFS, -#{?EMQX_AUTHENTICATION_CONFIG_ROOT_NAME_BINARY => - #{<<"mechanism">> => <<"password-based">>, - <<"name">> => <<"authenticator1">>, - <<"server_type">> => <<"built-in-database">>, - <<"user_id_type">> => <<"clientid">>}, - <<"name">> => <<"coap">>, - <<"enable">> => true, - <<"enable_stats">> => true,<<"heartbeat">> => <<"30s">>, - <<"idle_timeout">> => <<"30s">>, - <<"listeners">> => [ - #{<<"id">> => <<"coap:udp:default">>, - <<"type">> => <<"udp">>, - <<"running">> => true, - <<"acceptors">> => 8,<<"bind">> => 5683, - <<"max_conn_rate">> => 1000, - <<"max_connections">> => 10240}], - <<"mountpoint">> => <<>>,<<"notify_type">> => <<"qos">>, - <<"publish_qos">> => <<"qos1">>, - <<"subscribe_qos">> => <<"qos0">>} -). +convert_listener_struct(Schema) -> + {value, {listeners, + #{type := Type}}, Schema1} = lists:keytake(listeners, 1, Schema), + ListenerSchema = hoconsc:mk(listeners_schema(Type), + #{ nullable => {true, recursively} + , desc => <<"The gateway listeners">> + }), + lists:keystore(listeners, 1, Schema1, {listeners, ListenerSchema}). --define(EXPROTO_GATEWAY_CONFS, -#{<<"enable">> => true, - <<"name">> => <<"exproto">>, - <<"enable_stats">> => true, - <<"handler">> => - #{<<"address">> => <<"http://127.0.0.1:9001">>}, - <<"idle_timeout">> => <<"30s">>, - <<"listeners">> => [ - #{<<"id">> => <<"exproto:tcp:default">>, - <<"type">> => <<"tcp">>, - <<"running">> => true, - <<"acceptors">> => 8,<<"bind">> => 7993, - <<"max_conn_rate">> => 1000, - <<"max_connections">> => 10240}], - <<"mountpoint">> => <<>>, - <<"server">> => #{<<"bind">> => 9100}} -). - --define(LWM2M_GATEWAY_CONFS, -#{<<"auto_observe">> => false, - <<"name">> => <<"lwm2m">>, - <<"enable">> => true, - <<"enable_stats">> => true, - <<"idle_timeout">> => <<"30s">>, - <<"lifetime_max">> => <<"86400s">>, - <<"lifetime_min">> => <<"1s">>, - <<"listeners">> => [ - #{<<"id">> => <<"lwm2m:udp:default">>, - <<"type">> => <<"udp">>, - <<"running">> => true, - <<"bind">> => 5783}], - <<"mountpoint">> => <<"lwm2m/", ?PH_S_ENDPOINT_NAME, "/">>, - <<"qmode_time_windonw">> => 22, - <<"translators">> => - #{<<"command">> => <<"dn/#">>,<<"notify">> => <<"up/notify">>, - <<"register">> => <<"up/resp">>, - <<"response">> => <<"up/resp">>, - <<"update">> => <<"up/resp">>}, - <<"update_msg_publish_condition">> => - <<"contains_object_list">>, - <<"xml_dir">> => <<"etc/lwm2m_xml">>} -). - --define(MQTTSN_GATEWAY_CONFS, -#{<<"broadcast">> => true, - <<"clientinfo_override">> => - #{<<"password">> => <<"abc">>, - <<"username">> => <<"mqtt_sn_user">>}, - <<"enable">> => true, - <<"name">> => <<"mqtt-sn">>, - <<"enable_qos3">> => true,<<"enable_stats">> => true, - <<"gateway_id">> => 1,<<"idle_timeout">> => <<"30s">>, - <<"listeners">> => [ - #{<<"id">> => <<"mqttsn:udp:default">>, - <<"type">> => <<"udp">>, - <<"running">> => true, - <<"bind">> => 1884,<<"max_conn_rate">> => 1000, - <<"max_connections">> => 10240000}], - <<"mountpoint">> => <<>>, - <<"predefined">> => - [#{<<"id">> => 1, - <<"topic">> => <<"/predefined/topic/name/hello">>}, - #{<<"id">> => 2, - <<"topic">> => <<"/predefined/topic/name/nice">>}]} -). - --define(STOMP_GATEWAY_CONFS, -#{?EMQX_AUTHENTICATION_CONFIG_ROOT_NAME_BINARY => - #{<<"mechanism">> => <<"password-based">>, - <<"name">> => <<"authenticator1">>, - <<"server_type">> => <<"built-in-database">>, - <<"user_id_type">> => <<"clientid">>}, - <<"clientinfo_override">> => - #{<<"password">> => <<"${Packet.headers.passcode}">>, - <<"username">> => <<"${Packet.headers.login}">>}, - <<"enable">> => true, - <<"name">> => <<"stomp">>, - <<"enable_stats">> => true, - <<"frame">> => - #{<<"max_body_length">> => 8192,<<"max_headers">> => 10, - <<"max_headers_length">> => 1024}, - <<"idle_timeout">> => <<"30s">>, - <<"listeners">> => [ - #{<<"id">> => <<"stomp:tcp:default">>, - <<"type">> => <<"tcp">>, - <<"running">> => true, - <<"acceptors">> => 16,<<"active_n">> => 100, - <<"bind">> => 61613,<<"max_conn_rate">> => 1000, - <<"max_connections">> => 1024000}], - <<"mountpoint">> => <<>>} -). - -%% --- END - -schema_gateway_conf() -> - emqx_mgmt_util:schema( - #{oneOf => - [ emqx_mgmt_api_configs:gen_schema(?STOMP_GATEWAY_CONFS) - , emqx_mgmt_api_configs:gen_schema(?MQTTSN_GATEWAY_CONFS) - , emqx_mgmt_api_configs:gen_schema(?COAP_GATEWAY_CONFS) - , emqx_mgmt_api_configs:gen_schema(?LWM2M_GATEWAY_CONFS) - , emqx_mgmt_api_configs:gen_schema(?EXPROTO_GATEWAY_CONFS) - ]}). - -schema_gateway_stats() -> - emqx_mgmt_util:schema( - #{ type => object - , properties => - #{ a_key => #{type => string} - }}). +listeners_schema(?R_REF(_Mod, tcp_listeners)) -> + hoconsc:array(hoconsc:union([ref(tcp_listener), ref(ssl_listener)])); +listeners_schema(?R_REF(_Mod, udp_listeners)) -> + hoconsc:array(hoconsc:union([ref(udp_listener), ref(dtls_listener)])); +listeners_schema(?R_REF(_Mod, udp_tcp_listeners)) -> + hoconsc:array(hoconsc:union([ref(tcp_listener), ref(ssl_listener), + ref(udp_listener), ref(dtls_listener)])). %%-------------------------------------------------------------------- -%% properties +%% examples -properties_gateway_overview() -> - ListenerProps = - [ {id, string, - <<"Listener ID">>} - , {running, boolean, - <<"Listener Running status">>} - , {type, string, - <<"Listener Type">>, [<<"tcp">>, <<"ssl">>, <<"udp">>, <<"dtls">>]} - ], - emqx_mgmt_util:properties( - [ {name, string, - <<"Gateway Name">>} - , {status, string, - <<"Gateway Status">>, - [<<"running">>, <<"stopped">>, <<"unloaded">>]} - , {created_at, string, - <<>>} - , {started_at, string, - <<>>} - , {stopped_at, string, - <<>>} - , {max_connections, integer, <<>>} - , {current_connections, integer, <<>>} - , {listeners, {array, object}, ListenerProps} - ]). +examples_gateway_confs() -> + #{ stomp_gateway => + #{ summary => <<"A simple STOMP gateway configs">> + , value => + #{ enable => true + , enable_stats => true + , idle_timeout => <<"30s">> + , mountpoint => <<"stomp/">> + , frame => + #{ max_header => 10 + , make_header_length => 1024 + , max_body_length => 65535 + } + } + } + , mqttsn_gateway => + #{ summary => <<"A simple MQTT-SN gateway configs">> + , value => + #{ enable => true + , enable_stats => true + } + } + }. + +examples_gateway_stats() -> + #{}. diff --git a/apps/emqx_gateway/src/emqx_gateway_api_authn.erl b/apps/emqx_gateway/src/emqx_gateway_api_authn.erl index 105a96989..d9bdf4d7c 100644 --- a/apps/emqx_gateway/src/emqx_gateway_api_authn.erl +++ b/apps/emqx_gateway/src/emqx_gateway_api_authn.erl @@ -13,17 +13,14 @@ %% See the License for the specific language governing permissions and %% limitations under the License. %%-------------------------------------------------------------------- -%% + -module(emqx_gateway_api_authn). -behaviour(minirest_api). +-include("emqx_gateway_http.hrl"). -include_lib("typerefl/include/types.hrl"). --define(BAD_REQUEST, 'BAD_REQUEST'). --define(NOT_FOUND, 'NOT_FOUND'). --define(INTERNAL_ERROR, 'INTERNAL_SERVER_ERROR'). - -import(hoconsc, [mk/2, ref/2]). -import(emqx_dashboard_swagger, [error_codes/2]). @@ -82,17 +79,15 @@ authn(get, #{bindings := #{name := Name0}}) -> authn(put, #{bindings := #{name := Name0}, body := Body}) -> with_gateway(Name0, fun(GwName, _) -> - %% TODO: return the authn instances? - ok = emqx_gateway_http:update_authn(GwName, Body), - {204} + {ok, Authn} = emqx_gateway_http:update_authn(GwName, Body), + {200, Authn} end); authn(post, #{bindings := #{name := Name0}, body := Body}) -> with_gateway(Name0, fun(GwName, _) -> - %% TODO: return the authn instances? - ok = emqx_gateway_http:add_authn(GwName, Body), - {204} + {ok, Authn} = emqx_gateway_http:add_authn(GwName, Body), + {201, Authn} end); authn(delete, #{bindings := #{name := Name0}}) -> @@ -164,48 +159,30 @@ schema("/gateway/:name/authentication") -> #{ description => <<"Get the gateway authentication">> , parameters => params_gateway_name_in_path() , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 200 => schema_authn() - , 204 => <<"Authentication does not initiated">> - } + ?STANDARD_RESP( + #{ 200 => schema_authn() + , 204 => <<"Authentication does not initiated">> + }) }, put => #{ description => <<"Update authentication for the gateway">> , parameters => params_gateway_name_in_path() , 'requestBody' => schema_authn() , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 204 => <<"Updated">> %% XXX: ??? return the updated object - } + ?STANDARD_RESP(#{200 => schema_authn()}) }, post => #{ description => <<"Add authentication for the gateway">> , parameters => params_gateway_name_in_path() , 'requestBody' => schema_authn() , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 204 => <<"Added">> - } + ?STANDARD_RESP(#{201 => schema_authn()}) }, delete => #{ description => <<"Remove the gateway authentication">> , parameters => params_gateway_name_in_path() , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 204 => <<"Deleted">> - } + ?STANDARD_RESP(#{204 => <<"Deleted">>}) } }; schema("/gateway/:name/authentication/users") -> @@ -215,14 +192,11 @@ schema("/gateway/:name/authentication/users") -> , parameters => params_gateway_name_in_path() ++ params_paging_in_qs() , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 200 => emqx_dashboard_swagger:schema_with_example( - ref(emqx_authn_api, response_user), - emqx_authn_api:response_user_examples()) - } + ?STANDARD_RESP( + #{ 200 => emqx_dashboard_swagger:schema_with_example( + ref(emqx_authn_api, response_user), + emqx_authn_api:response_user_examples()) + }) }, post => #{ description => <<"Add user for the authentication">> @@ -231,14 +205,11 @@ schema("/gateway/:name/authentication/users") -> ref(emqx_authn_api, request_user_create), emqx_authn_api:request_user_create_examples()) , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 201 => emqx_dashboard_swagger:schema_with_example( - ref(emqx_authn_api, response_user), - emqx_authn_api:response_user_examples()) - } + ?STANDARD_RESP( + #{ 201 => emqx_dashboard_swagger:schema_with_example( + ref(emqx_authn_api, response_user), + emqx_authn_api:response_user_examples()) + }) } }; schema("/gateway/:name/authentication/users/:uid") -> @@ -249,14 +220,11 @@ schema("/gateway/:name/authentication/users/:uid") -> , parameters => params_gateway_name_in_path() ++ params_userid_in_path() , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 200 => emqx_dashboard_swagger:schema_with_example( - ref(emqx_authn_api, response_user), - emqx_authn_api:response_user_examples()) - } + ?STANDARD_RESP( + #{ 200 => emqx_dashboard_swagger:schema_with_example( + ref(emqx_authn_api, response_user), + emqx_authn_api:response_user_examples()) + }) }, put => #{ description => <<"Update the user info for the gateway " @@ -267,14 +235,11 @@ schema("/gateway/:name/authentication/users/:uid") -> ref(emqx_authn_api, request_user_update), emqx_authn_api:request_user_update_examples()) , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 200 => emqx_dashboard_swagger:schema_with_example( - ref(emqx_authn_api, response_user), - emqx_authn_api:response_user_examples()) - } + ?STANDARD_RESP( + #{ 200 => emqx_dashboard_swagger:schema_with_example( + ref(emqx_authn_api, response_user), + emqx_authn_api:response_user_examples()) + }) }, delete => #{ description => <<"Delete the user for the gateway " @@ -282,12 +247,7 @@ schema("/gateway/:name/authentication/users/:uid") -> , parameters => params_gateway_name_in_path() ++ params_userid_in_path() , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 204 => <<"User Deleted">> - } + ?STANDARD_RESP(#{204 => <<"User Deleted">>}) } }; schema("/gateway/:name/authentication/import_users") -> @@ -300,13 +260,7 @@ schema("/gateway/:name/authentication/import_users") -> emqx_authn_api:request_import_users_examples() ) , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - %% XXX: Put a hint message into 204 return ? - , 204 => <<"Imported">> - } + ?STANDARD_RESP(#{204 => <<"Imported">>}) } }. diff --git a/apps/emqx_gateway/src/emqx_gateway_api_clients.erl b/apps/emqx_gateway/src/emqx_gateway_api_clients.erl index 40e058e32..c7a77eb02 100644 --- a/apps/emqx_gateway/src/emqx_gateway_api_clients.erl +++ b/apps/emqx_gateway/src/emqx_gateway_api_clients.erl @@ -16,12 +16,30 @@ -module(emqx_gateway_api_clients). --behaviour(minirest_api). - +-include("emqx_gateway_http.hrl"). +-include_lib("typerefl/include/types.hrl"). +-include_lib("hocon/include/hoconsc.hrl"). +-include_lib("emqx/include/emqx_placeholder.hrl"). -include_lib("emqx/include/logger.hrl"). -%% minirest behaviour callbacks --export([api_spec/0]). +-behaviour(minirest_api). + +-import(hoconsc, [mk/2, ref/1, ref/2]). + +-import(emqx_gateway_http, + [ return_http_error/2 + , with_gateway/2 + ]). + +%% minirest/dashbaord_swagger behaviour callbacks +-export([ api_spec/0 + , paths/0 + , schema/1 + ]). + +-export([ roots/0 + , fields/1 + ]). %% http handlers -export([ clients/2 @@ -34,27 +52,18 @@ , format_channel_info/1 ]). --import(emqx_gateway_http, - [ return_http_error/2 - , with_gateway/2 - , schema_bad_request/0 - , schema_not_found/0 - , schema_internal_error/0 - , schema_no_content/0 - ]). - %%-------------------------------------------------------------------- %% APIs %%-------------------------------------------------------------------- api_spec() -> - {metadata(apis()), []}. + emqx_dashboard_swagger:spec(?MODULE, #{check_schema => true}). -apis() -> - [ {"/gateway/:name/clients", clients} - , {"/gateway/:name/clients/:clientid", clients_insta} - , {"/gateway/:name/clients/:clientid/subscriptions", subscriptions} - , {"/gateway/:name/clients/:clientid/subscriptions/:topic", subscriptions} +paths() -> + [ "/gateway/:name/clients" + , "/gateway/:name/clients/:clientid" + , "/gateway/:name/clients/:clientid/subscriptions" + , "/gateway/:name/clients/:clientid/subscriptions/:topic" ]. -define(CLIENT_QS_SCHEMA, @@ -88,14 +97,16 @@ clients(get, #{ bindings := #{name := Name0} TabName = emqx_gateway_cm:tabname(info, GwName), case maps:get(<<"node">>, Params, undefined) of undefined -> - Response = emqx_mgmt_api:cluster_query(Params, TabName, - ?CLIENT_QS_SCHEMA, ?query_fun), + Response = emqx_mgmt_api:cluster_query( + Params, TabName, + ?CLIENT_QS_SCHEMA, ?query_fun), emqx_mgmt_util:generate_response(Response); Node1 -> Node = binary_to_atom(Node1, utf8), ParamsWithoutNode = maps:without([<<"node">>], Params), - Response = emqx_mgmt_api:node_query(Node, ParamsWithoutNode, - TabName, ?CLIENT_QS_SCHEMA, ?query_fun), + Response = emqx_mgmt_api:node_query( + Node, ParamsWithoutNode, + TabName, ?CLIENT_QS_SCHEMA, ?query_fun), emqx_mgmt_util:generate_response(Response) end end). @@ -105,8 +116,9 @@ clients_insta(get, #{ bindings := #{name := Name0, }) -> ClientId = emqx_mgmt_util:urldecode(ClientId0), with_gateway(Name0, fun(GwName, _) -> - case emqx_gateway_http:lookup_client(GwName, ClientId, - {?MODULE, format_channel_info}) of + case emqx_gateway_http:lookup_client( + GwName, ClientId, + {?MODULE, format_channel_info}) of [ClientInfo] -> {200, ClientInfo}; [ClientInfo | _More] -> @@ -154,7 +166,8 @@ subscriptions(post, #{ bindings := #{name := Name0, {undefined, _} -> return_http_error(400, "Miss topic property"); {Topic, QoS} -> - case emqx_gateway_http:client_subscribe(GwName, ClientId, Topic, QoS) of + case emqx_gateway_http:client_subscribe( + GwName, ClientId, Topic, QoS) of {error, Reason} -> return_http_error(404, Reason); ok -> @@ -204,8 +217,9 @@ query(Tab, {Qs, []}, Continuation, Limit) -> query(Tab, {Qs, Fuzzy}, Continuation, Limit) -> Ms = qs2ms(Qs), FuzzyFilterFun = fuzzy_filter_fun(Fuzzy), - emqx_mgmt_api:select_table_with_count(Tab, {Ms, FuzzyFilterFun}, Continuation, Limit, - fun format_channel_info/1). + emqx_mgmt_api:select_table_with_count( + Tab, {Ms, FuzzyFilterFun}, Continuation, Limit, + fun format_channel_info/1). qs2ms(Qs) -> {MtchHead, Conds} = qs2ms(Qs, 2, {#{}, []}), @@ -218,8 +232,10 @@ qs2ms([{Key, '=:=', Value} | Rest], N, {MtchHead, Conds}) -> NMtchHead = emqx_mgmt_util:merge_maps(MtchHead, ms(Key, Value)), qs2ms(Rest, N, {NMtchHead, Conds}); qs2ms([Qs | Rest], N, {MtchHead, Conds}) -> - Holder = binary_to_atom(iolist_to_binary(["$", integer_to_list(N)]), utf8), - NMtchHead = emqx_mgmt_util:merge_maps(MtchHead, ms(element(1, Qs), Holder)), + Holder = binary_to_atom( + iolist_to_binary(["$", integer_to_list(N)]), utf8), + NMtchHead = emqx_mgmt_util:merge_maps( + MtchHead, ms(element(1, Qs), Holder)), NConds = put_conds(Qs, Holder, Conds), qs2ms(Rest, N+1, {NMtchHead, NConds}). @@ -271,12 +287,14 @@ escape(B) when is_binary(B) -> run_fuzzy_filter(_, []) -> true; -run_fuzzy_filter(E = {_, #{clientinfo := ClientInfo}, _}, [{Key, _, RE} | Fuzzy]) -> +run_fuzzy_filter(E = {_, #{clientinfo := ClientInfo}, _}, + [{Key, _, RE} | Fuzzy]) -> Val = case maps:get(Key, ClientInfo, "") of undefined -> ""; V -> V end, - re:run(Val, RE, [{capture, none}]) == match andalso run_fuzzy_filter(E, Fuzzy). + re:run(Val, RE, [{capture, none}]) == match + andalso run_fuzzy_filter(E, Fuzzy). %%-------------------------------------------------------------------- %% format funcs @@ -294,15 +312,19 @@ format_channel_info({_, Infos, Stats} = R) -> , {port, {peername, ConnInfo, fun peer_to_port/1}} , {is_bridge, ClientInfo, false} , {connected_at, - {connected_at, ConnInfo, fun emqx_gateway_utils:unix_ts_to_rfc3339/1}} + {connected_at, ConnInfo, + fun emqx_gateway_utils:unix_ts_to_rfc3339/1}} , {disconnected_at, - {disconnected_at, ConnInfo, fun emqx_gateway_utils:unix_ts_to_rfc3339/1}} - , {connected, {conn_state, Infos, fun conn_state_to_connected/1}} + {disconnected_at, ConnInfo, + fun emqx_gateway_utils:unix_ts_to_rfc3339/1}} + , {connected, {conn_state, Infos, + fun conn_state_to_connected/1}} , {keepalive, ClientInfo, 0} , {clean_start, ConnInfo, true} , {expiry_interval, ConnInfo, 0} , {created_at, - {created_at, SessInfo, fun emqx_gateway_utils:unix_ts_to_rfc3339/1}} + {created_at, SessInfo, + fun emqx_gateway_utils:unix_ts_to_rfc3339/1}} , {subscriptions_cnt, Stats, 0} , {subscriptions_max, Stats, infinity} , {inflight_cnt, Stats, 0} @@ -384,275 +406,338 @@ conn_state_to_connected(_) -> false. %% Swagger defines %%-------------------------------------------------------------------- -metadata(APIs) -> - metadata(APIs, []). -metadata([], APIAcc) -> - lists:reverse(APIAcc); -metadata([{Path, Fun} | More], APIAcc) -> - Methods = [get, post, put, delete, patch], - Mds = lists:foldl(fun(M, Acc) -> - try - Acc#{M => swagger(Path, M)} - catch - error : function_clause -> - Acc - end - end, #{}, Methods), - metadata(More, [{Path, Mds, Fun} | APIAcc]). - -swagger("/gateway/:name/clients", get) -> - #{ description => <<"Get the gateway clients">> - , parameters => params_client_query() - , responses => - #{ <<"400">> => schema_bad_request() - , <<"404">> => schema_not_found() - , <<"500">> => schema_internal_error() - , <<"200">> => schema_clients_list() - } +schema("/gateway/:name/clients") -> + #{ 'operationId' => clients + , get => + #{ description => <<"Get the gateway client list">> + , parameters => params_client_query() + , responses => + ?STANDARD_RESP( + #{ 200 => emqx_dashboard_swagger:schema_with_examples( + hoconsc:array(ref(client)), + examples_client_list())}) + } }; -swagger("/gateway/:name/clients/:clientid", get) -> - #{ description => <<"Get the gateway client infomation">> - , parameters => params_client_insta() - , responses => - #{ <<"400">> => schema_bad_request() - , <<"404">> => schema_not_found() - , <<"500">> => schema_internal_error() - , <<"200">> => schema_client() - } +schema("/gateway/:name/clients/:clientid") -> + #{ 'operationId' => clients_insta + , get => + #{ description => <<"Get the gateway client infomation">> + , parameters => params_client_insta() + , responses => + ?STANDARD_RESP( + #{ 200 => emqx_dashboard_swagger:schema_with_examples( + ref(client), + examples_client())}) + } + , delete => + #{ description => <<"Kick out the gateway client">> + , parameters => params_client_insta() + , responses => + ?STANDARD_RESP(#{204 => <<"Kicked">>}) + } }; -swagger("/gateway/:name/clients/:clientid", delete) -> - #{ description => <<"Kick out the gateway client">> - , parameters => params_client_insta() - , responses => - #{ <<"400">> => schema_bad_request() - , <<"404">> => schema_not_found() - , <<"500">> => schema_internal_error() - , <<"204">> => schema_no_content() - } +schema("/gateway/:name/clients/:clientid/subscriptions") -> + #{ 'operationId' => subscriptions + , get => + #{ description => <<"Get the gateway client subscriptions">> + , parameters => params_client_insta() + , responses => + ?STANDARD_RESP( + #{ 200 => emqx_dashboard_swagger:schema_with_examples( + hoconsc:array(ref(subscription)), + examples_subsctiption_list())}) + } + , post => + #{ description => <<"Create a subscription membership">> + , parameters => params_client_insta() + %% FIXME: + , requestBody => emqx_dashboard_swagger:schema_with_examples( + ref(subscription), + examples_subsctiption()) + , responses => + ?STANDARD_RESP( + #{ 201 => emqx_dashboard_swagger:schema_with_examples( + ref(subscription), + examples_subsctiption())}) + } }; -swagger("/gateway/:name/clients/:clientid/subscriptions", get) -> - #{ description => <<"Get the gateway client subscriptions">> - , parameters => params_client_insta() - , responses => - #{ <<"400">> => schema_bad_request() - , <<"404">> => schema_not_found() - , <<"500">> => schema_internal_error() - , <<"200">> => schema_subscription_list() - } - }; -swagger("/gateway/:name/clients/:clientid/subscriptions", post) -> - #{ description => <<"Get the gateway client subscriptions">> - , parameters => params_client_insta() - , requestBody => schema_subscription() - , responses => - #{ <<"400">> => schema_bad_request() - , <<"404">> => schema_not_found() - , <<"500">> => schema_internal_error() - , <<"204">> => schema_no_content() - } - }; -swagger("/gateway/:name/clients/:clientid/subscriptions/:topic", delete) -> - #{ description => <<"Unsubscribe the topic for client">> - , parameters => params_topic_name_in_path() ++ params_client_insta() - , responses => - #{ <<"400">> => schema_bad_request() - , <<"404">> => schema_not_found() - , <<"500">> => schema_internal_error() - , <<"204">> => schema_no_content() - } +schema("/gateway/:name/clients/:clientid/subscriptions/:topic") -> + #{ 'operationId' => subscriptions + , delete => + #{ description => <<"Delete a subscriptions membership">> + , parameters => params_topic_name_in_path() ++ params_client_insta() + , responses => + ?STANDARD_RESP(#{204 => <<"Unsubscribed">>}) + } }. params_client_query() -> params_gateway_name_in_path() ++ params_client_searching_in_qs() - ++ emqx_mgmt_util:page_params(). + ++ params_paging(). params_client_insta() -> params_clientid_in_path() ++ params_gateway_name_in_path(). params_client_searching_in_qs() -> - queries( - [ {node, string} - , {clientid, string} - , {username, string} - , {ip_address, string} - , {conn_state, string} - , {proto_ver, string} - , {clean_start, boolean} - , {like_clientid, string} - , {like_username, string} - , {gte_created_at, string} - , {lte_created_at, string} - , {gte_connected_at, string} - , {lte_connected_at, string} - ]). + M = #{in => query, nullable => true}, + [ {node, + mk(binary(), + M#{desc => <<"Match the client's node name">>})} + , {clientid, + mk(binary(), + M#{desc => <<"Match the client's ID">>})} + , {username, + mk(binary(), + M#{desc => <<"Match the client's Username">>})} + , {ip_address, + mk(binary(), + M#{desc => <<"Match the client's ip address">>})} + , {conn_state, + mk(binary(), + M#{desc => <<"Match the client's connection state">>})} + , {proto_ver, + mk(binary(), + M#{desc => <<"Match the client's protocol version">>})} + , {clean_start, + mk(boolean(), + M#{desc => <<"Match the client's clean start flag">>})} + , {like_clientid, + mk(binary(), + M#{desc => <<"Use sub-string to match client's ID">>})} + , {like_username, + mk(binary(), + M#{desc => <<"Use sub-string to match client's username">>})} + , {gte_created_at, + mk(binary(), + M#{desc => <<"Match the session created datetime greater than " + "a certain value">>})} + , {lte_created_at, + mk(binary(), + M#{desc => <<"Match the session created datetime less than " + "a certain value">>})} + , {gte_connected_at, + mk(binary(), + M#{desc => <<"Match the client socket connected datetime greater " + "than a certain value">>})} + , {lte_connected_at, + mk(binary(), + M#{desc => <<"Match the client socket connected datatime less than " + " a certain value">>})} + ]. + +params_paging() -> + [ {page, + mk(integer(), + #{ in => query + , nullable => true + , desc => <<"Page Index">>})} + , {limit, + mk(integer(), + #{ in => query + , desc => <<"Page Limit">> + , nullable => true})} + ]. params_gateway_name_in_path() -> - [#{ name => name - , in => path - , schema => #{type => string} - , required => true - }]. + [{name, + mk(binary(), + #{ in => path + , desc => <<"Gateway Name">> + })} + ]. params_clientid_in_path() -> - [#{ name => clientid - , in => path - , schema => #{type => string} - , required => true - }]. + [{clientid, + mk(binary(), + #{ in => path + , desc => <<"Client ID">> + })} + ]. params_topic_name_in_path() -> - [#{ name => topic - , in => path - , schema => #{type => string} - , required => true - }]. - -queries(Ls) -> - lists:map(fun({K, Type}) -> - #{name => K, in => query, - schema => #{type => Type}, - required => false - } - end, Ls). + [{topic, + mk(binary(), + #{ in => path + , desc => <<"Topic Filter/Name">> + })} + ]. %%-------------------------------------------------------------------- %% schemas -schema_clients_list() -> - emqx_mgmt_util:page_schema( - #{ type => object - , properties => properties_client() - } - ). +roots() -> + [ client + , subscription + ]. -schema_client() -> - emqx_mgmt_util:schema( - #{ type => object - , properties => properties_client() - }). - -schema_subscription_list() -> - emqx_mgmt_util:array_schema( - #{ type => object - , properties => properties_subscription() - }, - <<"Client subscriptions">> - ). - -schema_subscription() -> - emqx_mgmt_util:schema( - #{ type => object - , properties => properties_subscription() - } - ). +fields(client) -> + %% XXX: enum for every protocol's client + [ {node, + mk(string(), + #{ desc => <<"Name of the node to which the client is " + "connected">>})} + , {clientid, + mk(string(), + #{ desc => <<"Client identifier">>})} + , {username, + mk(string(), + #{ desc => <<"Username of client when connecting">>})} + , {proto_name, + mk(string(), + #{ desc => <<"Client protocol name">>})} + , {proto_ver, + mk(string(), + #{ desc => <<"Protocol version used by the client">>})} + , {ip_address, + mk(string(), + #{ desc => <<"Client's IP address">>})} + , {port, + mk(integer(), + #{ desc => <<"Client's port">>})} + , {is_bridge, + mk(boolean(), + #{ desc => <<"Indicates whether the client is connected via " + "bridge">>})} + , {connected_at, + mk(string(), + #{ desc => <<"Client connection time">>})} + , {disconnected_at, + mk(string(), + #{ desc => <<"Client offline time, This field is only valid and " + "returned when connected is false">>})} + , {connected, + mk(boolean(), + #{ desc => <<"Whether the client is connected">>})} + %% FIXME: the will_msg attribute is not a general attribute + %% for every protocol. But it should be returned to frontend if someone + %% want it + %% + %, {will_msg, + % mk(string(), + % #{ desc => <<"Client will message">>})} + %, {zone, + % mk(string(), + % #{ desc => <<"Indicate the configuration group used by the " + % "client">>})} + , {keepalive, + mk(integer(), + #{ desc => <<"keepalive time, with the unit of second">>})} + , {clean_start, + mk(boolean(), + #{ desc => <<"Indicate whether the client is using a brand " + "new session">>})} + , {expiry_interval, + mk(integer(), + #{ desc => <<"Session expiration interval, with the unit of " + "second">>})} + , {created_at, + mk(string(), + #{ desc => <<"Session creation time">>})} + , {subscriptions_cnt, + mk(integer(), + #{ desc => <<"Number of subscriptions established by this " + "client">>})} + , {subscriptions_max, + mk(integer(), + #{ desc => <<"Maximum number of subscriptions allowed by this " + "client">>})} + , {inflight_cnt, + mk(integer(), + #{ desc => <<"Current length of inflight">>})} + , {inflight_max, + mk(integer(), + #{ desc => <<"Maximum length of inflight">>})} + , {mqueue_len, + mk(integer(), + #{ desc => <<"Current length of message queue">>})} + , {mqueue_max, + mk(integer(), + #{ desc => <<"Maximum length of message queue">>})} + , {mqueue_dropped, + mk(integer(), + #{ desc => <<"Number of messages dropped by the message queue " + "due to exceeding the length">>})} + , {awaiting_rel_cnt, + mk(integer(), + #{ desc => <<"Number of awaiting PUBREC packet">>})} + , {awaiting_rel_max, + mk(integer(), + #{ desc => <<"Maximum allowed number of awaiting PUBREC " + "packet">>})} + , {recv_oct, + mk(integer(), + #{ desc => <<"Number of bytes received by EMQ X Broker">>})} + , {recv_cnt, + mk(integer(), + #{ desc => <<"Number of TCP packets received">>})} + , {recv_pkt, + mk(integer(), + #{ desc => <<"Number of MQTT packets received">>})} + , {recv_msg, + mk(integer(), + #{ desc => <<"Number of PUBLISH packets received">>})} + , {send_oct, + mk(integer(), + #{ desc => <<"Number of bytes sent">>})} + , {send_cnt, + mk(integer(), + #{ desc => <<"Number of TCP packets sent">>})} + , {send_pkt, + mk(integer(), + #{ desc => <<"Number of MQTT packets sent">>})} + , {send_msg, + mk(integer(), + #{ desc => <<"Number of PUBLISH packets sent">>})} + , {mailbox_len, + mk(integer(), + #{ desc => <<"Process mailbox size">>})} + , {heap_size, + mk(integer(), + #{ desc => <<"Process heap size with the unit of byte">>})} + , {reductions, + mk(integer(), + #{ desc => <<"Erlang reduction">>})} + ]; +fields(subscription) -> + [ {topic, + mk(string(), + #{ desc => <<"Topic Fillter">>})} + , {qos, + mk(integer(), + #{ desc => <<"QoS level, enum: 0, 1, 2">>})} + , {nl, + mk(integer(), %% FIXME: why not boolean? + #{ desc => <<"No Local option, enum: 0, 1">>})} + , {rap, + mk(integer(), + #{ desc => <<"Retain as Published option, enum: 0, 1">>})} + , {rh, + mk(integer(), + #{ desc => <<"Retain Handling option, enum: 0, 1, 2">>})} + , {sub_props, + mk(ref(extra_sub_props), + #{desc => <<"Subscription properties">>})} + ]; +fields(extra_sub_props) -> + [ {subid, + mk(string(), + #{ desc => <<"Only stomp protocol, an uniquely identity for " + "the subscription. range: 1-65535.">>})} + ]. %%-------------------------------------------------------------------- -%% properties defines +%% examples -properties_client() -> - %% FIXME: enum for every protocol's client - emqx_mgmt_util:properties( - [ {node, string, - <<"Name of the node to which the client is connected">>} - , {clientid, string, - <<"Client identifier">>} - , {username, string, - <<"Username of client when connecting">>} - , {proto_name, string, - <<"Client protocol name">>} - , {proto_ver, string, - <<"Protocol version used by the client">>} - , {ip_address, string, - <<"Client's IP address">>} - , {port, integer, - <<"Client's port">>} - , {is_bridge, boolean, - <<"Indicates whether the client is connectedvia bridge">>} - , {connected_at, string, - <<"Client connection time">>} - , {disconnected_at, string, - <<"Client offline time, This field is only valid and returned " - "when connected is false">>} - , {connected, boolean, - <<"Whether the client is connected">>} - %% FIXME: the will_msg attribute is not a general attribute - %% for every protocol. But it should be returned to frontend if someone - %% want it - %% - %, {will_msg, string, - % <<"Client will message">>} - %, {zone, string, - % <<"Indicate the configuration group used by the client">>} - , {keepalive, integer, - <<"keepalive time, with the unit of second">>} - , {clean_start, boolean, - <<"Indicate whether the client is using a brand new session">>} - , {expiry_interval, integer, - <<"Session expiration interval, with the unit of second">>} - , {created_at, string, - <<"Session creation time">>} - , {subscriptions_cnt, integer, - <<"Number of subscriptions established by this client">>} - , {subscriptions_max, integer, - <<"v4 api name [max_subscriptions] Maximum number of " - "subscriptions allowed by this client">>} - , {inflight_cnt, integer, - <<"Current length of inflight">>} - , {inflight_max, integer, - <<"v4 api name [max_inflight]. Maximum length of inflight">>} - , {mqueue_len, integer, - <<"Current length of message queue">>} - , {mqueue_max, integer, - <<"v4 api name [max_mqueue]. Maximum length of message queue">>} - , {mqueue_dropped, integer, - <<"Number of messages dropped by the message queue due to " - "exceeding the length">>} - , {awaiting_rel_cnt, integer, - <<"v4 api name [awaiting_rel] Number of awaiting PUBREC packet">>} - , {awaiting_rel_max, integer, - <<"v4 api name [max_awaiting_rel]. Maximum allowed number of " - "awaiting PUBREC packet">>} - , {recv_oct, integer, - <<"Number of bytes received by EMQ X Broker (the same below)">>} - , {recv_cnt, integer, - <<"Number of TCP packets received">>} - , {recv_pkt, integer, - <<"Number of MQTT packets received">>} - , {recv_msg, integer, - <<"Number of PUBLISH packets received">>} - , {send_oct, integer, - <<"Number of bytes sent">>} - , {send_cnt, integer, - <<"Number of TCP packets sent">>} - , {send_pkt, integer, - <<"Number of MQTT packets sent">>} - , {send_msg, integer, - <<"Number of PUBLISH packets sent">>} - , {mailbox_len, integer, - <<"Process mailbox size">>} - , {heap_size, integer, - <<"Process heap size with the unit of byte">>} - , {reductions, integer, - <<"Erlang reduction">>} - ]). +examples_client_list() -> + #{}. -properties_subscription() -> - ExtraProps = [ {subid, string, - <<"Only stomp protocol, an uniquely identity for " - "the subscription. range: 1-65535.">>} - ], - emqx_mgmt_util:properties( - [ {topic, string, - <<"Topic Fillter">>} - , {qos, integer, - <<"QoS level, enum: 0, 1, 2">>} - , {nl, integer, %% FIXME: why not boolean? - <<"No Local option, enum: 0, 1">>} - , {rap, integer, - <<"Retain as Published option, enum: 0, 1">>} - , {rh, integer, - <<"Retain Handling option, enum: 0, 1, 2">>} - , {sub_props, object, ExtraProps} - ]). +examples_client() -> + #{}. + +examples_subsctiption_list() -> + #{}. + +examples_subsctiption() -> + #{}. diff --git a/apps/emqx_gateway/src/emqx_gateway_api_listeners.erl b/apps/emqx_gateway/src/emqx_gateway_api_listeners.erl index f1744363c..e73dd707e 100644 --- a/apps/emqx_gateway/src/emqx_gateway_api_listeners.erl +++ b/apps/emqx_gateway/src/emqx_gateway_api_listeners.erl @@ -18,14 +18,10 @@ -behaviour(minirest_api). +-include("emqx_gateway_http.hrl"). -include_lib("typerefl/include/types.hrl"). --define(BAD_REQUEST, 'BAD_REQUEST'). --define(NOT_FOUND, 'NOT_FOUND'). --define(INTERNAL_ERROR, 'INTERNAL_SERVER_ERROR'). - -import(hoconsc, [mk/2, ref/1, ref/2]). --import(emqx_dashboard_swagger, [error_codes/2]). -import(emqx_gateway_http, [ return_http_error/2 @@ -93,8 +89,9 @@ listeners(post, #{bindings := #{name := Name0}, body := LConf}) -> undefined -> ListenerId = emqx_gateway_utils:listener_id( GwName, Type, LName), - ok = emqx_gateway_http:add_listener(ListenerId, LConf), - {204}; + {ok, RespConf} = emqx_gateway_http:add_listener( + ListenerId, LConf), + {201, RespConf}; _ -> return_http_error(400, "Listener name has occupied") end @@ -123,8 +120,8 @@ listeners_insta(put, #{body := LConf, }) -> ListenerId = emqx_mgmt_util:urldecode(ListenerId0), with_gateway(Name0, fun(_GwName, _) -> - ok = emqx_gateway_http:update_listener(ListenerId, LConf), - {204} + {ok, RespConf} = emqx_gateway_http:update_listener(ListenerId, LConf), + {200, RespConf} end). listeners_insta_authn(get, #{bindings := #{name := Name0, @@ -145,16 +142,17 @@ listeners_insta_authn(post, #{body := Conf, id := ListenerId0}}) -> ListenerId = emqx_mgmt_util:urldecode(ListenerId0), with_gateway(Name0, fun(GwName, _) -> - ok = emqx_gateway_http:add_authn(GwName, ListenerId, Conf), - {204} + {ok, Authn} = emqx_gateway_http:add_authn(GwName, ListenerId, Conf), + {201, Authn} end); listeners_insta_authn(put, #{body := Conf, bindings := #{name := Name0, id := ListenerId0}}) -> ListenerId = emqx_mgmt_util:urldecode(ListenerId0), with_gateway(Name0, fun(GwName, _) -> - ok = emqx_gateway_http:update_authn(GwName, ListenerId, Conf), - {204} + {ok, Authn} = emqx_gateway_http:update_authn( + GwName, ListenerId, Conf), + {200, Authn} end); listeners_insta_authn(delete, #{bindings := #{name := Name0, id := ListenerId0}}) -> @@ -226,14 +224,11 @@ schema("/gateway/:name/listeners") -> #{ description => <<"Get the gateway listeners">> , parameters => params_gateway_name_in_path() , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 200 => emqx_dashboard_swagger:schema_with_examples( - hoconsc:array(ref(listener)), - examples_listener_list()) - } + ?STANDARD_RESP( + #{ 200 => emqx_dashboard_swagger:schema_with_example( + hoconsc:array(ref(listener)), + examples_listener_list()) + }) }, post => #{ description => <<"Create the gateway listener">> @@ -242,12 +237,11 @@ schema("/gateway/:name/listeners") -> ref(listener), examples_listener()) , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 204 => <<"Created">> - } + ?STANDARD_RESP( + #{ 201 => emqx_dashboard_swagger:schema_with_examples( + ref(listener), + examples_listener()) + }) } }; schema("/gateway/:name/listeners/:id") -> @@ -257,26 +251,18 @@ schema("/gateway/:name/listeners/:id") -> , parameters => params_gateway_name_in_path() ++ params_listener_id_in_path() , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 200 => emqx_dashboard_swagger:schema_with_examples( - ref(listener), - examples_listener()) - } + ?STANDARD_RESP( + #{ 200 => emqx_dashboard_swagger:schema_with_examples( + ref(listener), + examples_listener()) + }) }, delete => #{ description => <<"Delete the gateway listener">> , parameters => params_gateway_name_in_path() ++ params_listener_id_in_path() , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 204 => <<"Deleted">> - } + ?STANDARD_RESP(#{204 => <<"Deleted">>}) }, put => #{ description => <<"Update the gateway listener">> @@ -286,12 +272,11 @@ schema("/gateway/:name/listeners/:id") -> ref(listener), examples_listener()) , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 200 => <<"Updated">> - } + ?STANDARD_RESP( + #{ 200 => emqx_dashboard_swagger:schema_with_examples( + ref(listener), + examples_listener()) + }) } }; schema("/gateway/:name/listeners/:id/authentication") -> @@ -301,13 +286,10 @@ schema("/gateway/:name/listeners/:id/authentication") -> , parameters => params_gateway_name_in_path() ++ params_listener_id_in_path() , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 200 => schema_authn() - , 204 => <<"Authentication does not initiated">> - } + ?STANDARD_RESP( + #{ 200 => schema_authn() + , 204 => <<"Authentication does not initiated">> + }) }, post => #{ description => <<"Add authentication for the listener">> @@ -315,12 +297,7 @@ schema("/gateway/:name/listeners/:id/authentication") -> ++ params_listener_id_in_path() , 'requestBody' => schema_authn() , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 204 => <<"Added">> - } + ?STANDARD_RESP(#{201 => schema_authn()}) }, put => #{ description => <<"Update authentication for the listener">> @@ -328,24 +305,14 @@ schema("/gateway/:name/listeners/:id/authentication") -> ++ params_listener_id_in_path() , 'requestBody' => schema_authn() , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 204 => <<"Updated">> - } + ?STANDARD_RESP(#{200 => schema_authn()}) }, delete => #{ description => <<"Remove authentication for the listener">> , parameters => params_gateway_name_in_path() ++ params_listener_id_in_path() , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 204 => <<"Deleted">> - } + ?STANDARD_RESP(#{200 => <<"Deleted">>}) } }; schema("/gateway/:name/listeners/:id/authentication/users") -> @@ -356,14 +323,11 @@ schema("/gateway/:name/listeners/:id/authentication/users") -> params_listener_id_in_path() ++ params_paging_in_qs() , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 200 => emqx_dashboard_swagger:schema_with_example( - ref(emqx_authn_api, response_user), - emqx_authn_api:response_user_examples()) - } + ?STANDARD_RESP( + #{ 200 => emqx_dashboard_swagger:schema_with_example( + ref(emqx_authn_api, response_user), + emqx_authn_api:response_user_examples()) + }) }, post => #{ description => <<"Add user for the authentication">> @@ -373,14 +337,11 @@ schema("/gateway/:name/listeners/:id/authentication/users") -> ref(emqx_authn_api, request_user_create), emqx_authn_api:request_user_create_examples()) , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 201 => emqx_dashboard_swagger:schema_with_example( - ref(emqx_authn_api, response_user), - emqx_authn_api:response_user_examples()) - } + ?STANDARD_RESP( + #{ 201 => emqx_dashboard_swagger:schema_with_example( + ref(emqx_authn_api, response_user), + emqx_authn_api:response_user_examples()) + }) } }; schema("/gateway/:name/listeners/:id/authentication/users/:uid") -> @@ -392,14 +353,11 @@ schema("/gateway/:name/listeners/:id/authentication/users/:uid") -> params_listener_id_in_path() ++ params_userid_in_path() , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 200 => emqx_dashboard_swagger:schema_with_example( - ref(emqx_authn_api, response_user), - emqx_authn_api:response_user_examples()) - } + ?STANDARD_RESP( + #{ 200 => emqx_dashboard_swagger:schema_with_example( + ref(emqx_authn_api, response_user), + emqx_authn_api:response_user_examples()) + }) }, put => #{ description => <<"Update the user info for the gateway " @@ -411,14 +369,11 @@ schema("/gateway/:name/listeners/:id/authentication/users/:uid") -> ref(emqx_authn_api, request_user_update), emqx_authn_api:request_user_update_examples()) , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 200 => emqx_dashboard_swagger:schema_with_example( - ref(emqx_authn_api, response_user), - emqx_authn_api:response_user_examples()) - } + ?STANDARD_RESP( + #{ 200 => emqx_dashboard_swagger:schema_with_example( + ref(emqx_authn_api, response_user), + emqx_authn_api:response_user_examples()) + }) }, delete => #{ description => <<"Delete the user for the gateway " @@ -427,14 +382,7 @@ schema("/gateway/:name/listeners/:id/authentication/users/:uid") -> params_listener_id_in_path() ++ params_userid_in_path() , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - , 200 => emqx_dashboard_swagger:schema_with_example( - ref(emqx_authn_api, response_user), - emqx_authn_api:response_user_examples()) - } + ?STANDARD_RESP(#{204 => <<"Deleted">>}) } }; schema("/gateway/:name/listeners/:id/authentication/import_users") -> @@ -448,13 +396,7 @@ schema("/gateway/:name/listeners/:id/authentication/import_users") -> emqx_authn_api:request_import_users_examples() ) , responses => - #{ 400 => error_codes([?BAD_REQUEST], <<"Bad Request">>) - , 404 => error_codes([?NOT_FOUND], <<"Not Found">>) - , 500 => error_codes([?INTERNAL_ERROR], - <<"Ineternal Server Error">>) - %% XXX: Put a hint message into 204 return ? - , 204 => <<"Imported">> - } + ?STANDARD_RESP(#{204 => <<"Imported">>}) } }. @@ -638,7 +580,96 @@ common_listener_opts() -> %% examples examples_listener_list() -> - #{stomp_listeners => [examples_listener()]}. + [Config || #{value := Config} <- maps:values(examples_listener())]. examples_listener() -> - #{}. + #{ tcp_listener=> + #{ summary => <<"A simple tcp listener example">> + , value => + #{ bind => <<"61613">> + , acceptors => 16 + , max_connections => 1024000 + , max_conn_rate => 1000 + , tcp => + #{ active_n => 100 + , backlog => 1024 + , send_timeout => <<"15s">> + , send_timeout_close => true + , recbuf => <<"10KB">> + , sndbuf => <<"10KB">> + , buffer => <<"10KB">> + , high_watermark => <<"1MB">> + , nodelay => false + , reuseaddr => true + } + } + } + , ssl_listener => + #{ summary => <<"A simple ssl listener example">> + , value => + #{ bind => <<"61614">> + , acceptors => 16 + , max_connections => 1024000 + , max_conn_rate => 1000 + , access_rules => [<<"allow all">>] + , ssl => + #{ versions => [<<"tlsv1.3">>, <<"tlsv1.2">>, + <<"tlsv1.1">>, <<"tlsv1">>] + , cacertfile => <<"etc/certs/cacert.pem">> + , certfile => <<"etc/certs/cert.pem">> + , keyfile => <<"etc/certs/key.pem">> + , verify => <<"verify_none">> + , fail_if_no_peer_cert => false + , server_name_indication => disable + } + , tcp => + #{ active_n => 100 + , backlog => 1024 + } + } + } + , udp_listener => + #{ summary => <<"A simple udp listener example">> + , value => + #{ bind => <<"0.0.0.0:1884">> + , udp => + #{ active_n => 100 + , recbuf => <<"10KB">> + , sndbuf => <<"10KB">> + , buffer => <<"10KB">> + , reuseaddr => true + } + } + } + , dtls_listener => + #{ summary => <<"A simple dtls listener example">> + , value => + #{ bind => <<"5684">> + , acceptors => 16 + , max_connections => 1024000 + , max_conn_rate => 1000 + , access_rules => [<<"allow all">>] + , ssl => + #{ versions => [<<"dtlsv1.2">>, <<"dtlsv1">>] + , cacertfile => <<"etc/certs/cacert.pem">> + , certfile => <<"etc/certs/cert.pem">> + , keyfile => <<"etc/certs/key.pem">> + , verify => <<"verify_none">> + , fail_if_no_peer_cert => false + , server_name_indication => disable + } + , tcp => + #{ active_n => 100 + , backlog => 1024 + } + } + } + , dtls_listener_with_psk_ciphers => + #{ summary => <<"todo">> + , value => + #{} + } + , lisetner_with_authn => + #{ summary => <<"todo">> + , value => #{}} + }. diff --git a/apps/emqx_gateway/src/emqx_gateway_conf.erl b/apps/emqx_gateway/src/emqx_gateway_conf.erl index e97b0062d..a799b7fb7 100644 --- a/apps/emqx_gateway/src/emqx_gateway_conf.erl +++ b/apps/emqx_gateway/src/emqx_gateway_conf.erl @@ -59,7 +59,8 @@ -define(AUTHN_BIN, ?EMQX_AUTHENTICATION_CONFIG_ROOT_NAME_BINARY). -type atom_or_bin() :: atom() | binary(). --type ok_or_err() :: ok_or_err(). +-type ok_or_err() :: ok | {error, term()}. +-type map_or_err() :: {ok, map()} | {error, term()}. -type listener_ref() :: {ListenerType :: atom_or_bin(), ListenerName :: atom_or_bin()}. @@ -85,7 +86,8 @@ load_gateway(GwName, Conf) -> {Ls, Conf1} -> Conf1#{<<"listeners">> => unconvert_listeners(Ls)} end, - update({?FUNCTION_NAME, bin(GwName), NConf}). + %% TODO: + ret_ok_err(update({?FUNCTION_NAME, bin(GwName), NConf})). %% @doc convert listener array to map unconvert_listeners(Ls) when is_list(Ls) -> @@ -111,13 +113,14 @@ update_gateway(GwName, Conf0) -> Exclude0 = [listeners, ?EMQX_AUTHENTICATION_CONFIG_ROOT_NAME_ATOM], Exclude1 = [atom_to_binary(K, utf8) || K <- Exclude0], Conf = maps:without(Exclude0 ++ Exclude1, Conf0), - update({?FUNCTION_NAME, bin(GwName), Conf}). + + ret_ok_err(update({?FUNCTION_NAME, bin(GwName), Conf})). %% FIXME: delete cert files ?? -spec unload_gateway(atom_or_bin()) -> ok_or_err(). unload_gateway(GwName) -> - update({?FUNCTION_NAME, bin(GwName)}). + ret_ok_err(update({?FUNCTION_NAME, bin(GwName)})). %% @doc Get the gateway configurations. %% Missing fields are filled with default values. This function is typically @@ -139,18 +142,20 @@ convert_listeners(GwName, Ls) when is_map(Ls) -> lists:append([do_convert_listener(GwName, Type, maps:to_list(Conf)) || {Type, Conf} <- maps:to_list(Ls)]). -do_convert_listener(GwName, Type, Conf) -> - [begin - ListenerId = emqx_gateway_utils:listener_id(GwName, Type, LName), - Running = emqx_gateway_utils:is_running(ListenerId, LConf), - bind2str( - LConf#{ - id => ListenerId, - type => Type, - name => LName, - running => Running - }) - end || {LName, LConf} <- Conf, is_map(LConf)]. +do_convert_listener(GwName, LType, Conf) -> + [ do_convert_listener2(GwName, LType, LName, LConf) + || {LName, LConf} <- Conf, is_map(LConf)]. + +do_convert_listener2(GwName, LType, LName, LConf) -> + ListenerId = emqx_gateway_utils:listener_id(GwName, LType, LName), + Running = emqx_gateway_utils:is_running(ListenerId, LConf), + bind2str( + LConf#{ + id => ListenerId, + type => LType, + name => LName, + running => Running + }). bind2str(LConf = #{bind := Bind}) when is_integer(Bind) -> maps:put(bind, integer_to_binary(Bind), LConf); @@ -194,48 +199,56 @@ listener(ListenerId) -> {error, Reason} end. --spec add_listener(atom_or_bin(), listener_ref(), map()) -> ok_or_err(). +-spec add_listener(atom_or_bin(), listener_ref(), map()) -> map_or_err(). add_listener(GwName, ListenerRef, Conf) -> - update({?FUNCTION_NAME, bin(GwName), bin(ListenerRef), Conf}). + ret_listener_or_err( + GwName, ListenerRef, + update({?FUNCTION_NAME, bin(GwName), bin(ListenerRef), Conf})). --spec update_listener(atom_or_bin(), listener_ref(), map()) -> ok_or_err(). +-spec update_listener(atom_or_bin(), listener_ref(), map()) -> map_or_err(). update_listener(GwName, ListenerRef, Conf) -> - update({?FUNCTION_NAME, bin(GwName), bin(ListenerRef), Conf}). + ret_listener_or_err( + GwName, ListenerRef, + update({?FUNCTION_NAME, bin(GwName), bin(ListenerRef), Conf})). -spec remove_listener(atom_or_bin(), listener_ref()) -> ok_or_err(). remove_listener(GwName, ListenerRef) -> - update({?FUNCTION_NAME, bin(GwName), bin(ListenerRef)}). + ret_ok_err(update({?FUNCTION_NAME, bin(GwName), bin(ListenerRef)})). --spec add_authn(atom_or_bin(), map()) -> ok_or_err(). +-spec add_authn(atom_or_bin(), map()) -> map_or_err(). add_authn(GwName, Conf) -> - update({?FUNCTION_NAME, bin(GwName), Conf}). + ret_authn(GwName, update({?FUNCTION_NAME, bin(GwName), Conf})). --spec add_authn(atom_or_bin(), listener_ref(), map()) -> ok_or_err(). +-spec add_authn(atom_or_bin(), listener_ref(), map()) -> map_or_err(). add_authn(GwName, ListenerRef, Conf) -> - update({?FUNCTION_NAME, bin(GwName), bin(ListenerRef), Conf}). + ret_authn( + GwName, ListenerRef, + update({?FUNCTION_NAME, bin(GwName), bin(ListenerRef), Conf})). --spec update_authn(atom_or_bin(), map()) -> ok_or_err(). +-spec update_authn(atom_or_bin(), map()) -> map_or_err(). update_authn(GwName, Conf) -> - update({?FUNCTION_NAME, bin(GwName), Conf}). + ret_authn(GwName, update({?FUNCTION_NAME, bin(GwName), Conf})). --spec update_authn(atom_or_bin(), listener_ref(), map()) -> ok_or_err(). +-spec update_authn(atom_or_bin(), listener_ref(), map()) -> map_or_err(). update_authn(GwName, ListenerRef, Conf) -> - update({?FUNCTION_NAME, bin(GwName), bin(ListenerRef), Conf}). + ret_authn( + GwName, ListenerRef, + update({?FUNCTION_NAME, bin(GwName), bin(ListenerRef), Conf})). -spec remove_authn(atom_or_bin()) -> ok_or_err(). remove_authn(GwName) -> - update({?FUNCTION_NAME, bin(GwName)}). + ret_ok_err(update({?FUNCTION_NAME, bin(GwName)})). -spec remove_authn(atom_or_bin(), listener_ref()) -> ok_or_err(). remove_authn(GwName, ListenerRef) -> - update({?FUNCTION_NAME, bin(GwName), bin(ListenerRef)}). + ret_ok_err(update({?FUNCTION_NAME, bin(GwName), bin(ListenerRef)})). %% @private update(Req) -> res(emqx_conf:update([gateway], Req, #{override_to => cluster})). -res({ok, _Result}) -> ok; -res({error, {pre_config_update, emqx_gateway_conf, Reason}}) -> {error, Reason}; +res({ok, Result}) -> {ok, Result}; +res({error, {pre_config_update,emqx_gateway_conf,Reason}}) -> {error, Reason}; res({error, Reason}) -> {error, Reason}. bin({LType, LName}) -> @@ -245,6 +258,32 @@ bin(A) when is_atom(A) -> bin(B) when is_binary(B) -> B. +ret_ok_err({ok, _}) -> ok; +ret_ok_err(Err) -> Err. + +ret_authn(GwName, {ok, #{raw_config := GwConf}}) -> + Authn = emqx_map_lib:deep_get( + [bin(GwName), <<"authentication">>], + GwConf), + {ok, Authn}; +ret_authn(_GwName, Err) -> Err. + +ret_authn(GwName, {LType, LName}, {ok, #{raw_config := GwConf}}) -> + Authn = emqx_map_lib:deep_get( + [bin(GwName), <<"listeners">>, bin(LType), + bin(LName), <<"authentication">>], + GwConf), + {ok, Authn}; +ret_authn(_, _, Err) -> Err. + +ret_listener_or_err(GwName, {LType, LName}, {ok, #{raw_config := GwConf}}) -> + LConf = emqx_map_lib:deep_get( + [bin(GwName), <<"listeners">>, bin(LType), bin(LName)], + GwConf), + {ok, do_convert_listener2(GwName, LType, LName, LConf)}; +ret_listener_or_err(_, _, Err) -> + Err. + %%-------------------------------------------------------------------- %% Config Handler %%-------------------------------------------------------------------- diff --git a/apps/emqx_gateway/src/emqx_gateway_http.erl b/apps/emqx_gateway/src/emqx_gateway_http.erl index 98344f968..2a9840a9c 100644 --- a/apps/emqx_gateway/src/emqx_gateway_http.erl +++ b/apps/emqx_gateway/src/emqx_gateway_http.erl @@ -146,14 +146,14 @@ get_listeners_status(GwName, Config) -> %% Mgmt APIs - listeners %%-------------------------------------------------------------------- --spec add_listener(atom() | binary(), map()) -> ok. +-spec add_listener(atom() | binary(), map()) -> {ok, map()}. add_listener(ListenerId, NewConf0) -> {GwName, Type, Name} = emqx_gateway_utils:parse_listener_id(ListenerId), NewConf = maps:without([<<"id">>, <<"name">>, <<"type">>, <<"running">>], NewConf0), confexp(emqx_gateway_conf:add_listener(GwName, {Type, Name}, NewConf)). --spec update_listener(atom() | binary(), map()) -> ok. +-spec update_listener(atom() | binary(), map()) -> {ok, map()}. update_listener(ListenerId, NewConf0) -> {GwName, Type, Name} = emqx_gateway_utils:parse_listener_id(ListenerId), @@ -194,23 +194,23 @@ wrap_chain_name(ChainName, Conf) -> Conf end. --spec add_authn(gateway_name(), map()) -> ok. +-spec add_authn(gateway_name(), map()) -> {ok, map()}. add_authn(GwName, AuthConf) -> confexp(emqx_gateway_conf:add_authn(GwName, AuthConf)). --spec add_authn(gateway_name(), binary(), map()) -> ok. +-spec add_authn(gateway_name(), binary(), map()) -> {ok, map()}. add_authn(GwName, ListenerId, AuthConf) -> - {_, Type, Name} = emqx_gateway_utils:parse_listener_id(ListenerId), - confexp(emqx_gateway_conf:add_authn(GwName, {Type, Name}, AuthConf)). + {_, LType, LName} = emqx_gateway_utils:parse_listener_id(ListenerId), + confexp(emqx_gateway_conf:add_authn(GwName, {LType, LName}, AuthConf)). --spec update_authn(gateway_name(), map()) -> ok. +-spec update_authn(gateway_name(), map()) -> {ok, map()}. update_authn(GwName, AuthConf) -> confexp(emqx_gateway_conf:update_authn(GwName, AuthConf)). --spec update_authn(gateway_name(), binary(), map()) -> ok. +-spec update_authn(gateway_name(), binary(), map()) -> {ok, map()}. update_authn(GwName, ListenerId, AuthConf) -> - {_, Type, Name} = emqx_gateway_utils:parse_listener_id(ListenerId), - confexp(emqx_gateway_conf:update_authn(GwName, {Type, Name}, AuthConf)). + {_, LType, LName} = emqx_gateway_utils:parse_listener_id(ListenerId), + confexp(emqx_gateway_conf:update_authn(GwName, {LType, LName}, AuthConf)). -spec remove_authn(gateway_name()) -> ok. remove_authn(GwName) -> @@ -218,10 +218,11 @@ remove_authn(GwName) -> -spec remove_authn(gateway_name(), binary()) -> ok. remove_authn(GwName, ListenerId) -> - {_, Type, Name} = emqx_gateway_utils:parse_listener_id(ListenerId), - confexp(emqx_gateway_conf:remove_authn(GwName, {Type, Name})). + {_, LType, LName} = emqx_gateway_utils:parse_listener_id(ListenerId), + confexp(emqx_gateway_conf:remove_authn(GwName, {LType, LName})). confexp(ok) -> ok; +confexp({ok, Res}) -> {ok, Res}; confexp({error, not_found}) -> error({update_conf_error, not_found}); confexp({error, already_exist}) -> diff --git a/apps/emqx_gateway/src/emqx_gateway_schema.erl b/apps/emqx_gateway/src/emqx_gateway_schema.erl index 5f87131d7..2215c3a96 100644 --- a/apps/emqx_gateway/src/emqx_gateway_schema.erl +++ b/apps/emqx_gateway/src/emqx_gateway_schema.erl @@ -143,9 +143,9 @@ The client just sends its PUBLISH messages to a GW" sc(hoconsc:array(ref(mqttsn_predefined)), #{ default => [] , desc => -"The Pre-defined topic ids and topic names.
+<<"The Pre-defined topic ids and topic names.
A 'pre-defined' topic id is a topic id whose mapping to a topic name -is known in advance by both the client's application and the gateway" +is known in advance by both the client’s application and the gateway">> })} , {listeners, sc(ref(udp_listeners))} ] ++ gateway_common_options(); diff --git a/apps/emqx_gateway/src/lwm2m/emqx_lwm2m_api.erl b/apps/emqx_gateway/src/lwm2m/emqx_lwm2m_api.erl index b174fdcc0..0366f499b 100644 --- a/apps/emqx_gateway/src/lwm2m/emqx_lwm2m_api.erl +++ b/apps/emqx_gateway/src/lwm2m/emqx_lwm2m_api.erl @@ -23,7 +23,7 @@ -export([lookup_cmd/2, observe/2, read/2, write/2]). --define(PATH(Suffix), "/gateway/lwm2m/:clientid"Suffix). +-define(PATH(Suffix), "/gateway/lwm2m/clients/:clientid"Suffix). -define(DATA_TYPE, ['Integer', 'Float', 'Time', 'String', 'Boolean', 'Opaque', 'Objlnk']). -import(hoconsc, [mk/2, ref/1, ref/2]). diff --git a/apps/emqx_gateway/test/emqx_coap_api_SUITE.erl b/apps/emqx_gateway/test/emqx_coap_api_SUITE.erl index 573a2550d..5e342ec76 100644 --- a/apps/emqx_gateway/test/emqx_coap_api_SUITE.erl +++ b/apps/emqx_gateway/test/emqx_coap_api_SUITE.erl @@ -40,7 +40,9 @@ gateway.coap { -define(HOST, "127.0.0.1"). -define(PORT, 5683). --define(CONN_URI, "coap://127.0.0.1/mqtt/connection?clientid=client1&username=admin&password=public"). +-define(CONN_URI, + "coap://127.0.0.1/mqtt/connection?clientid=client1&" + "username=admin&password=public"). -define(LOGT(Format, Args), ct:pal("TEST_SUITE: " ++ Format, Args)). @@ -67,7 +69,7 @@ end_per_suite(Config) -> t_send_request_api(_) -> ClientId = start_client(), timer:sleep(200), - Path = emqx_mgmt_api_test_util:api_path(["gateway/coap/client1/request"]), + Path = emqx_mgmt_api_test_util:api_path(["gateway/coap/clients/client1/request"]), Token = <<"atoken">>, Payload = <<"simple echo this">>, Req = #{token => Token, @@ -117,26 +119,40 @@ test_send_coap_request(UdpSock, Method, Content, Options, MsgId) -> Request = Request0#coap_message{id = MsgId}, ?LOGT("send_coap_request Request=~p", [Request]), RequestBinary = emqx_coap_frame:serialize_pkt(Request, undefined), - ?LOGT("test udp socket send to ~p:~p, data=~p", [IpAddr, Port, RequestBinary]), + ?LOGT("test udp socket send to ~p:~p, data=~p", + [IpAddr, Port, RequestBinary]), ok = gen_udp:send(UdpSock, IpAddr, Port, RequestBinary); {SchemeDiff, ChIdDiff, _, _} -> - error(lists:flatten(io_lib:format("scheme ~ts or ChId ~ts does not match with socket", [SchemeDiff, ChIdDiff]))) + error( + lists:flatten( + io_lib:format( + "scheme ~ts or ChId ~ts does not match with socket", + [SchemeDiff, ChIdDiff]) + )) end. test_recv_coap_response(UdpSock) -> {ok, {Address, Port, Packet}} = gen_udp:recv(UdpSock, 0, 2000), {ok, Response, _, _} = emqx_coap_frame:parse(Packet, undefined), - ?LOGT("test udp receive from ~p:~p, data1=~p, Response=~p", [Address, Port, Packet, Response]), - #coap_message{type = ack, method = Method, id=Id, token = Token, options = Options, payload = Payload} = Response, - ?LOGT("receive coap response Method=~p, Id=~p, Token=~p, Options=~p, Payload=~p", [Method, Id, Token, Options, Payload]), + ?LOGT("test udp receive from ~p:~p, data1=~p, Response=~p", + [Address, Port, Packet, Response]), + #coap_message{ + type = ack, method = Method, id = Id, + token = Token, options = Options, payload = Payload} = Response, + ?LOGT("receive coap response Method=~p, Id=~p, Token=~p, " + "Options=~p, Payload=~p", [Method, Id, Token, Options, Payload]), Response. test_recv_coap_request(UdpSock) -> case gen_udp:recv(UdpSock, 0) of {ok, {_Address, _Port, Packet}} -> {ok, Request, _, _} = emqx_coap_frame:parse(Packet, undefined), - #coap_message{type = con, method = Method, id=Id, token = Token, payload = Payload, options = Options} = Request, - ?LOGT("receive coap request Method=~p, Id=~p, Token=~p, Options=~p, Payload=~p", [Method, Id, Token, Options, Payload]), + #coap_message{ + type = con, method = Method, id = Id, + token = Token, payload = Payload, options = Options} = Request, + ?LOGT("receive coap request Method=~p, Id=~p, " + "Token=~p, Options=~p, Payload=~p", + [Method, Id, Token, Options, Payload]), Request; {error, Reason} -> ?LOGT("test_recv_coap_request failed, Reason=~p", [Reason]), diff --git a/apps/emqx_gateway/test/emqx_gateway_api_SUITE.erl b/apps/emqx_gateway/test/emqx_gateway_api_SUITE.erl index 8e25a4e6e..3ae9bcc12 100644 --- a/apps/emqx_gateway/test/emqx_gateway_api_SUITE.erl +++ b/apps/emqx_gateway/test/emqx_gateway_api_SUITE.erl @@ -196,12 +196,12 @@ t_authn(_) -> backend => <<"built-in-database">>, user_id_type => <<"clientid">> }, - {204, _} = request(post, "/gateway/stomp/authentication", AuthConf), + {201, _} = request(post, "/gateway/stomp/authentication", AuthConf), {200, ConfResp} = request(get, "/gateway/stomp/authentication"), assert_confs(AuthConf, ConfResp), AuthConf2 = maps:merge(AuthConf, #{user_id_type => <<"username">>}), - {204, _} = request(put, "/gateway/stomp/authentication", AuthConf2), + {200, _} = request(put, "/gateway/stomp/authentication", AuthConf2), {200, ConfResp2} = request(get, "/gateway/stomp/authentication"), assert_confs(AuthConf2, ConfResp2), @@ -219,7 +219,7 @@ t_authn_data_mgmt(_) -> backend => <<"built-in-database">>, user_id_type => <<"clientid">> }, - {204, _} = request(post, "/gateway/stomp/authentication", AuthConf), + {201, _} = request(post, "/gateway/stomp/authentication", AuthConf), {200, ConfResp} = request(get, "/gateway/stomp/authentication"), assert_confs(AuthConf, ConfResp), @@ -262,14 +262,14 @@ t_listeners(_) -> type => <<"tcp">>, bind => <<"61613">> }, - {204, _} = request(post, "/gateway/stomp/listeners", LisConf), + {201, _} = request(post, "/gateway/stomp/listeners", LisConf), {200, ConfResp} = request(get, "/gateway/stomp/listeners"), assert_confs([LisConf], ConfResp), {200, ConfResp1} = request(get, "/gateway/stomp/listeners/stomp:tcp:def"), assert_confs(LisConf, ConfResp1), LisConf2 = maps:merge(LisConf, #{bind => <<"61614">>}), - {204, _} = request( + {200, _} = request( put, "/gateway/stomp/listeners/stomp:tcp:def", LisConf2 @@ -298,12 +298,12 @@ t_listeners_authn(_) -> user_id_type => <<"clientid">> }, Path = "/gateway/stomp/listeners/stomp:tcp:def/authentication", - {204, _} = request(post, Path, AuthConf), + {201, _} = request(post, Path, AuthConf), {200, ConfResp2} = request(get, Path), assert_confs(AuthConf, ConfResp2), AuthConf2 = maps:merge(AuthConf, #{user_id_type => <<"username">>}), - {204, _} = request(put, Path, AuthConf2), + {200, _} = request(put, Path, AuthConf2), {200, ConfResp3} = request(get, Path), assert_confs(AuthConf2, ConfResp3), @@ -325,7 +325,7 @@ t_listeners_authn_data_mgmt(_) -> user_id_type => <<"clientid">> }, Path = "/gateway/stomp/listeners/stomp:tcp:def/authentication", - {204, _} = request(post, Path, AuthConf), + {201, _} = request(post, Path, AuthConf), {200, ConfResp2} = request(get, Path), assert_confs(AuthConf, ConfResp2), diff --git a/apps/emqx_gateway/test/emqx_gateway_conf_SUITE.erl b/apps/emqx_gateway/test/emqx_gateway_conf_SUITE.erl index 810f56e76..916913856 100644 --- a/apps/emqx_gateway/test/emqx_gateway_conf_SUITE.erl +++ b/apps/emqx_gateway/test/emqx_gateway_conf_SUITE.erl @@ -268,12 +268,12 @@ t_load_remove_authn(_) -> ok = emqx_gateway_conf:load_gateway(<<"stomp">>, StompConf), assert_confs(StompConf, emqx:get_raw_config([gateway, stomp])), - ok = emqx_gateway_conf:add_authn(<<"stomp">>, ?CONF_STOMP_AUTHN_1), + {ok, _} = emqx_gateway_conf:add_authn(<<"stomp">>, ?CONF_STOMP_AUTHN_1), assert_confs( maps:put(<<"authentication">>, ?CONF_STOMP_AUTHN_1, StompConf), emqx:get_raw_config([gateway, stomp])), - ok = emqx_gateway_conf:update_authn(<<"stomp">>, ?CONF_STOMP_AUTHN_2), + {ok, _} = emqx_gateway_conf:update_authn(<<"stomp">>, ?CONF_STOMP_AUTHN_2), assert_confs( maps:put(<<"authentication">>, ?CONF_STOMP_AUTHN_2, StompConf), emqx:get_raw_config([gateway, stomp])), @@ -295,14 +295,16 @@ t_load_remove_listeners(_) -> ok = emqx_gateway_conf:load_gateway(<<"stomp">>, StompConf), assert_confs(StompConf, emqx:get_raw_config([gateway, stomp])), - ok = emqx_gateway_conf:add_listener( - <<"stomp">>, {<<"tcp">>, <<"default">>}, ?CONF_STOMP_LISTENER_1), + {ok, _} = emqx_gateway_conf:add_listener( + <<"stomp">>, {<<"tcp">>, <<"default">>}, + ?CONF_STOMP_LISTENER_1), assert_confs( maps:merge(StompConf, listener(?CONF_STOMP_LISTENER_1)), emqx:get_raw_config([gateway, stomp])), - ok = emqx_gateway_conf:update_listener( - <<"stomp">>, {<<"tcp">>, <<"default">>}, ?CONF_STOMP_LISTENER_2), + {ok, _} = emqx_gateway_conf:update_listener( + <<"stomp">>, {<<"tcp">>, <<"default">>}, + ?CONF_STOMP_LISTENER_2), assert_confs( maps:merge(StompConf, listener(?CONF_STOMP_LISTENER_2)), emqx:get_raw_config([gateway, stomp])), @@ -339,12 +341,12 @@ t_load_remove_listener_authn(_) -> ok = emqx_gateway_conf:load_gateway(<<"stomp">>, StompConf), assert_confs(StompConf, emqx:get_raw_config([gateway, stomp])), - ok = emqx_gateway_conf:add_authn( - <<"stomp">>, {<<"tcp">>, <<"default">>}, ?CONF_STOMP_AUTHN_1), + {ok, _} = emqx_gateway_conf:add_authn( + <<"stomp">>, {<<"tcp">>, <<"default">>}, ?CONF_STOMP_AUTHN_1), assert_confs(StompConf1, emqx:get_raw_config([gateway, stomp])), - ok = emqx_gateway_conf:update_authn( - <<"stomp">>, {<<"tcp">>, <<"default">>}, ?CONF_STOMP_AUTHN_2), + {ok, _} = emqx_gateway_conf:update_authn( + <<"stomp">>, {<<"tcp">>, <<"default">>}, ?CONF_STOMP_AUTHN_2), assert_confs(StompConf2, emqx:get_raw_config([gateway, stomp])), ok = emqx_gateway_conf:remove_authn( @@ -403,14 +405,16 @@ t_add_listener_with_certs_content(_) -> ok = emqx_gateway_conf:load_gateway(<<"stomp">>, StompConf), assert_confs(StompConf, emqx:get_raw_config([gateway, stomp])), - ok = emqx_gateway_conf:add_listener( - <<"stomp">>, {<<"ssl">>, <<"default">>}, ?CONF_STOMP_LISTENER_SSL), + {ok, _} = emqx_gateway_conf:add_listener( + <<"stomp">>, {<<"ssl">>, <<"default">>}, + ?CONF_STOMP_LISTENER_SSL), assert_confs( maps:merge(StompConf, ssl_listener(?CONF_STOMP_LISTENER_SSL)), emqx:get_raw_config([gateway, stomp])), - ok = emqx_gateway_conf:update_listener( - <<"stomp">>, {<<"ssl">>, <<"default">>}, ?CONF_STOMP_LISTENER_SSL_2), + {ok, _} = emqx_gateway_conf:update_listener( + <<"stomp">>, {<<"ssl">>, <<"default">>}, + ?CONF_STOMP_LISTENER_SSL_2), assert_confs( maps:merge(StompConf, ssl_listener(?CONF_STOMP_LISTENER_SSL_2)), emqx:get_raw_config([gateway, stomp])), diff --git a/apps/emqx_gateway/test/emqx_lwm2m_api_SUITE.erl b/apps/emqx_gateway/test/emqx_lwm2m_api_SUITE.erl index 65c3ba2fb..1e53d0486 100644 --- a/apps/emqx_gateway/test/emqx_lwm2m_api_SUITE.erl +++ b/apps/emqx_gateway/test/emqx_lwm2m_api_SUITE.erl @@ -301,7 +301,7 @@ t_observe(Config) -> %%% Internal Functions %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% call_lookup_api(ClientId, Path, Action) -> - ApiPath = emqx_mgmt_api_test_util:api_path(["gateway/lwm2m", ClientId, "lookup_cmd"]), + ApiPath = emqx_mgmt_api_test_util:api_path(["gateway/lwm2m/clients", ClientId, "lookup_cmd"]), Auth = emqx_mgmt_api_test_util:auth_header_(), Query = io_lib:format("path=~ts&action=~ts", [Path, Action]), {ok, Response} = emqx_mgmt_api_test_util:request_api(get, ApiPath, Query, Auth), @@ -309,7 +309,7 @@ call_lookup_api(ClientId, Path, Action) -> Response. call_send_api(ClientId, Cmd, Query) -> - ApiPath = emqx_mgmt_api_test_util:api_path(["gateway/lwm2m", ClientId, Cmd]), + ApiPath = emqx_mgmt_api_test_util:api_path(["gateway/lwm2m/clients", ClientId, Cmd]), Auth = emqx_mgmt_api_test_util:auth_header_(), {ok, Response} = emqx_mgmt_api_test_util:request_api(post, ApiPath, Query, Auth), ?LOGT("rest api response:~ts~n", [Response]),