diff --git a/apps/endpoint/src/endpoint.app.src b/apps/endpoint/src/endpoint.app.src index 794990a..e2cf43a 100644 --- a/apps/endpoint/src/endpoint.app.src +++ b/apps/endpoint/src/endpoint.app.src @@ -14,6 +14,7 @@ brod, hackney, gproc, + crypto, kernel, stdlib ]}, diff --git a/apps/endpoint/src/endpoint.erl b/apps/endpoint/src/endpoint.erl index 8c1121c..e7177d5 100644 --- a/apps/endpoint/src/endpoint.erl +++ b/apps/endpoint/src/endpoint.erl @@ -68,7 +68,7 @@ get_protocol(#endpoint{config = #kafka_endpoint{}}) -> -spec is_support(Protocol :: atom()) -> boolean(). is_support(Protocol) when is_atom(Protocol) -> - {ok, Props} = application:get_env(iot, endpoints), + {ok, Props} = application:get_env(endpoint, endpoints), SupportProtocols = proplists:get_value(support_protocols, Props, []), lists:member(Protocol, SupportProtocols). diff --git a/apps/endpoint/src/endpoint_buffer.erl b/apps/endpoint/src/endpoint_buffer.erl index 52e5f5b..05f5da7 100644 --- a/apps/endpoint/src/endpoint_buffer.erl +++ b/apps/endpoint/src/endpoint_buffer.erl @@ -42,7 +42,7 @@ -spec new(Endpoint :: #endpoint{}, WindowSize :: integer()) -> Buffer :: #buffer{}. new(Endpoint = #endpoint{id = Id}, WindowSize) when is_integer(WindowSize), WindowSize > 0 -> %% 读取配置文件 - {ok, Endpoints} = application:get_env(iot, endpoints), + {ok, Endpoints} = application:get_env(endpoint, endpoints), RootDir = proplists:get_value(root_dir, Endpoints), OutboxDir = filename:join(RootDir, integer_to_list(Id)), diff --git a/apps/endpoint/src/endpoint_http.erl b/apps/endpoint/src/endpoint_http.erl index 875cb2b..38a36fe 100644 --- a/apps/endpoint/src/endpoint_http.erl +++ b/apps/endpoint/src/endpoint_http.erl @@ -45,7 +45,7 @@ start_link(LocalName, Endpoint = #endpoint{config = #http_endpoint{}}) -> {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} | {stop, Reason :: term()} | ignore). init([Endpoint = #endpoint{matcher = Matcher, config = #http_endpoint{pool_size = PoolSize}}]) -> - ok = iot_log:set_metadata(), + ok = endpoint_util:set_metadata(), endpoint_subscription:subscribe(Matcher, self()), Buffer = endpoint_buffer:new(Endpoint, PoolSize), {ok, #state{endpoint = Endpoint, buffer = Buffer}}. @@ -151,7 +151,7 @@ read_response_body(ClientRef) -> -spec patch_headers(Metric :: binary(), any()) -> list(). patch_headers(Metric, Token) when is_binary(Token), Token /= <<>> -> - Sign = iot_util:sha256(erlang:iolist_to_binary([Token, Metric, Token])), + Sign = endpoint_util:sha256(erlang:iolist_to_binary([Token, Metric, Token])), [{<<"X-Signature">>, Sign}]; patch_headers(_, _) -> []. diff --git a/apps/endpoint/src/endpoint_kafka.erl b/apps/endpoint/src/endpoint_kafka.erl index 4c59c58..2286c30 100644 --- a/apps/endpoint/src/endpoint_kafka.erl +++ b/apps/endpoint/src/endpoint_kafka.erl @@ -49,7 +49,7 @@ callback_mode() -> -spec init(term()) -> gen_statem:init_result(kafka_state(), #state{}). init([Endpoint = #endpoint{id = Id, matcher = Matcher}]) -> - ok = iot_log:set_metadata(), + ok = endpoint_util:set_metadata(), erlang:process_flag(trap_exit, true), ok = endpoint_subscription:subscribe(Matcher, self()), Buffer = endpoint_buffer:new(Endpoint, 10), diff --git a/apps/endpoint/src/endpoint_log.erl b/apps/endpoint/src/endpoint_log.erl index c0bc86b..3967c3e 100644 --- a/apps/endpoint/src/endpoint_log.erl +++ b/apps/endpoint/src/endpoint_log.erl @@ -51,7 +51,7 @@ unmatched_publish(RouteKey, Content) when is_binary(RouteKey), is_binary(Content -spec init(term()) -> {ok, #state{}}. init([]) -> - ok = iot_log:set_metadata(), + ok = endpoint_util:set_metadata(), case open_log() of ok -> {ok, #state{enabled = true}}; @@ -126,7 +126,7 @@ open_log() -> -spec log_config() -> proplists:proplist(). log_config() -> - case application:get_env(iot, endpoint_log) of + case application:get_env(endpoint, endpoint_log) of {ok, Config} when is_list(Config) -> Config; _ -> @@ -140,7 +140,7 @@ log_path(Config) -> -spec endpoint_root_dir() -> file:filename_all(). endpoint_root_dir() -> - case application:get_env(iot, endpoints) of + case application:get_env(endpoint, endpoints) of {ok, Endpoints} -> proplists:get_value(root_dir, Endpoints, ?DEFAULT_ENDPOINT_ROOT_DIR); undefined -> diff --git a/apps/endpoint/src/endpoint_mqtt.erl b/apps/endpoint/src/endpoint_mqtt.erl index cafd84c..ce75188 100644 --- a/apps/endpoint/src/endpoint_mqtt.erl +++ b/apps/endpoint/src/endpoint_mqtt.erl @@ -50,7 +50,7 @@ callback_mode() -> -spec init(term()) -> gen_statem:init_result(mqtt_state(), #state{}). init([Endpoint = #endpoint{matcher = Matcher}]) -> - ok = iot_log:set_metadata(), + ok = endpoint_util:set_metadata(), erlang:process_flag(trap_exit, true), ok = endpoint_subscription:subscribe(Matcher, self()), Buffer = endpoint_buffer:new(Endpoint, 10), diff --git a/apps/endpoint/src/endpoint_subscription.erl b/apps/endpoint/src/endpoint_subscription.erl index 6197962..742feb3 100644 --- a/apps/endpoint/src/endpoint_subscription.erl +++ b/apps/endpoint/src/endpoint_subscription.erl @@ -141,7 +141,7 @@ start_link() -> {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} | {stop, Reason :: term()} | ignore). init([]) -> - ok = iot_log:set_metadata(), + ok = endpoint_util:set_metadata(), ExactTid = ets:new(?EXACT_TAB, [named_table, protected, bag, {read_concurrency, true}]), EdgeTid = ets:new(?TRIE_EDGE_TAB, [named_table, protected, set, {read_concurrency, true}]), TrieSubTid = ets:new(?TRIE_SUB_TAB, [named_table, protected, bag, {read_concurrency, true}]), diff --git a/apps/endpoint/src/endpoint_util.erl b/apps/endpoint/src/endpoint_util.erl new file mode 100644 index 0000000..d573a2a --- /dev/null +++ b/apps/endpoint/src/endpoint_util.erl @@ -0,0 +1,18 @@ +%%%------------------------------------------------------------------- +%%% @doc Endpoint-local utility functions. +%%%------------------------------------------------------------------- +-module(endpoint_util). + +-export([set_metadata/0, sha256/1]). + +-spec set_metadata() -> ok. +set_metadata() -> + logger:set_process_metadata(#{domain => [iot]}). + +-spec sha256(string() | binary()) -> binary(). +sha256(Str) when is_list(Str) -> + sha256(unicode:characters_to_binary(Str)); +sha256(Bin) when is_binary(Bin) -> + HashBin = crypto:hash(sha256, Bin), + HexStr = lists:flatten([io_lib:format("~2.16.0B", [B]) || B <- binary:bin_to_list(HashBin)]), + list_to_binary(string:lowercase(HexStr)). diff --git a/config/sys-dev.config b/config/sys-dev.config index 67ffd9f..5a4caec 100644 --- a/config/sys-dev.config +++ b/config/sys-dev.config @@ -52,6 +52,22 @@ ]} ]}, + {endpoint, [ + %% 支持的协议 + {endpoints, [ + {root_dir, "/usr/local/code/database/"}, + {support_protocols, [ + http + ]} + ]}, + + {endpoint_log, [ + {path, "${endpoint_root}/endpoint_log/unmatched_publish.log"}, + {max_bytes, 10485760}, + {max_files, 10} + ]} + ]}, + %% 系统日志配置,使用 OTP logger {kernel, [ %% 设置 Logger 的 primary log level diff --git a/config/sys-prod.config b/config/sys-prod.config index b98468c..edd7775 100644 --- a/config/sys-prod.config +++ b/config/sys-prod.config @@ -47,6 +47,21 @@ ]} ]}, + {endpoint, [ + {endpoints, [ + {root_dir, "/usr/local/code/database/"}, + {support_protocols, [ + http + ]} + ]}, + + {endpoint_log, [ + {path, "${endpoint_root}/endpoint_log/unmatched_publish.log"}, + {max_bytes, 10485760}, + {max_files, 10} + ]} + ]}, + %% 系统日志配置,使用 OTP logger {kernel, [ %% 设置 Logger 的 primary log level