refactor(rule): generate swagger from hocon schema for /bridges
This commit is contained in:
Shawn
parent
affe69afd6
commit
56d46c80eb
|
|
@ -17,23 +17,32 @@
|
|||
|
||||
-behaviour(minirest_api).
|
||||
|
||||
-export([api_spec/0]).
|
||||
-include_lib("typerefl/include/types.hrl").
|
||||
|
||||
-export([ list_create_bridges_in_cluster/2
|
||||
, list_local_bridges/1
|
||||
, crud_bridges_in_cluster/2
|
||||
, manage_bridges/2
|
||||
-import(hoconsc, [mk/2, array/1, enum/1]).
|
||||
|
||||
%% Swagger specs from hocon schema
|
||||
-export([api_spec/0, paths/0, schema/1, namespace/0]).
|
||||
|
||||
%% API callbacks
|
||||
-export(['/bridges'/2, '/bridges/:id'/2,
|
||||
'/nodes/:node/bridges/:id/operation/:operation'/2]).
|
||||
|
||||
-export([ list_local_bridges/1
|
||||
, lookup_from_local_node/2
|
||||
]).
|
||||
|
||||
-define(TYPES, [mqtt, http]).
|
||||
|
||||
-define(CONN_TYPES, [mqtt]).
|
||||
|
||||
-define(TRY_PARSE_ID(ID, EXPR),
|
||||
try emqx_bridge:parse_bridge_id(Id) of
|
||||
{BridgeType, BridgeName} -> EXPR
|
||||
catch
|
||||
error:{invalid_bridge_id, Id0} ->
|
||||
{400, #{code => 'INVALID_ID', message => <<"invalid_bridge_id: ", Id0/binary,
|
||||
". Bridge ID must be of format 'bridge_type:name'">>}}
|
||||
". Bridge Ids must be of format {type}:{name}">>}}
|
||||
end).
|
||||
|
||||
-define(METRICS(MATCH, SUCC, FAILED, RATE, RATE_5, RATE_MAX),
|
||||
|
|
@ -53,184 +62,221 @@
|
|||
rate_max := RATE_MAX
|
||||
}).
|
||||
|
||||
req_schema() ->
|
||||
Schema = [
|
||||
case maps:to_list(emqx:get_raw_config([bridges, T], #{})) of
|
||||
%% the bridge is not configured, so we have no method to get the schema
|
||||
[] -> #{};
|
||||
[{_K, Conf} | _] ->
|
||||
emqx_mgmt_api_configs:gen_schema(Conf)
|
||||
end
|
||||
|| T <- ?TYPES],
|
||||
#{'oneOf' => Schema}.
|
||||
|
||||
node_schema() ->
|
||||
#{type => string, example => "emqx@127.0.0.1"}.
|
||||
|
||||
status_schema() ->
|
||||
#{type => string, enum => [connected, disconnected]}.
|
||||
|
||||
metrics_schema() ->
|
||||
#{ type => object
|
||||
, properties => #{
|
||||
matched => #{type => integer, example => "0"},
|
||||
success => #{type => integer, example => "0"},
|
||||
failed => #{type => integer, example => "0"},
|
||||
rate => #{type => number, format => float, example => "0.0"},
|
||||
rate_last5m => #{type => number, format => float, example => "0.0"},
|
||||
rate_max => #{type => number, format => float, example => "0.0"}
|
||||
}
|
||||
}.
|
||||
|
||||
per_node_schema(Key, Schema) ->
|
||||
#{
|
||||
type => array,
|
||||
items => #{
|
||||
type => object,
|
||||
properties => #{
|
||||
node => node_schema(),
|
||||
Key => Schema
|
||||
}
|
||||
}
|
||||
}.
|
||||
|
||||
resp_schema() ->
|
||||
AddMetadata = fun(Prop) ->
|
||||
Prop#{status => status_schema(),
|
||||
node_status => per_node_schema(status, status_schema()),
|
||||
metrics => metrics_schema(),
|
||||
node_metrics => per_node_schema(metrics, metrics_schema()),
|
||||
id => #{type => string, example => "http:my_http_bridge"},
|
||||
bridge_type => #{type => string, enum => ?TYPES},
|
||||
node => node_schema()
|
||||
}
|
||||
end,
|
||||
more_props_resp_schema(AddMetadata).
|
||||
|
||||
more_props_resp_schema(AddMetadata) ->
|
||||
#{'oneOf' := Schema} = req_schema(),
|
||||
Schema1 = [S#{properties => AddMetadata(Prop)}
|
||||
|| S = #{properties := Prop} <- Schema],
|
||||
#{'oneOf' => Schema1}.
|
||||
namespace() -> "bridge".
|
||||
|
||||
api_spec() ->
|
||||
{bridge_apis(), []}.
|
||||
emqx_dashboard_swagger:spec(?MODULE, #{check_schema => false}).
|
||||
|
||||
bridge_apis() ->
|
||||
[list_all_bridges_api(), crud_bridges_apis(), operation_apis()].
|
||||
paths() -> ["/bridges", "/bridges/:id", "/nodes/:node/bridges/:id/operation/:operation"].
|
||||
|
||||
list_all_bridges_api() ->
|
||||
ReqSchema = more_props_resp_schema(fun(Prop) ->
|
||||
Prop#{id => #{type => string, required => true}}
|
||||
end),
|
||||
RespSchema = resp_schema(),
|
||||
Metadata = #{
|
||||
error_schema(Code, Message) ->
|
||||
[ {code, mk(string(), #{example => Code})}
|
||||
, {message, mk(string(), #{example => Message})}
|
||||
].
|
||||
|
||||
get_response_body_schema() ->
|
||||
emqx_dashboard_swagger:schema_with_examples(emqx_bridge_schema:get_response(),
|
||||
bridge_info_examples(get)).
|
||||
|
||||
param_path_node() ->
|
||||
path_param(node, binary(), atom_to_binary(node(), utf8)).
|
||||
|
||||
param_path_operation() ->
|
||||
path_param(operation, enum([start, stop, restart]), <<"start">>).
|
||||
|
||||
param_path_id() ->
|
||||
path_param(id, binary(), <<"http:my_http_bridge">>).
|
||||
|
||||
path_param(Name, Type, Example) ->
|
||||
{Name, mk(Type,
|
||||
#{ in => path
|
||||
, required => true
|
||||
, example => Example
|
||||
})}.
|
||||
|
||||
bridge_info_array_example(Method) ->
|
||||
[Config || #{value := Config} <- maps:values(bridge_info_examples(Method))].
|
||||
|
||||
bridge_info_examples(Method) ->
|
||||
maps:merge(conn_bridge_examples(Method), #{
|
||||
<<"http_bridge">> => #{
|
||||
summary => <<"HTTP Bridge">>,
|
||||
value => info_example(http, awesome, Method)
|
||||
}
|
||||
}).
|
||||
|
||||
conn_bridge_examples(Method) ->
|
||||
lists:foldl(fun(Type, Acc) ->
|
||||
SType = atom_to_list(Type),
|
||||
KeyIngress = bin(SType ++ "_ingress"),
|
||||
KeyEgress = bin(SType ++ "_egress"),
|
||||
maps:merge(Acc, #{
|
||||
KeyIngress => #{
|
||||
summary => bin(string:uppercase(SType) ++ " Ingress Bridge"),
|
||||
value => info_example(Type, ingress, Method)
|
||||
},
|
||||
KeyEgress => #{
|
||||
summary => bin(string:uppercase(SType) ++ " Egress Bridge"),
|
||||
value => info_example(Type, egress, Method)
|
||||
}
|
||||
})
|
||||
end, #{}, ?CONN_TYPES).
|
||||
|
||||
info_example(Type, Direction, Method) ->
|
||||
maps:merge(info_example_basic(Type, Direction),
|
||||
method_example(Type, Direction, Method)).
|
||||
|
||||
method_example(Type, Direction, get) ->
|
||||
SType = atom_to_list(Type),
|
||||
SDir = atom_to_list(Direction),
|
||||
SName = "my_" ++ SDir ++ "_" ++ SType ++ "_bridge",
|
||||
#{
|
||||
id => bin(SType ++ ":" ++ SName),
|
||||
type => bin(SType),
|
||||
name => bin(SName)
|
||||
};
|
||||
method_example(Type, Direction, post) ->
|
||||
SType = atom_to_list(Type),
|
||||
SDir = atom_to_list(Direction),
|
||||
SName = "my_" ++ SDir ++ "_" ++ SType ++ "_bridge",
|
||||
#{
|
||||
type => bin(SType),
|
||||
name => bin(SName)
|
||||
};
|
||||
method_example(_Type, _Direction, put) ->
|
||||
#{}.
|
||||
|
||||
info_example_basic(http, _) ->
|
||||
#{
|
||||
url => <<"http://localhost:9901/messages/${topic}">>,
|
||||
request_timeout => <<"30s">>,
|
||||
connect_timeout => <<"30s">>,
|
||||
max_retries => 3,
|
||||
retry_interval => <<"10s">>,
|
||||
pool_type => <<"random">>,
|
||||
pool_size => 4,
|
||||
enable_pipelining => true,
|
||||
ssl => #{enable => false},
|
||||
from_local_topic => <<"emqx_http/#">>,
|
||||
method => post,
|
||||
body => <<"${payload}">>
|
||||
};
|
||||
info_example_basic(mqtt, ingress) ->
|
||||
#{
|
||||
connector => <<"mqtt:my_mqtt_connector">>,
|
||||
direction => ingress,
|
||||
from_remote_topic => <<"aws/#">>,
|
||||
subscribe_qos => 1,
|
||||
to_local_topic => <<"from_aws/${topic}">>,
|
||||
payload => <<"${payload}">>,
|
||||
qos => <<"${qos}">>,
|
||||
retain => <<"${retain}">>
|
||||
};
|
||||
info_example_basic(mqtt, egress) ->
|
||||
#{
|
||||
connector => <<"mqtt:my_mqtt_connector">>,
|
||||
direction => egress,
|
||||
from_local_topic => <<"emqx/#">>,
|
||||
to_remote_topic => <<"from_emqx/${topic}">>,
|
||||
payload => <<"${payload}">>,
|
||||
qos => 1,
|
||||
retain => false
|
||||
}.
|
||||
|
||||
schema("/bridges") ->
|
||||
#{
|
||||
operationId => '/bridges',
|
||||
get => #{
|
||||
tags => [<<"bridges">>],
|
||||
summary => <<"List Bridges">>,
|
||||
description => <<"List all created bridges">>,
|
||||
responses => #{
|
||||
<<"200">> => emqx_mgmt_util:array_schema(resp_schema(),
|
||||
<<"A list of the bridges">>)
|
||||
200 => emqx_dashboard_swagger:schema_with_example(
|
||||
array(emqx_bridge_schema:get_response()),
|
||||
bridge_info_array_example(get))
|
||||
}
|
||||
},
|
||||
post => #{
|
||||
tags => [<<"bridges">>],
|
||||
summary => <<"Create Bridge">>,
|
||||
description => <<"Create a new bridge">>,
|
||||
'requestBody' => emqx_mgmt_util:schema(ReqSchema),
|
||||
requestBody => emqx_dashboard_swagger:schema_with_examples(
|
||||
emqx_bridge_schema:post_request(),
|
||||
bridge_info_examples(post)),
|
||||
responses => #{
|
||||
<<"201">> => emqx_mgmt_util:schema(RespSchema, <<"Bridge created">>),
|
||||
<<"400">> => emqx_mgmt_util:error_schema(<<"Create bridge failed">>,
|
||||
['UPDATE_FAILED'])
|
||||
201 => get_response_body_schema(),
|
||||
400 => error_schema('BAD_ARG', "Create bridge failed")
|
||||
}
|
||||
}
|
||||
},
|
||||
{"/bridges/", Metadata, list_create_bridges_in_cluster}.
|
||||
};
|
||||
|
||||
crud_bridges_apis() ->
|
||||
ReqSchema = req_schema(),
|
||||
RespSchema = resp_schema(),
|
||||
Metadata = #{
|
||||
schema("/bridges/:id") ->
|
||||
#{
|
||||
operationId => '/bridges/:id',
|
||||
get => #{
|
||||
tags => [<<"bridges">>],
|
||||
summary => <<"Get Bridge">>,
|
||||
description => <<"Get a bridge by Id">>,
|
||||
parameters => [param_path_id()],
|
||||
responses => #{
|
||||
<<"200">> => emqx_mgmt_util:array_schema(RespSchema,
|
||||
<<"The details of the bridge">>),
|
||||
<<"404">> => emqx_mgmt_util:error_schema(<<"Bridge not found">>, ['NOT_FOUND'])
|
||||
200 => get_response_body_schema(),
|
||||
404 => error_schema('NOT_FOUND', "Bridge not found")
|
||||
}
|
||||
},
|
||||
put => #{
|
||||
tags => [<<"bridges">>],
|
||||
summary => <<"Update Bridge">>,
|
||||
description => <<"Update a bridge">>,
|
||||
parameters => [param_path_id()],
|
||||
'requestBody' => emqx_mgmt_util:schema(ReqSchema),
|
||||
requestBody => emqx_dashboard_swagger:schema_with_examples(
|
||||
emqx_bridge_schema:put_request(),
|
||||
bridge_info_examples(put)),
|
||||
responses => #{
|
||||
<<"200">> => emqx_mgmt_util:array_schema(RespSchema, <<"Bridge updated">>),
|
||||
<<"400">> => emqx_mgmt_util:error_schema(<<"Update bridge failed">>,
|
||||
['UPDATE_FAILED'])
|
||||
200 => get_response_body_schema(),
|
||||
400 => error_schema('BAD_ARG', "Update bridge failed")
|
||||
}
|
||||
},
|
||||
delete => #{
|
||||
tags => [<<"bridges">>],
|
||||
summary => <<"Delete Bridge">>,
|
||||
description => <<"Delete a bridge">>,
|
||||
parameters => [param_path_id()],
|
||||
responses => #{
|
||||
<<"204">> => emqx_mgmt_util:schema(<<"Bridge deleted">>),
|
||||
<<"404">> => emqx_mgmt_util:error_schema(<<"Bridge not found">>, ['NOT_FOUND'])
|
||||
204 => <<"Bridge deleted">>
|
||||
}
|
||||
}
|
||||
},
|
||||
{"/bridges/:id", Metadata, crud_bridges_in_cluster}.
|
||||
};
|
||||
|
||||
operation_apis() ->
|
||||
Metadata = #{
|
||||
schema("/nodes/:node/bridges/:id/operation/:operation") ->
|
||||
#{
|
||||
operationId => '/nodes/:node/bridges/:id/operation/:operation',
|
||||
post => #{
|
||||
tags => [<<"bridges">>],
|
||||
summary => <<"Start/Stop/Restart Bridge">>,
|
||||
description => <<"Start/Stop/Restart bridges on a specific node">>,
|
||||
parameters => [
|
||||
param_path_node(),
|
||||
param_path_id(),
|
||||
param_path_operation()],
|
||||
param_path_operation()
|
||||
],
|
||||
responses => #{
|
||||
<<"500">> => emqx_mgmt_util:error_schema(<<"Operation Failed">>,
|
||||
['INTERNAL_ERROR']),
|
||||
<<"200">> => emqx_mgmt_util:schema(<<"Operation success">>)}}},
|
||||
{"/nodes/:node/bridges/:id/operation/:operation", Metadata, manage_bridges}.
|
||||
|
||||
param_path_node() ->
|
||||
#{
|
||||
name => node,
|
||||
in => path,
|
||||
schema => #{type => string},
|
||||
required => true,
|
||||
example => node()
|
||||
500 => error_schema('INTERNAL_ERROR', "Operation Failed"),
|
||||
200 => <<"Operation success">>
|
||||
}
|
||||
}
|
||||
}.
|
||||
|
||||
param_path_id() ->
|
||||
#{
|
||||
name => id,
|
||||
in => path,
|
||||
schema => #{type => string},
|
||||
required => true
|
||||
}.
|
||||
|
||||
param_path_operation()->
|
||||
#{
|
||||
name => operation,
|
||||
in => path,
|
||||
required => true,
|
||||
schema => #{
|
||||
type => string,
|
||||
enum => [start, stop, restart]},
|
||||
example => restart
|
||||
}.
|
||||
|
||||
list_create_bridges_in_cluster(post, #{body := #{<<"id">> := Id} = Conf}) ->
|
||||
?TRY_PARSE_ID(Id,
|
||||
'/bridges'(post, #{body := #{<<"type">> := BridgeType} = Conf}) ->
|
||||
BridgeName = maps:get(<<"name">>, Conf, emqx_misc:gen_id()),
|
||||
case emqx_bridge:lookup(BridgeType, BridgeName) of
|
||||
{ok, _} -> {400, #{code => 'ALREADY_EXISTS', message => <<"bridge already exists">>}};
|
||||
{error, not_found} ->
|
||||
case ensure_bridge(BridgeType, BridgeName, maps:remove(<<"id">>, Conf)) of
|
||||
ok -> lookup_from_all_nodes(Id, BridgeType, BridgeName, 201);
|
||||
case ensure_bridge_created(BridgeType, BridgeName, Conf) of
|
||||
ok -> lookup_from_all_nodes(BridgeType, BridgeName, 201);
|
||||
{error, Error} -> {400, Error}
|
||||
end
|
||||
end);
|
||||
list_create_bridges_in_cluster(get, _Params) ->
|
||||
end;
|
||||
'/bridges'(get, _Params) ->
|
||||
{200, zip_bridges([list_local_bridges(Node) || Node <- mria_mnesia:running_nodes()])}.
|
||||
|
||||
list_local_bridges(Node) when Node =:= node() ->
|
||||
|
|
@ -238,22 +284,22 @@ list_local_bridges(Node) when Node =:= node() ->
|
|||
list_local_bridges(Node) ->
|
||||
rpc_call(Node, list_local_bridges, [Node]).
|
||||
|
||||
crud_bridges_in_cluster(get, #{bindings := #{id := Id}}) ->
|
||||
?TRY_PARSE_ID(Id, lookup_from_all_nodes(Id, BridgeType, BridgeName, 200));
|
||||
'/bridges/:id'(get, #{bindings := #{id := Id}}) ->
|
||||
?TRY_PARSE_ID(Id, lookup_from_all_nodes(BridgeType, BridgeName, 200));
|
||||
|
||||
crud_bridges_in_cluster(put, #{bindings := #{id := Id}, body := Conf}) ->
|
||||
'/bridges/:id'(put, #{bindings := #{id := Id}, body := Conf}) ->
|
||||
?TRY_PARSE_ID(Id,
|
||||
case emqx_bridge:lookup(BridgeType, BridgeName) of
|
||||
{ok, _} ->
|
||||
case ensure_bridge(BridgeType, BridgeName, Conf) of
|
||||
ok -> lookup_from_all_nodes(Id, BridgeType, BridgeName, 200);
|
||||
case ensure_bridge_created(BridgeType, BridgeName, Conf) of
|
||||
ok -> lookup_from_all_nodes(BridgeType, BridgeName, 200);
|
||||
{error, Error} -> {400, Error}
|
||||
end;
|
||||
{error, not_found} ->
|
||||
{404, #{code => 'NOT_FOUND', message => <<"bridge not found">>}}
|
||||
end);
|
||||
|
||||
crud_bridges_in_cluster(delete, #{bindings := #{id := Id}}) ->
|
||||
'/bridges/:id'(delete, #{bindings := #{id := Id}}) ->
|
||||
?TRY_PARSE_ID(Id,
|
||||
case emqx_conf:remove(emqx_bridge:config_key_path() ++ [BridgeType, BridgeName],
|
||||
#{override_to => cluster}) of
|
||||
|
|
@ -262,12 +308,12 @@ crud_bridges_in_cluster(delete, #{bindings := #{id := Id}}) ->
|
|||
{500, #{code => 102, message => emqx_resource_api:stringify(Reason)}}
|
||||
end).
|
||||
|
||||
lookup_from_all_nodes(Id, BridgeType, BridgeName, SuccCode) ->
|
||||
lookup_from_all_nodes(BridgeType, BridgeName, SuccCode) ->
|
||||
case rpc_multicall(lookup_from_local_node, [BridgeType, BridgeName]) of
|
||||
{ok, [{ok, _} | _] = Results} ->
|
||||
{SuccCode, format_bridge_info([R || {ok, R} <- Results])};
|
||||
{ok, [{error, not_found} | _]} ->
|
||||
{404, error_msg('NOT_FOUND', <<"not_found: ", Id/binary>>)};
|
||||
{404, error_msg('NOT_FOUND', <<"not_found">>)};
|
||||
{error, ErrL} ->
|
||||
{500, error_msg('UNKNOWN_ERROR', ErrL)}
|
||||
end.
|
||||
|
|
@ -278,7 +324,8 @@ lookup_from_local_node(BridgeType, BridgeName) ->
|
|||
Error -> Error
|
||||
end.
|
||||
|
||||
manage_bridges(post, #{bindings := #{node := Node, id := Id, operation := Op}}) ->
|
||||
'/nodes/:node/bridges/:id/operation/:operation'(post, #{bindings :=
|
||||
#{node := Node, id := Id, operation := Op}}) ->
|
||||
OperFun =
|
||||
fun (<<"start">>) -> start;
|
||||
(<<"stop">>) -> stop;
|
||||
|
|
@ -292,9 +339,10 @@ manage_bridges(post, #{bindings := #{node := Node, id := Id, operation := Op}})
|
|||
{500, #{code => 102, message => emqx_resource_api:stringify(Reason)}}
|
||||
end).
|
||||
|
||||
ensure_bridge(BridgeType, BridgeName, Conf) ->
|
||||
case emqx_conf:update(emqx_bridge:config_key_path() ++ [BridgeType, BridgeName], Conf,
|
||||
#{override_to => cluster}) of
|
||||
ensure_bridge_created(BridgeType, BridgeName, Conf) ->
|
||||
Conf1 = maps:without([<<"type">>, <<"name">>], Conf),
|
||||
case emqx_conf:update(emqx_bridge:config_key_path() ++ [BridgeType, BridgeName],
|
||||
Conf1, #{override_to => cluster}) of
|
||||
{ok, _} -> ok;
|
||||
{error, Reason} ->
|
||||
{error, error_msg('BAD_ARG', Reason)}
|
||||
|
|
@ -351,7 +399,7 @@ format_resp(#{id := Id, raw_config := RawConf,
|
|||
RawConf#{
|
||||
id => Id,
|
||||
node => node(),
|
||||
bridge_type => emqx_bridge:bridge_type(Mod),
|
||||
type => emqx_bridge:bridge_type(Mod),
|
||||
status => IsConnected(Status),
|
||||
metrics => Metrics
|
||||
}.
|
||||
|
|
@ -379,3 +427,10 @@ error_msg(Code, Msg) when is_binary(Msg) ->
|
|||
#{code => Code, message => Msg};
|
||||
error_msg(Code, Msg) ->
|
||||
#{code => Code, message => list_to_binary(io_lib:format("~p", [Msg]))}.
|
||||
|
||||
bin(S) when is_atom(S) ->
|
||||
atom_to_binary(S, utf8);
|
||||
bin(S) when is_list(S) ->
|
||||
list_to_binary(S);
|
||||
bin(S) when is_binary(S) ->
|
||||
S.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,95 @@
|
|||
-module(emqx_bridge_http_schema).
|
||||
|
||||
-include_lib("typerefl/include/types.hrl").
|
||||
|
||||
-import(hoconsc, [mk/2, enum/1]).
|
||||
|
||||
-export([roots/0, fields/1]).
|
||||
|
||||
%%======================================================================================
|
||||
%% Hocon Schema Definitions
|
||||
roots() -> [].
|
||||
|
||||
fields("bridge") ->
|
||||
basic_config() ++
|
||||
[ {url, mk(binary(),
|
||||
#{ nullable => false
|
||||
, desc =>"""
|
||||
The URL of the HTTP Bridge.<br>
|
||||
Template with variables is allowed in the path, but variables cannot be used in the scheme, host,
|
||||
or port part.<br>
|
||||
For example, <code> http://localhost:9901/${topic} </code> is allowed, but
|
||||
<code> http://${host}:9901/message </code> or <code> http://localhost:${port}/message </code>
|
||||
is not allowed.
|
||||
"""
|
||||
})}
|
||||
, {from_local_topic, mk(binary(),
|
||||
#{ desc =>"""
|
||||
The MQTT topic filter to be forwarded to the HTTP server. All MQTT PUBLISH messages which topic
|
||||
match the from_local_topic will be forwarded.<br>
|
||||
NOTE: if this bridge is used as the output of a rule (emqx rule engine), and also from_local_topic is configured, then both the data got from the rule and the MQTT messages that matches
|
||||
from_local_topic will be forwarded.
|
||||
"""
|
||||
})}
|
||||
, {method, mk(method(),
|
||||
#{ default => post
|
||||
, desc =>"""
|
||||
The method of the HTTP request. All the available methods are: post, put, get, delete.<br>
|
||||
Template with variables is allowed.<br>
|
||||
"""
|
||||
})}
|
||||
, {headers, mk(map(),
|
||||
#{ default => #{
|
||||
<<"accept">> => <<"application/json">>,
|
||||
<<"cache-control">> => <<"no-cache">>,
|
||||
<<"connection">> => <<"keep-alive">>,
|
||||
<<"content-type">> => <<"application/json">>,
|
||||
<<"keep-alive">> => <<"timeout=5">>}
|
||||
, desc =>"""
|
||||
The headers of the HTTP request.<br>
|
||||
Template with variables is allowed.
|
||||
"""
|
||||
})
|
||||
}
|
||||
, {body, mk(binary(),
|
||||
#{ default => <<"${payload}">>
|
||||
, desc =>"""
|
||||
The body of the HTTP request.<br>
|
||||
Template with variables is allowed.
|
||||
"""
|
||||
})}
|
||||
, {request_timeout, mk(emqx_schema:duration_ms(),
|
||||
#{ default => <<"30s">>
|
||||
, desc =>"""
|
||||
How long will the HTTP request timeout.
|
||||
"""
|
||||
})}
|
||||
];
|
||||
|
||||
fields("post") ->
|
||||
[ type_field()
|
||||
, name_field()
|
||||
] ++ fields("bridge");
|
||||
|
||||
fields("put") ->
|
||||
fields("bridge");
|
||||
|
||||
fields("get") ->
|
||||
[ id_field()
|
||||
] ++ fields("post").
|
||||
|
||||
basic_config() ->
|
||||
proplists:delete(base_url, emqx_connector_http:fields(config)).
|
||||
|
||||
%%======================================================================================
|
||||
id_field() ->
|
||||
{id, mk(binary(), #{desc => "The Bridge Id", example => "http:my_http_bridge"})}.
|
||||
|
||||
type_field() ->
|
||||
{type, mk(http, #{desc => "The Bridge Type"})}.
|
||||
|
||||
name_field() ->
|
||||
{name, mk(binary(), #{desc => "The Bridge Name"})}.
|
||||
|
||||
method() ->
|
||||
enum([post, put, get, delete]).
|
||||
|
|
@ -0,0 +1,62 @@
|
|||
-module(emqx_bridge_mqtt_schema).
|
||||
|
||||
-include_lib("typerefl/include/types.hrl").
|
||||
|
||||
-import(hoconsc, [mk/2]).
|
||||
|
||||
-export([roots/0, fields/1]).
|
||||
|
||||
%%======================================================================================
|
||||
%% Hocon Schema Definitions
|
||||
roots() -> [].
|
||||
|
||||
fields("ingress") ->
|
||||
[ direction(ingress, emqx_connector_mqtt_schema:ingress_desc())
|
||||
, emqx_bridge_schema:connector_name()
|
||||
] ++ proplists:delete(hookpoint, emqx_connector_mqtt_schema:fields("ingress"));
|
||||
|
||||
fields("egress") ->
|
||||
[ direction(egress, emqx_connector_mqtt_schema:egress_desc())
|
||||
, emqx_bridge_schema:connector_name()
|
||||
] ++ emqx_connector_mqtt_schema:fields("egress");
|
||||
|
||||
fields("post_ingress") ->
|
||||
[ type_field()
|
||||
, name_field()
|
||||
] ++ fields("ingress");
|
||||
fields("post_egress") ->
|
||||
[ type_field()
|
||||
, name_field()
|
||||
] ++ fields("egress");
|
||||
|
||||
fields("put_ingress") ->
|
||||
fields("ingress");
|
||||
fields("put_egress") ->
|
||||
fields("egress");
|
||||
|
||||
fields("get_ingress") ->
|
||||
[ id_field()
|
||||
] ++ fields("post_ingress");
|
||||
fields("get_egress") ->
|
||||
[ id_field()
|
||||
] ++ fields("post_egress").
|
||||
|
||||
%%======================================================================================
|
||||
direction(Dir, Desc) ->
|
||||
{direction, mk(Dir,
|
||||
#{ nullable => false
|
||||
, desc => "The direction of the bridge. Can be one of 'ingress' or 'egress'.<br>"
|
||||
++ Desc
|
||||
})}.
|
||||
|
||||
id_field() ->
|
||||
{id, mk(binary(), #{desc => "The Bridge Id", example => "mqtt:my_mqtt_bridge"})}.
|
||||
|
||||
type_field() ->
|
||||
{type, mk(mqtt, #{desc => "The Bridge Type"})}.
|
||||
|
||||
name_field() ->
|
||||
{name, mk(binary(),
|
||||
#{ desc => "The Bridge Name"
|
||||
, example => "some_bridge_name"
|
||||
})}.
|
||||
|
|
@ -2,105 +2,27 @@
|
|||
|
||||
-include_lib("typerefl/include/types.hrl").
|
||||
|
||||
-import(hoconsc, [mk/2, ref/2]).
|
||||
|
||||
-export([roots/0, fields/1]).
|
||||
|
||||
-export([ get_response/0
|
||||
, put_request/0
|
||||
, post_request/0
|
||||
]).
|
||||
|
||||
-export([ connector_name/0
|
||||
]).
|
||||
|
||||
%%======================================================================================
|
||||
%% Hocon Schema Definitions
|
||||
|
||||
roots() -> [bridges].
|
||||
-define(CONN_TYPES, [mqtt]).
|
||||
|
||||
fields(bridges) ->
|
||||
[ {mqtt,
|
||||
sc(hoconsc:map(name, hoconsc:union([ ref("ingress_mqtt_bridge")
|
||||
, ref("egress_mqtt_bridge")
|
||||
])),
|
||||
#{ desc => "MQTT bridges"
|
||||
})}
|
||||
, {http,
|
||||
sc(hoconsc:map(name, ref("http_bridge")),
|
||||
#{ desc => "HTTP bridges"
|
||||
})}
|
||||
];
|
||||
|
||||
fields("ingress_mqtt_bridge") ->
|
||||
[ direction(ingress, emqx_connector_mqtt_schema:ingress_desc())
|
||||
, connector_name()
|
||||
] ++ proplists:delete(hookpoint, emqx_connector_mqtt_schema:fields("ingress"));
|
||||
|
||||
fields("egress_mqtt_bridge") ->
|
||||
[ direction(egress, emqx_connector_mqtt_schema:egress_desc())
|
||||
, connector_name()
|
||||
] ++ emqx_connector_mqtt_schema:fields("egress");
|
||||
|
||||
fields("http_bridge") ->
|
||||
basic_config_http() ++
|
||||
[ {url,
|
||||
sc(binary(),
|
||||
#{ nullable => false
|
||||
, desc =>"""
|
||||
The URL of the HTTP Bridge.<br>
|
||||
Template with variables is allowed in the path, but variables cannot be used in the scheme, host,
|
||||
or port part.<br>
|
||||
For example, <code> http://localhost:9901/${topic} </code> is allowed, but
|
||||
<code> http://${host}:9901/message </code> or <code> http://localhost:${port}/message </code>
|
||||
is not allowed.
|
||||
"""
|
||||
})}
|
||||
, {from_local_topic,
|
||||
sc(binary(),
|
||||
#{ desc =>"""
|
||||
The MQTT topic filter to be forwarded to the HTTP server. All MQTT PUBLISH messages which topic
|
||||
match the from_local_topic will be forwarded.<br>
|
||||
NOTE: if this bridge is used as the output of a rule (emqx rule engine), and also from_local_topic is configured, then both the data got from the rule and the MQTT messages that matches
|
||||
from_local_topic will be forwarded.
|
||||
"""
|
||||
})}
|
||||
, {method,
|
||||
sc(method(),
|
||||
#{ default => post
|
||||
, desc =>"""
|
||||
The method of the HTTP request. All the available methods are: post, put, get, delete.<br>
|
||||
Template with variables is allowed.<br>
|
||||
"""
|
||||
})}
|
||||
, {headers,
|
||||
sc(map(),
|
||||
#{ default => #{
|
||||
<<"accept">> => <<"application/json">>,
|
||||
<<"cache-control">> => <<"no-cache">>,
|
||||
<<"connection">> => <<"keep-alive">>,
|
||||
<<"content-type">> => <<"application/json">>,
|
||||
<<"keep-alive">> => <<"timeout=5">>}
|
||||
, desc =>"""
|
||||
The headers of the HTTP request.<br>
|
||||
Template with variables is allowed.
|
||||
"""
|
||||
})
|
||||
}
|
||||
, {body,
|
||||
sc(binary(),
|
||||
#{ default => <<"${payload}">>
|
||||
, desc =>"""
|
||||
The body of the HTTP request.<br>
|
||||
Template with variables is allowed.
|
||||
"""
|
||||
})}
|
||||
, {request_timeout,
|
||||
sc(emqx_schema:duration_ms(),
|
||||
#{ default => <<"30s">>
|
||||
, desc =>"""
|
||||
How long will the HTTP request timeout.
|
||||
"""
|
||||
})}
|
||||
].
|
||||
|
||||
direction(Dir, Desc) ->
|
||||
{direction,
|
||||
sc(Dir,
|
||||
#{ nullable => false
|
||||
, desc => "The direction of the bridge. Can be one of 'ingress' or 'egress'.<br>" ++
|
||||
Desc
|
||||
})}.
|
||||
%%======================================================================================
|
||||
%% For HTTP APIs
|
||||
get_response() ->
|
||||
http_schema("get").
|
||||
|
||||
connector_name() ->
|
||||
{connector,
|
||||
|
|
@ -108,17 +30,35 @@ connector_name() ->
|
|||
#{ nullable => false
|
||||
, desc =>"""
|
||||
The connector name to be used for this bridge.
|
||||
Connectors are configured as 'connectors.type.name',
|
||||
Connectors are configured as 'connectors.{type}.{name}',
|
||||
for example 'connectors.http.mybridge'.
|
||||
"""
|
||||
})}.
|
||||
|
||||
basic_config_http() ->
|
||||
proplists:delete(base_url, emqx_connector_http:fields(config)).
|
||||
put_request() ->
|
||||
http_schema("put").
|
||||
|
||||
method() ->
|
||||
hoconsc:enum([post, put, get, delete]).
|
||||
post_request() ->
|
||||
http_schema("post").
|
||||
|
||||
sc(Type, Meta) -> hoconsc:mk(Type, Meta).
|
||||
http_schema(Method) ->
|
||||
Schemas = lists:flatmap(fun(Type) ->
|
||||
[ref(schema_mod(Type), Method ++ "_ingress"),
|
||||
ref(schema_mod(Type), Method ++ "_egress")]
|
||||
end, ?CONN_TYPES),
|
||||
hoconsc:union([ref(emqx_bridge_http_schema, Method)
|
||||
| Schemas]).
|
||||
|
||||
ref(Field) -> hoconsc:ref(?MODULE, Field).
|
||||
%%======================================================================================
|
||||
%% For config files
|
||||
roots() -> [bridges].
|
||||
|
||||
fields(bridges) ->
|
||||
[{http, mk(hoconsc:map(name, ref(emqx_bridge_http_schema, "bridge")), #{})}]
|
||||
++ [{T, mk(hoconsc:map(name, hoconsc:union([
|
||||
ref(schema_mod(T), "ingress"),
|
||||
ref(schema_mod(T), "egress")
|
||||
])), #{})} || T <- ?CONN_TYPES].
|
||||
|
||||
schema_mod(Type) ->
|
||||
list_to_atom(lists:concat(["emqx_bridge_", Type, "_schema"])).
|
||||
|
|
|
|||
|
|
@ -62,7 +62,7 @@ post_request_body_schema() ->
|
|||
connector_info(post_req), connector_info_examples()).
|
||||
|
||||
get_response_body_schema() ->
|
||||
emqx_dashboard_swagger:schema_with_example(
|
||||
emqx_dashboard_swagger:schema_with_examples(
|
||||
connector_info(), connector_info_examples()).
|
||||
|
||||
connector_info() ->
|
||||
|
|
|
|||
|
|
@ -100,7 +100,7 @@ on_start(InstId, Conf) ->
|
|||
BasicConf = basic_config(Conf),
|
||||
BridgeConf = BasicConf#{
|
||||
name => InstanceId,
|
||||
clientid => clientid(maps:get(clientid, Conf, InstanceId)),
|
||||
clientid => clientid(maps:get(clientid, Conf, InstId)),
|
||||
subscriptions => make_sub_confs(maps:get(ingress, Conf, undefined)),
|
||||
forwards => make_forward_confs(maps:get(egress, Conf, undefined))
|
||||
},
|
||||
|
|
@ -190,4 +190,4 @@ basic_config(#{
|
|||
}.
|
||||
|
||||
clientid(Id) ->
|
||||
unicode:characters_to_binary([Id, ":", atom_to_list(node())], utf8).
|
||||
iolist_to_binary([Id, ":", atom_to_list(node())]).
|
||||
|
|
|
|||
Loading…
Reference in New Issue