anlicheng/iot_cloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修改日志系统

Browse Source
This commit is contained in:
anlicheng 2026-04-18 15:44:58 +08:00
parent 92f0b2b100
commit f2847053b7
25 changed files with 211 additions and 218 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
config/sys-dev.config
View File

@ -77,39 +77,30 @@
]},
%% 系统日志配置，使用 OTP logger
{kernel, [
%% 设置 Logger 的 primary log level
{logger_level, debug},
{logger, [
{handler, default, logger_std_h,
#{
level => debug,
formatter => {logger_formatter, #{template => [time, " [", level, "] ", msg, "\n"]}}
}
},
%% 系统日志配置，系统日志为lager, 支持日志按日期自动分割
{lager, [
{colored, true},
%% Whether to write a crash log, and where. Undefined means no crash logger.
{crash_log, "trade_hub.crash.log"},
%% Maximum size in bytes of events in the crash log - defaults to 65536
{crash_log_msg_size, 65536},
%% Maximum size of the crash log in bytes, before its rotated, set
%% to 0 to disable rotation - default is 0
{crash_log_size, 10485760},
%% What time to rotate the crash log - default is no time
%% rotation. See the README for a description of this format.
{crash_log_date, "$D0"},
%% Number of rotated crash logs to keep, 0 means keep only the
%% current one - default is 0
{crash_log_count, 5},
%% Whether to redirect error_logger messages into lager - defaults to true
{error_logger_redirect, true},
{handler, disk, logger_disk_log_h,
#{
level => debug,
config => #{
file => "log/debug.log",
max_no_files => 10,
max_no_bytes => 524288000
},
formatter => {logger_formatter, #{template => [time, " [", level, "] ", msg, "\n"]}}
}
}
%% How big the gen_event mailbox can get before it is switched into sync mode
{async_threshold, 20},
%% Switch back to async mode, when gen_event mailbox size decrease from `async_threshold'
%% to async_threshold - async_threshold_window
{async_threshold_window, 5},
{handlers, [
%% debug | info | warning | error, 日志级别
{lager_console_backend, debug},
{lager_file_backend, [{file, "debug.log"}, {level, debug}, {size, 314572800}]},
{lager_file_backend, [{file, "notice.log"}, {level, notice}, {size, 314572800}]},
{lager_file_backend, [{file, "error.log"}, {level, error}, {size, 314572800}]},
{lager_file_backend, [{file, "info.log"}, {level, info}, {size, 314572800}]}
]}
]}

52
config/sys-prod.config
View File

@ -47,38 +47,30 @@
]},
%% 系统日志配置，系统日志为lager, 支持日志按日期自动分割
{lager, [
{colored, true},
%% Whether to write a crash log, and where. Undefined means no crash logger.
{crash_log, "trade_hub.crash.log"},
%% Maximum size in bytes of events in the crash log - defaults to 65536
{crash_log_msg_size, 65536},
%% Maximum size of the crash log in bytes, before its rotated, set
%% to 0 to disable rotation - default is 0
{crash_log_size, 10485760},
%% What time to rotate the crash log - default is no time
%% rotation. See the README for a description of this format.
{crash_log_date, "$D0"},
%% Number of rotated crash logs to keep, 0 means keep only the
%% current one - default is 0
{crash_log_count, 5},
%% Whether to redirect error_logger messages into lager - defaults to true
{error_logger_redirect, true},
%% 系统日志配置，使用 OTP logger
{kernel, [
%% 设置 Logger 的 primary log level
{logger_level, debug},
{logger, [
{handler, default, logger_std_h,
#{
level => debug,
formatter => {logger_formatter, #{template => [time, " [", level, "] ", msg, "\n"]}}
}
},
%% How big the gen_event mailbox can get before it is switched into sync mode
{async_threshold, 20},
%% Switch back to async mode, when gen_event mailbox size decrease from `async_threshold'
%% to async_threshold - async_threshold_window
{async_threshold_window, 5},
{handler, disk, logger_disk_log_h,
#{
level => debug,
config => #{
file => "log/debug.log",
max_no_files => 10,
max_no_bytes => 524288000
},
formatter => {logger_formatter, #{template => [time, " [", level, "] ", msg, "\n"]}}
}
}
{handlers, [
%% debug | info | warning | error, 日志级别
{lager_console_backend, debug},
{lager_file_backend, [{file, "debug.log"}, {level, debug}, {size, 314572800}]},
{lager_file_backend, [{file, "notice.log"}, {level, notice}, {size, 314572800}]},
{lager_file_backend, [{file, "error.log"}, {level, error}, {size, 314572800}]},
{lager_file_backend, [{file, "info.log"}, {level, info}, {size, 314572800}]}
]}
]}

28
rebar.config
View File

@ -1,4 +1,26 @@
{erl_opts, [debug_info]}.
{plugins, [
{rebar3_gpb_plugin, ".*", {git, "https://github.com/lrascao/rebar3_gpb_plugin.git", {tag, "2.23.8"}}}
]}.
% ======================
% 核心：强制指定 proto 路径 + 输出到你的 apps/sdlan
% ======================
{gpb_opts, [
{i, "proto"}, % proto 文件路径
{src_dirs, ["proto"]}, % 源码目录（必须）
recursive, % 递归查找 proto 文件
{module_name_suffix, "_pb"}, % 生成模块后缀
{o_erl, "src"}, % .erl 输出目录
{o_hrl, "include"}, % .hrl 输出目录
include_as_lib, % gpb.hrl 通过 -include_lib("gpb/include/gpb.hrl")
{strings_as_binaries, true}, % proto string → Erlang binary
type_specs, % 生成 type specs
report, % 编译报告
verbose % 打印详细信息
]}.
{deps, [
{poolboy, ".*", {git, "https://github.com/devinus/poolboy.git", {tag, "1.5.1"}}},
{hackney, ".*", {git, "https://github.com/benoitc/hackney.git", {tag, "1.25.0"}}},
@ -11,8 +33,7 @@
{eredis, ".*", {git, "https://github.com/wooga/eredis.git", {tag, "v1.2.0"}}},
{emqtt, ".*", {git, "https://gitea.s5s8.com/anlicheng/emqtt.git", {tag, "v1.2"}}},
{gproc, ".*", {git, "https://github.com/uwiger/gproc.git", {tag, "0.9.1"}}},
{parse_trans, ".*", {git, "https://github.com/uwiger/parse_trans", {tag, "3.0.0"}}},
{lager, ".*", {git,"https://github.com/erlang-lager/lager.git", {tag, "3.9.2"}}}
{parse_trans, ".*", {git, "https://github.com/uwiger/parse_trans", {tag, "3.0.0"}}}
]}.
{relx, [{release, {iot, "0.1.0"},
@ -46,7 +67,4 @@
]
}]}]}.
{erl_opts, [{parse_transform,lager_transform}]}.
{rebar_packages_cdn, "https://hexpm.upyun.com"}.

7
rebar.lock
View File

@ -22,7 +22,6 @@
{ref,"9ad91f149310a7d002cb966f62b7e2c3330abb04"}},
0},
{<<"fs">>,{pkg,<<"fs">>,<<"6.1.1">>},1},
{<<"goldrush">>,{pkg,<<"goldrush">>,<<"0.1.9">>},1},
{<<"gproc">>,
{git,"https://github.com/uwiger/gproc.git",
{ref,"4ca45e0a97722a418a31eb1753f4e3b953f7fb1d"}},
@ -37,10 +36,6 @@
{ref,"9ea1b35b6e60ba21dfd4adbd18e7916a831fd7d4"}},
0},
{<<"kafka_protocol">>,{pkg,<<"kafka_protocol">>,<<"4.2.7">>},1},
{<<"lager">>,
{git,"https://github.com/erlang-lager/lager.git",
{ref,"459a3b2cdd9eadd29e5a7ce5c43932f5ccd6eb88"}},
0},
{<<"metrics">>,{pkg,<<"metrics">>,<<"1.0.1">>},1},
{<<"mimerl">>,{pkg,<<"mimerl">>,<<"1.4.0">>},1},
{<<"mysql">>,
@ -70,7 +65,6 @@
{<<"certifi">>, <<"0E6E882FCDAAA0A5A9F2B3DB55B1394DBA07E8D6D9BCAD08318FB604C6839712">>},
{<<"crc32cer">>, <<"AD0E42BED8603F2C72DE2A00F1B5063FFE12D5988615CAD984096900431D1C1A">>},
{<<"fs">>, <<"9D147B944D60CFA48A349F12D06C8EE71128F610C90870BDF9A6773206452ED0">>},
{<<"goldrush">>, <<"F06E5D5F1277DA5C413E84D5A2924174182FB108DABB39D5EC548B27424CD106">>},
{<<"idna">>, <<"8A63070E9F7D0C62EB9D9FCB360A7DE382448200FBBD1B106CC96D3D8099DF8D">>},
{<<"kafka_protocol">>, <<"6F53B15CD6F6A12C1D0010DB074B4A15985C71BC7F594BC2D67D9837B3B378A1">>},
{<<"metrics">>, <<"25F094DEA2CDA98213CECC3AEFF09E940299D950904393B2A29D191C346A8486">>},
@ -81,7 +75,6 @@
{<<"certifi">>, <<"B147ED22CE71D72EAFDAD94F055165C1C182F61A2FF49DF28BCC71D1D5B94A60">>},
{<<"crc32cer">>, <<"08FDCD5CE51ACD839A12E98742F0F0EDA19A2A679FC9FBFAF6AAB958310FB70E">>},
{<<"fs">>, <<"EF94E95FFE79916860649FED80AC62B04C322B0BB70F5128144C026B4D171F8B">>},
{<<"goldrush">>, <<"99CB4128CFFCB3227581E5D4D803D5413FA643F4EB96523F77D9E6937D994CEB">>},
{<<"idna">>, <<"92376EB7894412ED19AC475E4A86F7B413C1B9FBB5BD16DCCD57934157944CEA">>},
{<<"kafka_protocol">>, <<"1D5E9597AD3C0776C86DC5E08D3BAAEA7DB805A52E5FD35E3F071AAD7789FC4C">>},
{<<"metrics">>, <<"69B09ADDDC4F74A40716AE54D140F93BEB0FB8978D8636EADED0C31B6F099F16">>},

18
src/adapters/control_api/iot_api_client.erl
View File

@ -110,14 +110,14 @@ ai_event(Id) when is_integer(Id) ->
case hackney:request(post, Url, Headers, Body, [{pool, false}]) of
{ok, 200, _, ClientRef} ->
{ok, RespBody} = hackney:body(ClientRef),
lager:debug("[iot_api_client] send body: ~p, get error is: ~p", [Body, RespBody]),
logger:debug("[iot_api_client] send body: ~p, get error is: ~p", [Body, RespBody]),
hackney:close(ClientRef);
{ok, HttpCode, _, ClientRef} ->
{ok, RespBody} = hackney:body(ClientRef),
hackney:close(ClientRef),
lager:warning("[iot_api_client] send body: ~p, get error is: ~p", [Body, {HttpCode, RespBody}]);
logger:warning("[iot_api_client] send body: ~p, get error is: ~p", [Body, {HttpCode, RespBody}]);
{error, Reason} ->
lager:warning("[iot_api_client] send body: ~p, get error is: ~p", [Body, Reason])
logger:warning("[iot_api_client] send body: ~p, get error is: ~p", [Body, Reason])
end.
%%%-------------------------------------------------------------------
@ -136,7 +136,7 @@ do_post(Path, Params) when is_list(Path), is_map(Params) ->
case hackney:request(post, Url, Headers, Body, [{pool, false}]) of
{ok, 200, _, ClientRef} ->
{ok, RespBody} = hackney:body(ClientRef),
lager:debug("[iot_api_client] request url: ~p, send body: ~p, get response is: ~p", [Url, Body, RespBody]),
logger:debug("[iot_api_client] request url: ~p, send body: ~p, get response is: ~p", [Url, Body, RespBody]),
hackney:close(ClientRef),
case catch jiffy:decode(RespBody, [return_maps]) of
#{<<"result">> := Result} ->
@ -151,10 +151,10 @@ do_post(Path, Params) when is_list(Path), is_map(Params) ->
{ok, HttpCode, _, ClientRef} ->
{ok, RespBody} = hackney:body(ClientRef),
hackney:close(ClientRef),
lager:warning("[iot_api_client] request url: ~p, send body: ~p, get error is: ~p", [Url, Body, {HttpCode, RespBody}]),
logger:warning("[iot_api_client] request url: ~p, send body: ~p, get error is: ~p", [Url, Body, {HttpCode, RespBody}]),
{error, {HttpCode, RespBody}};
{error, Reason} ->
lager:warning("[iot_api_client] request url: ~p, send body: ~p, get error is: ~p", [Url, Body, Reason]),
logger:warning("[iot_api_client] request url: ~p, send body: ~p, get error is: ~p", [Url, Body, Reason]),
{error, Reason}
end.
@ -177,7 +177,7 @@ do_get(Path, Params) when is_list(Path), is_list(Params) ->
{ok, 200, _, ClientRef} ->
{ok, RespBody} = hackney:body(ClientRef),
hackney:close(ClientRef),
lager:debug("[iot_api_client] url: ~p, get response is: ~p", [Url, RespBody]),
logger:debug("[iot_api_client] url: ~p, get response is: ~p", [Url, RespBody]),
case catch jiffy:decode(RespBody, [return_maps]) of
#{<<"result">> := Result} ->
{ok, Result};
@ -191,9 +191,9 @@ do_get(Path, Params) when is_list(Path), is_list(Params) ->
{ok, HttpCode, _, ClientRef} ->
{ok, RespBody} = hackney:body(ClientRef),
hackney:close(ClientRef),
lager:warning("[iot_api_client] request url: ~p, get error is: ~p", [Url, {HttpCode, RespBody}]),
logger:warning("[iot_api_client] request url: ~p, get error is: ~p", [Url, {HttpCode, RespBody}]),
{error, {HttpCode, RespBody}};
{error, Reason} ->
lager:warning("[iot_api_client] request url: ~p, get error is: ~p", [Url, Reason]),
logger:warning("[iot_api_client] request url: ~p, get error is: ~p", [Url, Reason]),
{error, Reason}
end.

6
src/adapters/http/http_client.erl
View File

@ -20,15 +20,15 @@ post(Url, Headers, Body) when is_list(Url), is_list(Headers), is_binary(Body) ->
case hackney:request(post, Url, Headers, Body, [{pool, false}]) of
{ok, 200, _, ClientRef} ->
{ok, RespBody} = hackney:body(ClientRef),
lager:debug("[http_client] send body: ~p, get error is: ~p", [Body, RespBody]),
logger:debug("[http_client] send body: ~p, get error is: ~p", [Body, RespBody]),
hackney:close(ClientRef),
{ok, RespBody};
{ok, HttpCode, _, ClientRef} ->
{ok, RespBody} = hackney:body(ClientRef),
hackney:close(ClientRef),
lager:warning("[http_client] send body: ~p, get error is: ~p", [Body, {HttpCode, RespBody}]),
logger:warning("[http_client] send body: ~p, get error is: ~p", [Body, {HttpCode, RespBody}]),
{error, {HttpCode, RespBody}};
{error, Reason} ->
lager:warning("[http_client] send body: ~p, get error is: ~p", [Body, Reason]),
logger:warning("[http_client] send body: ~p, get error is: ~p", [Body, Reason]),
{error, Reason}
end.

10
src/adapters/http/iot_http_client.erl
View File

@ -17,24 +17,24 @@ post(Url, Body) when is_list(Url), is_binary(Body) ->
{ok, 200, _, ClientRef} ->
case hackney:body(ClientRef) of
{ok, RespBody} ->
lager:debug("[iot_http_client] url: ~p, response is: ~p", [Url, RespBody]),
logger:debug("[iot_http_client] url: ~p, response is: ~p", [Url, RespBody]),
ok;
{error, Reason} ->
lager:warning("[iot_http_client] url: ~p, get error: ~p", [Url, Reason]),
logger:warning("[iot_http_client] url: ~p, get error: ~p", [Url, Reason]),
{error, Reason}
end;
{ok, HttpCode, _, ClientRef} ->
case hackney:body(ClientRef) of
{ok, RespBody} ->
lager:debug("[iot_http_client] url: ~p, http_code: ~p, response is: ~p", [Url, HttpCode, RespBody]),
logger:debug("[iot_http_client] url: ~p, http_code: ~p, response is: ~p", [Url, HttpCode, RespBody]),
ok;
{error, Reason} ->
lager:warning("[iot_http_client] url: ~p, http_code: ~p, get error: ~p", [Url, HttpCode, Reason]),
logger:warning("[iot_http_client] url: ~p, http_code: ~p, get error: ~p", [Url, HttpCode, Reason]),
{error, Reason}
end;
{error, Reason} ->
lager:warning("[iot_http_client] url: ~p, get error: ~p", [Url, Reason]),
logger:warning("[iot_http_client] url: ~p, get error: ~p", [Url, Reason]),
{error, Reason}
end.

8
src/adapters/influxdb/influx_client.erl
View File

@ -151,18 +151,18 @@ handle_cast({write, Bucket, Org, Precision, Points}, State = #state{host = Host,
query => Query
}),
lager:debug("[influx_client] url is: ~p, headers: ~p, body: ~ts", [Url, Headers, Body]),
logger:debug("[influx_client] url is: ~p, headers: ~p, body: ~ts", [Url, Headers, Body]),
case hackney:request(post, Url, Headers, GZipBody, [{pool, false}]) of
{ok, StatusCode, _RespHeaders, ClientRef} ->
case hackney:body(ClientRef) of
{ok, RespBody} ->
lager:debug("[influx_client] status_code: ~p, response body is: ~p", [StatusCode, RespBody]);
logger:debug("[influx_client] status_code: ~p, response body is: ~p", [StatusCode, RespBody]);
{error, Error} ->
lager:warning("[influx_client] status_code: ~p, error is: ~p", [Error])
logger:warning("[influx_client] status_code: ~p, error is: ~p", [Error])
end,
hackney:close(ClientRef);
{error, Reason} ->
lager:warning("[influx_client] request result is: ~p", [Reason])
logger:warning("[influx_client] request result is: ~p", [Reason])
end,
{noreply, State}.

26
src/adapters/mysql/mysql_provider.erl
View File

@ -16,49 +16,49 @@
%% 从数据库中查找一行记录
-spec get_row(ConnPid :: pid(), Sql::binary()) -> {ok, Record::map()} | undefined.
get_row(ConnPid, Sql) when is_pid(ConnPid), is_binary(Sql) ->
lager:debug("[mysql_client] get_row sql is: ~p", [Sql]),
logger:debug("[mysql_client] get_row sql is: ~p", [Sql]),
case mysql:query(ConnPid, Sql) of
{ok, Names, [Row | _]} ->
{ok, maps:from_list(lists:zip(Names, Row))};
{ok, _, []} ->
undefined;
Error ->
lager:warning("[mysql_client] get error: ~p", [Error]),
logger:warning("[mysql_client] get error: ~p", [Error]),
undefined
end.
-spec get_row(ConnPid :: pid(), Sql::binary(), Params::list()) -> {ok, Record::map()} | undefined.
get_row(ConnPid, Sql, Params) when is_pid(ConnPid), is_binary(Sql), is_list(Params) ->
lager:debug("[mysql_client] get_row sql is: ~p, params: ~p", [Sql, Params]),
logger:debug("[mysql_client] get_row sql is: ~p, params: ~p", [Sql, Params]),
case mysql:query(ConnPid, Sql, Params) of
{ok, Names, [Row | _]} ->
{ok, maps:from_list(lists:zip(Names, Row))};
{ok, _, []} ->
undefined;
Error ->
lager:warning("[mysql_client] get error: ~p", [Error]),
logger:warning("[mysql_client] get error: ~p", [Error]),
undefined
end.
-spec get_all(ConnPid :: pid(), Sql::binary()) -> {ok, Rows::list()} | {error, Reason :: any()}.
get_all(ConnPid, Sql) when is_pid(ConnPid), is_binary(Sql) ->
lager:debug("[mysql_client] get_all sql is: ~p", [Sql]),
logger:debug("[mysql_client] get_all sql is: ~p", [Sql]),
case mysql:query(ConnPid, Sql) of
{ok, Names, Rows} ->
{ok, lists:map(fun(Row) -> maps:from_list(lists:zip(Names, Row)) end, Rows)};
{error, Reason} ->
lager:warning("[mysql_client] get error: ~p", [Reason]),
logger:warning("[mysql_client] get error: ~p", [Reason]),
{error, Reason}
end.
-spec get_all(ConnPid :: pid(), Sql::binary(), Params::list()) -> {ok, Rows::list()} | {error, Reason::any()}.
get_all(ConnPid, Sql, Params) when is_pid(ConnPid), is_binary(Sql), is_list(Params) ->
lager:debug("[mysql_client] get_all sql is: ~p, params: ~p", [Sql, Params]),
logger:debug("[mysql_client] get_all sql is: ~p, params: ~p", [Sql, Params]),
case mysql:query(ConnPid, Sql, Params) of
{ok, Names, Rows} ->
{ok, lists:map(fun(Row) -> maps:from_list(lists:zip(Names, Row)) end, Rows)};
{error, Reason} ->
lager:warning("[mysql_client] get error: ~p", [Reason]),
logger:warning("[mysql_client] get error: ~p", [Reason]),
{error, Reason}
end.
@ -74,7 +74,7 @@ insert(ConnPid, Table, Fields, FetchInsertId) when is_pid(ConnPid), is_binary(Ta
ValuesPlaceholder = iolist_to_binary(lists:join(<<", ">>, Placeholders)),
Sql = <<"INSERT INTO ", Table/binary, "(", FieldSql/binary, ") VALUES(", ValuesPlaceholder/binary, ")">>,
lager:debug("[mysql_client] insert sql is: ~p, params: ~p", [Sql, Values]),
logger:debug("[mysql_client] insert sql is: ~p, params: ~p", [Sql, Values]),
case mysql:query(ConnPid, Sql, Values) of
ok ->
case FetchInsertId of
@ -90,7 +90,7 @@ insert(ConnPid, Table, Fields, FetchInsertId) when is_pid(ConnPid), is_binary(Ta
-spec update_by(ConnPid :: pid(), UpdateSql :: binary()) -> {ok, AffectedRows :: integer()} | {error, Reason :: any()}.
update_by(ConnPid, UpdateSql) when is_pid(ConnPid), is_binary(UpdateSql) ->
lager:debug("[mysql_client] updateBySql sql: ~p", [UpdateSql]),
logger:debug("[mysql_client] updateBySql sql: ~p", [UpdateSql]),
case mysql:query(ConnPid, UpdateSql) of
ok ->
AffectedRows = mysql:affected_rows(ConnPid),
@ -101,7 +101,7 @@ update_by(ConnPid, UpdateSql) when is_pid(ConnPid), is_binary(UpdateSql) ->
-spec update_by(ConnPid :: pid(), UpdateSql :: binary(), Params :: list()) -> {ok, AffectedRows :: integer()} | {error, Reason :: any()}.
update_by(ConnPid, UpdateSql, Params) when is_pid(ConnPid), is_binary(UpdateSql) ->
lager:debug("[mysql_client] updateBySql sql: ~p, params: ~p", [UpdateSql, Params]),
logger:debug("[mysql_client] updateBySql sql: ~p, params: ~p", [UpdateSql, Params]),
case mysql:query(ConnPid, UpdateSql, Params) of
ok ->
AffectedRows = mysql:affected_rows(ConnPid),
@ -126,13 +126,13 @@ update(ConnPid, Table, Fields, WhereFields) when is_pid(ConnPid), is_binary(Tabl
Params = SetVals ++ WhereVals,
Sql = <<"UPDATE ", Table/binary, " SET ", SetSql/binary, " WHERE ", WhereSql/binary>>,
lager:debug("[mysql_client] update sql is: ~p, params: ~p", [Sql, Params]),
logger:debug("[mysql_client] update sql is: ~p, params: ~p", [Sql, Params]),
case mysql:query(ConnPid, Sql, Params) of
ok ->
AffectedRows = mysql:affected_rows(ConnPid),
{ok, AffectedRows};
Error ->
lager:error("[mysql_client] update sql: ~p, params: ~p, get a error: ~p", [Sql, Params, Error]),
logger:error("[mysql_client] update sql: ~p, params: ~p, get a error: ~p", [Sql, Params, Error]),
Error
end.

4
src/devtools/endpoint_kafka_test.erl
View File

@ -49,7 +49,7 @@ test_consumer() ->
ok = brod:start_client(KafkaBootstrapEndpoints, client1, ClientConfig),
SubscriberCallbackFun = fun(_Partition, Msg, ShellPid = CallbackState) ->
lager:debug("call here msg: ~p", [Msg]),
logger:debug("call here msg: ~p", [Msg]),
ShellPid ! Msg, {ok, ack, CallbackState}
end,
Res = brod_topic_subscriber:start_link(client1, Topic, all,
@ -57,4 +57,4 @@ test_consumer() ->
_CommittedOffsets=[], message, SubscriberCallbackFun,
_CallbackState=self()),
lager:debug("start subscriber res: ~p", [Res]).
logger:debug("start subscriber res: ~p", [Res]).

24
src/devtools/endpoint_mqtt_subscriber.erl
View File

@ -67,23 +67,23 @@ init([]) ->
{retry_interval, 5}
],
lager:debug("[opts] is: ~p", [Opts]),
logger:debug("[opts] is: ~p", [Opts]),
case emqtt:start_link(Opts) of
{ok, ConnPid} ->
%% 监听和host相关的全部事件
lager:debug("[iot_mqtt_subscriber] start conntecting, pid: ~p", [ConnPid]),
logger:debug("[iot_mqtt_subscriber] start conntecting, pid: ~p", [ConnPid]),
{ok, _} = emqtt:connect(ConnPid),
lager:debug("[iot_mqtt_subscriber] connect success, pid: ~p", [ConnPid]),
logger:debug("[iot_mqtt_subscriber] connect success, pid: ~p", [ConnPid]),
SubscribeResult = emqtt:subscribe(ConnPid, ?Topics),
lager:debug("[iot_mqtt_subscriber] subscribe topics: ~p, result is: ~p", [?Topics, SubscribeResult]),
logger:debug("[iot_mqtt_subscriber] subscribe topics: ~p, result is: ~p", [?Topics, SubscribeResult]),
{ok, #state{conn_pid = ConnPid}};
ignore ->
lager:debug("[iot_mqtt_subscriber] connect emqx get ignore"),
logger:debug("[iot_mqtt_subscriber] connect emqx get ignore"),
{stop, ignore};
{error, Reason} ->
lager:debug("[iot_mqtt_subscriber] connect emqx get error: ~p", [Reason]),
logger:debug("[iot_mqtt_subscriber] connect emqx get error: ~p", [Reason]),
{stop, Reason}
end.
@ -116,19 +116,19 @@ handle_cast(_Request, State = #state{}) ->
{noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), NewState :: #state{}}).
handle_info({disconnect, ReasonCode, Properties}, State = #state{}) ->
lager:debug("[iot_mqtt_subscriber] Recv a DISONNECT packet - ReasonCode: ~p, Properties: ~p", [ReasonCode, Properties]),
logger:debug("[iot_mqtt_subscriber] Recv a DISONNECT packet - ReasonCode: ~p, Properties: ~p", [ReasonCode, Properties]),
{stop, disconnected, State};
%% 必须要做到消息的快速分发，数据的json反序列需要在host进程进行
handle_info({publish, #{packet_id := _PacketId, payload := Payload, qos := Qos, topic := Topic}}, State = #state{conn_pid = _ConnPid}) ->
lager:debug("[iot_mqtt_subscriber] Recv a topic: ~p, publish packet: ~p, qos: ~p", [Topic, Payload, Qos]),
logger:debug("[iot_mqtt_subscriber] Recv a topic: ~p, publish packet: ~p, qos: ~p", [Topic, Payload, Qos]),
%% 将消息分发到对应的host进程去处理
{noreply, State};
handle_info({puback, Packet = #{packet_id := _PacketId}}, State = #state{}) ->
lager:debug("[iot_mqtt_subscriber] receive puback packet: ~p", [Packet]),
logger:debug("[iot_mqtt_subscriber] receive puback packet: ~p", [Packet]),
{noreply, State};
handle_info(Info, State = #state{}) ->
lager:debug("[iot_mqtt_subscriber] get info: ~p", [Info]),
logger:debug("[iot_mqtt_subscriber] get info: ~p", [Info]),
{noreply, State}.
%% @private
@ -144,10 +144,10 @@ terminate(Reason, _State = #state{conn_pid = ConnPid}) when is_pid(ConnPid) ->
{ok, _Props, _ReasonCode} = emqtt:unsubscribe(ConnPid, #{}, TopicNames),
ok = emqtt:disconnect(ConnPid),
lager:debug("[iot_mqtt_subscriber] terminate with reason: ~p", [Reason]),
logger:debug("[iot_mqtt_subscriber] terminate with reason: ~p", [Reason]),
ok;
terminate(Reason, _State) ->
lager:debug("[iot_mqtt_subscriber] terminate with reason: ~p", [Reason]),
logger:debug("[iot_mqtt_subscriber] terminate with reason: ~p", [Reason]),
ok.
%% @private

18
src/devtools/iot_mock.erl
View File

@ -40,7 +40,7 @@ insert_services(Num) ->
<<"url">> => <<"https://www.baidu.com">>,
<<"detail">> => <<"这是一个关于测试的微服务"/utf8>>
}, false),
lager:debug("insert service result is: ~p", [Res])
logger:debug("insert service result is: ~p", [Res])
end, lists:seq(1, Num)).
rsa_encode(Data) when is_binary(Data) ->
@ -49,14 +49,14 @@ rsa_encode(Data) when is_binary(Data) ->
%% 私钥保存解析后的
{ok, PubBin} = file:read_file(PublicPemFile),
lager:debug("pub bin is: ~p", [PubBin]),
logger:debug("pub bin is: ~p", [PubBin]),
[Pub] = public_key:pem_decode(PubBin),
lager:debug("pub pem bin is: ~p", [Pub]),
logger:debug("pub pem bin is: ~p", [Pub]),
PubKey = public_key:pem_entry_decode(Pub),
lager:debug("the public key is: ~p", [PubKey]),
logger:debug("the public key is: ~p", [PubKey]),
EncData = public_key:encrypt_public(Data, PubKey),
lager:debug("enc data is: ~p", [EncData]),
logger:debug("enc data is: ~p", [EncData]),
rsa_decode(EncData),
@ -68,13 +68,13 @@ rsa_decode(EncData) when is_binary(EncData) ->
%% 私钥保存解析后的
{ok, PubBin} = file:read_file(PublicPemFile),
lager:debug("pub bin is: ~p", [PubBin]),
logger:debug("pub bin is: ~p", [PubBin]),
[Pub] = public_key:pem_decode(PubBin),
lager:debug("pub pem bin is: ~p", [Pub]),
logger:debug("pub pem bin is: ~p", [Pub]),
PubKey = public_key:pem_entry_decode(Pub),
lager:debug("the public key is: ~p", [PubKey]),
logger:debug("the public key is: ~p", [PubKey]),
PlainData = public_key:decrypt_private(EncData, PubKey),
lager:debug("plain data is: ~p", [PlainData]),
logger:debug("plain data is: ~p", [PlainData]),
ok.

6
src/endpoint/endpoint_http.erl
View File

@ -102,18 +102,18 @@ handle_info({next_data, Id, {Metric, Sign}}, State = #state{buffer = Buffer, end
{ok, 200, _, ClientRef} ->
{ok, RespBody} = hackney:body(ClientRef),
hackney:close(ClientRef),
lager:debug("[endpoint_http] url: ~p, response is: ~p", [Url, RespBody]),
logger:debug("[endpoint_http] url: ~p, response is: ~p", [Url, RespBody]),
NBuffer = endpoint_buffer:ack(Id, Buffer),
{noreply, State#state{buffer = NBuffer}};
{ok, HttpCode, _, ClientRef} ->
{ok, RespBody} = hackney:body(ClientRef),
hackney:close(ClientRef),
lager:debug("[endpoint_http] url: ~p, http_code: ~p, response is: ~p", [Url, HttpCode, RespBody]),
logger:debug("[endpoint_http] url: ~p, http_code: ~p, response is: ~p", [Url, HttpCode, RespBody]),
NBuffer = endpoint_buffer:ack(Id, Buffer),
{noreply, State#state{buffer = NBuffer}};
{error, Reason} ->
lager:warning("[endpoint_http] url: ~p, get error: ~p", [Url, Reason]),
logger:warning("[endpoint_http] url: ~p, get error: ~p", [Url, Reason]),
{noreply, State}
end.

14
src/endpoint/endpoint_kafka.erl
View File

@ -94,7 +94,7 @@ handle_cast({forward, Metric}, State = #state{buffer = Buffer}) ->
{stop, Reason :: term(), NewState :: #state{}}).
handle_info({timeout, _, connect}, State = #state{buffer = Buffer, status = ?DISCONNECTED, client_id = ClientId,
endpoint = #endpoint{title = Title, config = #kafka_endpoint{sasl_config = SaslConfig, bootstrap_servers = BootstrapServers, topic = Topic}}}) ->
lager:debug("[endpoint_kafka] endpoint: ~p, create postman", [Title]),
logger:debug("[endpoint_kafka] endpoint: ~p, create postman", [Title]),
BaseConfig = [
{reconnect_cool_down_seconds, 5},
@ -115,13 +115,13 @@ handle_info({timeout, _, connect}, State = #state{buffer = Buffer, status = ?DIS
NBuffer = endpoint_buffer:trigger_next(Buffer),
{noreply, State#state{buffer = NBuffer, client_pid = ClientPid, status = ?CONNECTED}};
{error, Reason} ->
lager:debug("[endpoint_kafka] start_producer: ~p, get error: ~p", [ClientId, Reason]),
logger:debug("[endpoint_kafka] start_producer: ~p, get error: ~p", [ClientId, Reason]),
brod:stop_client(ClientId),
retry_connect(),
{noreply, State#state{status = ?DISCONNECTED, client_pid = undefined}}
end;
Error ->
lager:debug("[endpoint_kafka] start_client: ~p, get error: ~p", [ClientId, Error]),
logger:debug("[endpoint_kafka] start_client: ~p, get error: ~p", [ClientId, Error]),
retry_connect(),
{noreply, State#state{status = ?DISCONNECTED, client_pid = undefined}}
end;
@ -135,7 +135,7 @@ handle_info({next_data, Id, Metric}, State = #state{status = ?CONNECTED, client_
ReceiverPid = self(),
AckCb = fun(Partition, BaseOffset) ->
lager:debug("[endpoint_kafka] ack partion: ~p, offset: ~p", [Partition, BaseOffset]),
logger:debug("[endpoint_kafka] ack partion: ~p, offset: ~p", [Partition, BaseOffset]),
ReceiverPid ! {ack, Id}
end,
_ = brod:produce_cb(ClientPid, Topic, random, <<>>, Metric, AckCb),
@ -148,12 +148,12 @@ handle_info({ack, Id}, State = #state{buffer = Buffer}) ->
%% postman进程挂掉时，重新建立新的
handle_info({'EXIT', ClientPid, Reason}, State = #state{client_pid = ClientPid, endpoint = #endpoint{title = Title}}) ->
lager:warning("[endpoint_kafka] endpoint: ~p, conn pid exit with reason: ~p", [Title, Reason]),
logger:warning("[endpoint_kafka] endpoint: ~p, conn pid exit with reason: ~p", [Title, Reason]),
retry_connect(),
{noreply, State#state{client_pid = undefined, status = ?DISCONNECTED}};
handle_info(Info, State = #state{status = Status}) ->
lager:warning("[endpoint_kafka] unknown message: ~p, status: ~p", [Info, Status]),
logger:warning("[endpoint_kafka] unknown message: ~p, status: ~p", [Info, Status]),
{noreply, State}.
%% @private
@ -164,7 +164,7 @@ handle_info(Info, State = #state{status = Status}) ->
-spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
State :: #state{}) -> term()).
terminate(Reason, #state{endpoint = #endpoint{title = Title}, buffer = Buffer}) ->
lager:debug("[endpoint_kafka] endpoint: ~p, terminate with reason: ~p", [Title, Reason]),
logger:debug("[endpoint_kafka] endpoint: ~p, terminate with reason: ~p", [Title, Reason]),
endpoint_buffer:cleanup(Buffer),
ok.

24
src/endpoint/endpoint_mqtt.erl
View File

@ -95,7 +95,7 @@ handle_cast({forward, Metric}, State = #state{buffer = Buffer}) ->
{stop, Reason :: term(), NewState :: #state{}}).
handle_info({timeout, _, create_postman}, State = #state{buffer = Buffer, status = ?DISCONNECTED,
endpoint = #endpoint{title = Title, config = #mqtt_endpoint{host = Host, port = Port, username = Username, password = Password, client_id = ClientId}}}) ->
lager:debug("[endpoint_mqtt] endpoint: ~ts, create postman", [Title]),
logger:debug("[endpoint_mqtt] endpoint: ~ts, create postman", [Title]),
Opts = [
{owner, self()},
{clientid, ClientId},
@ -113,19 +113,19 @@ handle_info({timeout, _, create_postman}, State = #state{buffer = Buffer, status
try
{ok, ConnPid} = emqtt:start_link(Opts),
lager:debug("[endpoint_mqtt] start connect, options: ~p", [Opts]),
logger:debug("[endpoint_mqtt] start connect, options: ~p", [Opts]),
case emqtt:connect(ConnPid, 5000) of
{ok, _} ->
lager:debug("[endpoint_mqtt] connect success, pid: ~p", [ConnPid]),
logger:debug("[endpoint_mqtt] connect success, pid: ~p", [ConnPid]),
NBuffer = endpoint_buffer:trigger_n(Buffer),
{noreply, State#state{conn_pid = ConnPid, buffer = NBuffer, status = ?CONNECTED}};
{error, Reason} ->
lager:warning("[endpoint_mqtt] connect get error: ~p", [Reason]),
logger:warning("[endpoint_mqtt] connect get error: ~p", [Reason]),
erlang:start_timer(?RETRY_INTERVAL, self(), create_postman),
{noreply, State}
end
catch _:Error->
lager:warning("[endpoint_mqtt] connect get error: ~p", [Error]),
logger:warning("[endpoint_mqtt] connect get error: ~p", [Error]),
erlang:start_timer(?RETRY_INTERVAL, self(), create_postman),
{noreply, State}
end;
@ -137,7 +137,7 @@ handle_info({next_data, _Id, _Tuple}, State = #state{status = ?DISCONNECTED}) ->
handle_info({next_data, Id, Metric}, State = #state{status = ?CONNECTED, conn_pid = ConnPid, buffer = Buffer, inflight = InFlight,
endpoint = #endpoint{config = #mqtt_endpoint{topic = Topic, qos = Qos}}}) ->
lager:debug("[endpoint_mqtt] will publish topic: ~p, metric: ~p, qos: ~p", [Topic, Metric, Qos]),
logger:debug("[endpoint_mqtt] will publish topic: ~p, metric: ~p, qos: ~p", [Topic, Metric, Qos]),
case emqtt:publish(ConnPid, Topic, #{}, Metric, [{qos, Qos}, {retain, true}]) of
ok ->
NBuffer = endpoint_buffer:ack(Id, Buffer),
@ -145,17 +145,17 @@ handle_info({next_data, Id, Metric}, State = #state{status = ?CONNECTED, conn_pi
{ok, PacketId} ->
{noreply, State#state{inflight = maps:put(PacketId, Id, InFlight)}};
{error, Reason} ->
lager:warning("[endpoint_mqtt] send message to topic: ~p, get error: ~p", [Topic, Reason]),
logger:warning("[endpoint_mqtt] send message to topic: ~p, get error: ~p", [Topic, Reason]),
{stop, Reason, State}
end;
handle_info({disconnected, ReasonCode, Properties}, State = #state{status = ?CONNECTED}) ->
lager:debug("[endpoint_mqtt] Recv a DISONNECT packet - ReasonCode: ~p, Properties: ~p", [ReasonCode, Properties]),
logger:debug("[endpoint_mqtt] Recv a DISONNECT packet - ReasonCode: ~p, Properties: ~p", [ReasonCode, Properties]),
erlang:start_timer(?RETRY_INTERVAL, self(), create_postman),
{noreply, State#state{conn_pid = undefined, status = ?DISCONNECTED}};
handle_info({publish, Message = #{packet_id := _PacketId, payload := Payload}}, State = #state{status = ?CONNECTED}) ->
lager:debug("[endpoint_mqtt] Recv a publish packet: ~p, payload: ~p", [Message, Payload]),
logger:debug("[endpoint_mqtt] Recv a publish packet: ~p, payload: ~p", [Message, Payload]),
{noreply, State};
%% 收到确认的消息
@ -170,12 +170,12 @@ handle_info({puback, #{packet_id := PacketId}}, State = #state{status = ?CONNECT
%% postman进程挂掉时，重新建立新的
handle_info({'EXIT', ConnPid, Reason}, State = #state{endpoint = #endpoint{title = Title}, conn_pid = ConnPid}) ->
lager:warning("[endpoint_mqtt] endpoint: ~p, conn pid exit with reason: ~p", [Title, Reason]),
logger:warning("[endpoint_mqtt] endpoint: ~p, conn pid exit with reason: ~p", [Title, Reason]),
erlang:start_timer(?RETRY_INTERVAL, self(), create_postman),
{noreply, State#state{conn_pid = undefined, status = ?DISCONNECTED}};
handle_info(Info, State = #state{status = Status}) ->
lager:warning("[endpoint_mqtt] unknown message: ~p, status: ~p", [Info, Status]),
logger:warning("[endpoint_mqtt] unknown message: ~p, status: ~p", [Info, Status]),
{noreply, State}.
%% @private
@ -186,7 +186,7 @@ handle_info(Info, State = #state{status = Status}) ->
-spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
State :: #state{}) -> term()).
terminate(Reason, #state{endpoint = #endpoint{title = Title}, buffer = Buffer}) ->
lager:debug("[endpoint_mqtt] endpoint: ~p, terminate with reason: ~p", [Title, Reason]),
logger:debug("[endpoint_mqtt] endpoint: ~p, terminate with reason: ~p", [Title, Reason]),
endpoint_buffer:cleanup(Buffer),
ok.

6
src/endpoint/endpoint_subscription.erl
View File

@ -109,7 +109,7 @@ handle_cast({publish, RouteKey, Metric}, State = #state{subscribers = Subscriber
lists:foreach(fun(#subscriber{subscriber_pid = SubscriberPid}) ->
endpoint:forward(SubscriberPid, Metric)
end, MatchedSubscribers),
lager:debug("[efka_subscription] route_key: ~p, metric: ~p, match subscribers: ~p", [RouteKey, Metric, MatchedSubscribers]),
logger:debug("[efka_subscription] route_key: ~p, metric: ~p, match subscribers: ~p", [RouteKey, Metric, MatchedSubscribers]),
{noreply, State}.
%% @private
@ -119,12 +119,12 @@ handle_cast({publish, RouteKey, Metric}, State = #state{subscribers = Subscriber
{noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), NewState :: #state{}}).
handle_info({'DOWN', _Ref, process, SubscriberPid, Reason}, State = #state{subscribers = Subscribers}) ->
lager:debug("[efka_subscription] subscriber: ~p, down with reason: ~p", [SubscriberPid, Reason]),
logger:debug("[efka_subscription] subscriber: ~p, down with reason: ~p", [SubscriberPid, Reason]),
NSubscribers = lists:filter(fun(#subscriber{subscriber_pid = Pid0}) -> SubscriberPid /= Pid0 end, Subscribers),
{noreply, State#state{subscribers = NSubscribers}};
handle_info(Info, State = #state{}) ->
lager:debug("[efka_subscription] get unknown info: ~p", [Info]),
logger:debug("[efka_subscription] get unknown info: ~p", [Info]),
{noreply, State}.
%% @private

44
src/host/iot_host.erl
View File

@ -189,7 +189,7 @@ init([UUID]) ->
end,
{ok, StateName, #state{host_id = HostId, uuid = UUID, has_session = false}};
undefined ->
lager:warning("[iot_host] host uuid: ~p, loaded from mysql failed", [UUID]),
logger:warning("[iot_host] host uuid: ~p, loaded from mysql failed", [UUID]),
ignore
end.
@ -226,7 +226,7 @@ handle_event({call, From}, {jsonrpc_call, ReceiverPid, RpcCall}, _, State = #sta
Ref = tcp_channel:jsonrpc_call(ChannelPid, ReceiverPid, RpcCall),
{keep_state, State, [{reply, From, {ok, Ref}}]};
false ->
lager:debug("[iot_host] uuid: ~p, invalid state: ~p", [UUID, state_map(State)]),
logger:debug("[iot_host] uuid: ~p, invalid state: ~p", [UUID, state_map(State)]),
{keep_state, State, [{reply, From, {error, <<"主机离线，发送请求失败"/utf8>>}}]}
end;
@ -234,13 +234,13 @@ handle_event({call, From}, {jsonrpc_call, ReceiverPid, RpcCall}, _, State = #sta
handle_event({call, From}, {pub, Topic, Qos, Content}, ?STATE_ACTIVATED, State = #state{uuid = UUID, channel_pid = ChannelPid, has_session = HasSession}) ->
case HasSession andalso is_pid(ChannelPid) of
true ->
lager:debug("[iot_host] host: ~p, publish to topic: ~p, content: ~p", [UUID, Topic, Content]),
logger:debug("[iot_host] host: ~p, publish to topic: ~p, content: ~p", [UUID, Topic, Content]),
%% 通过websocket发送请求
tcp_channel:pub(ChannelPid, Topic, Qos, Content),
{keep_state, State, [{reply, From, ok}]};
false ->
lager:debug("[iot_host] uuid: ~p, publish to topic: ~p, content: ~p, invalid state: ~p", [UUID, Topic, Content, state_map(State)]),
logger:debug("[iot_host] uuid: ~p, publish to topic: ~p, content: ~p, invalid state: ~p", [UUID, Topic, Content, state_map(State)]),
{keep_state, State, [{reply, From, {error, <<"主机离线，发送失败"/utf8>>}}]}
end;
@ -248,12 +248,12 @@ handle_event({call, From}, {pub, Topic, Qos, Content}, ?STATE_ACTIVATED, State =
handle_event({call, From}, {command, CommandType, Command}, ?STATE_ACTIVATED, State = #state{uuid = UUID, channel_pid = ChannelPid, has_session = HasSession}) ->
case HasSession andalso is_pid(ChannelPid) of
true ->
lager:debug("[iot_host] host: ~p, command_type: ~p, command: ~p", [UUID, CommandType, Command]),
logger:debug("[iot_host] host: ~p, command_type: ~p, command: ~p", [UUID, CommandType, Command]),
%% 通过websocket发送请求
tcp_channel:command(ChannelPid, CommandType, Command),
{keep_state, State, [{reply, From, ok}]};
false ->
lager:debug("[iot_host] host: ~p, command_type: ~p, command: ~p, invalid state: ~p", [UUID, CommandType, Command, state_map(State)]),
logger:debug("[iot_host] host: ~p, command_type: ~p, command: ~p, invalid state: ~p", [UUID, CommandType, Command, state_map(State)]),
{keep_state, State, [{reply, From, {error, <<"主机离线，发送指令失败"/utf8>>}}]}
end;
@ -261,10 +261,10 @@ handle_event({call, From}, {command, CommandType, Command}, ?STATE_ACTIVATED, St
handle_event({call, From}, {activate, true}, _, State = #state{uuid = UUID, channel_pid = ChannelPid}) ->
case is_pid(ChannelPid) of
true ->
lager:debug("[iot_host] uuid: ~p, activate: true", [UUID]),
logger:debug("[iot_host] uuid: ~p, activate: true", [UUID]),
tcp_channel:command(ChannelPid, ?COMMAND_AUTH, <<1:8>>);
false ->
lager:debug("[iot_host] uuid: ~p, activate: true, no channel", [UUID])
logger:debug("[iot_host] uuid: ~p, activate: true, no channel", [UUID])
end,
{next_state, ?STATE_ACTIVATED, State, [{reply, From, ok}]};
@ -273,10 +273,10 @@ handle_event({call, From}, {activate, false}, _, State = #state{uuid = UUID, cha
case is_pid(ChannelPid) of
true ->
tcp_channel:command(ChannelPid, ?COMMAND_AUTH, <<0:8>>),
lager:debug("[iot_host] uuid: ~p, activate: false", [UUID]),
logger:debug("[iot_host] uuid: ~p, activate: false", [UUID]),
tcp_channel:stop(ChannelPid, closed);
false ->
lager:debug("[iot_host] uuid: ~p, activate: false, no channel", [UUID])
logger:debug("[iot_host] uuid: ~p, activate: false, no channel", [UUID])
end,
{next_state, ?STATE_DENIED, State#state{channel_pid = undefined, has_session = false}, [{reply, From, ok}]};
@ -287,30 +287,30 @@ handle_event({call, From}, {attach_channel, ChannelPid}, StateName, State = #sta
erlang:monitor(process, ChannelPid),
%% 更新主机为在线状态
ChangeResult = iot_api_client:change_host_status(UUID, ?HOST_ONLINE),
lager:debug("[iot_host] host_id(attach_channel) uuid: ~p, will change status, result: ~p", [UUID, ChangeResult]),
logger:debug("[iot_host] host_id(attach_channel) uuid: ~p, will change status, result: ~p", [UUID, ChangeResult]),
{keep_state, State#state{channel_pid = ChannelPid, has_session = true}, [{reply, From, ok}]};
%% 主机未激活
?STATE_DENIED ->
lager:notice("[iot_host] attach_channel host_id uuid: ~p, channel: ~p, host inactivated", [UUID, ChannelPid]),
logger:notice("[iot_host] attach_channel host_id uuid: ~p, channel: ~p, host inactivated", [UUID, ChannelPid]),
erlang:monitor(process, ChannelPid),
{keep_state, State#state{channel_pid = ChannelPid}, [{reply, From, {denied, <<"host inactivated">>}}]}
end;
%% 已经绑定了channel
handle_event({call, From}, {attach_channel, _}, _, State = #state{uuid = UUID, channel_pid = OldChannelPid}) ->
lager:notice("[iot_host] attach_channel host_id uuid: ~p, old channel exists: ~p", [UUID, OldChannelPid]),
logger:notice("[iot_host] attach_channel host_id uuid: ~p, old channel exists: ~p", [UUID, OldChannelPid]),
{keep_state, State, [{reply, From, {error, <<"channel existed">>}}]};
%% 数据分发
handle_event(cast, {handle, {data, #data{route_key = RouteKey0, metric = Metric}}}, ?STATE_ACTIVATED,
State = #state{uuid = UUID, has_session = true}) ->
lager:debug("[iot_host] metric_data host: ~p, route_key: ~p, metric: ~p", [UUID, RouteKey0, Metric]),
logger:debug("[iot_host] metric_data host: ~p, route_key: ~p, metric: ~p", [UUID, RouteKey0, Metric]),
RouteKey = get_route_key(RouteKey0),
endpoint_subscription:publish(RouteKey, Metric),
{keep_state, State};
%% ping的数据是通过aes加密后的，因此需要在有会话的情况下才行
handle_event(cast, {handle, {ping, Metrics}}, ?STATE_ACTIVATED, State = #state{uuid = UUID, has_session = true}) ->
lager:debug("[iot_host] ping host_id uuid: ~p, get ping: ~p", [UUID, Metrics]),
logger:debug("[iot_host] ping host_id uuid: ~p, get ping: ~p", [UUID, Metrics]),
{keep_state, State#state{metrics = Metrics}};
%% 心跳机制
@ -319,13 +319,13 @@ handle_event(cast, heartbeat, _, State = #state{heartbeat_counter = HeartbeatCou
%% 没有收到心跳包，主机下线, 设备状态不变
handle_event(info, {timeout, _, heartbeat_ticker}, _, State = #state{uuid = UUID, heartbeat_counter = 0, channel_pid = ChannelPid}) ->
lager:warning("[iot_host] uuid: ~p, heartbeat lost, devices will unknown", [UUID]),
logger:warning("[iot_host] uuid: ~p, heartbeat lost, devices will unknown", [UUID]),
{ok, #{<<"status">> := Status}} = iot_api_client:get_host_by_uuid(UUID),
case Status of
?HOST_NOT_JOINED ->
lager:debug("[iot_host] host: ~p, host_maybe_offline, host not joined, can not change to offline", [UUID]);
logger:debug("[iot_host] host: ~p, host_maybe_offline, host not joined, can not change to offline", [UUID]);
?HOST_OFFLINE ->
lager:debug("[iot_host] host: ~p, host_maybe_offline, host now is offline, do nothing", [UUID]);
logger:debug("[iot_host] host: ~p, host_maybe_offline, host now is offline, do nothing", [UUID]);
?HOST_ONLINE ->
iot_api_client:change_host_status(UUID, ?HOST_OFFLINE)
end,
@ -343,15 +343,15 @@ handle_event(info, {timeout, _, heartbeat_ticker}, _, State = #state{}) ->
%% 当websocket断开的时候，主机的状态不一定改变；主机的状态改变通过心跳机制，会话状态需要改变
handle_event(info, {'DOWN', _Ref, process, ChannelPid, Reason}, _, State = #state{uuid = UUID, channel_pid = ChannelPid, has_session = HasSession}) ->
lager:warning("[iot_host] uuid: ~p, channel: ~p, down with reason: ~p, has_session: ~p, state: ~p", [UUID, ChannelPid, Reason, HasSession, State]),
logger:warning("[iot_host] uuid: ~p, channel: ~p, down with reason: ~p, has_session: ~p, state: ~p", [UUID, ChannelPid, Reason, HasSession, State]),
{keep_state, State#state{channel_pid = undefined, has_session = false}};
handle_event(info, {'DOWN', _Ref, process, Pid, Reason}, _, State = #state{uuid = UUID}) ->
lager:debug("[iot_host] uuid: ~p, process_pid: ~p, down with reason: ~p, state: ~p", [UUID, Pid, Reason, State]),
logger:debug("[iot_host] uuid: ~p, process_pid: ~p, down with reason: ~p, state: ~p", [UUID, Pid, Reason, State]),
{keep_state, State};
handle_event(Event, Info, StateName, State = #state{uuid = UUID}) ->
lager:warning("[iot_host] host: ~p, event: ~p, unknown message: ~p, state_name: ~p, state: ~p", [UUID, Event, Info, StateName, state_map(State)]),
logger:warning("[iot_host] host: ~p, event: ~p, unknown message: ~p, state_name: ~p, state: ~p", [UUID, Event, Info, StateName, state_map(State)]),
{keep_state, State}.
%% @private
@ -360,7 +360,7 @@ handle_event(Event, Info, StateName, State = #state{uuid = UUID}) ->
%% necessary cleaning up. When it returns, the gen_statem terminates with
%% Reason. The return value is ignored.
terminate(Reason, _StateName, _State = #state{uuid = UUID, has_session = HasSession}) ->
lager:debug("[iot_host] host: ~p, terminate with reason: ~p, has_session: ~p", [UUID, Reason, HasSession]),
logger:debug("[iot_host] host: ~p, terminate with reason: ~p, has_session: ~p", [UUID, Reason, HasSession]),
ok.
%% @private

1
src/iot.app.src
View File

@ -10,7 +10,6 @@
eredis,
ranch,
cowboy,
lager,
jiffy,
brod,
parse_trans,

8
src/iot_app.erl
View File

@ -34,7 +34,7 @@ start_mnesia() ->
%% 启动数据库
ok = mnesia:start(),
Tables = mnesia:system_info(tables),
lager:debug("[iot_app] tables: ~p", [Tables]),
logger:debug("[iot_app] tables: ~p", [Tables]),
%% 创建数据库表
ok.
@ -65,7 +65,7 @@ start_http_server() ->
},
{ok, Pid} = cowboy:start_clear(http_listener, TransOpts, #{env => #{dispatch => Dispatcher}}),
lager:debug("[http_server] the http server start at: ~p, pid is: ~p", [Port, Pid]).
logger:debug("[http_server] the http server start at: ~p, pid is: ~p", [Port, Pid]).
%% 启动tcp服务
start_tcp_server() ->
@ -86,7 +86,7 @@ start_tcp_server() ->
]
},
{ok, _} = ranch:start_listener(tcp_server, ranch_tcp, TransOpts, tcp_channel, []),
lager:debug("[iot_app] the tcp server start at: ~p", [Port]).
logger:debug("[iot_app] the tcp server start at: ~p", [Port]).
-spec ensure_mnesia_schema() -> any().
ensure_mnesia_schema() ->
@ -99,7 +99,7 @@ ensure_mnesia_schema() ->
ok -> ok;
{error, {_, {already_exists, _}}} -> ok;
Error ->
lager:debug("[iot_app] create mnesia schema failed with error: ~p", [Error]),
logger:debug("[iot_app] create mnesia schema failed with error: ~p", [Error]),
throw({init_schema, Error})
end
end.

26
src/transport/http/endpoint_handler.erl
View File

@ -40,7 +40,7 @@ handle_request("POST", "/endpoint/start", _, #{<<"id">> := Id}) when is_integer(
{ok, Pid} when is_pid(Pid) ->
{ok, 200, iot_util:json_data(<<"success">>)};
{error, Reason} ->
lager:warning("[endpoint_handler] start endpoint: ~p, get error: ~p", [Title, Reason]),
logger:warning("[endpoint_handler] start endpoint: ~p, get error: ~p", [Title, Reason]),
{ok, 200, iot_util:json_error(404, <<"start endpoint error">>)}
end;
error ->
@ -57,7 +57,7 @@ handle_request("POST", "/endpoint/stop", _, #{<<"id">> := Id}) when is_integer(I
ok ->
{ok, 200, iot_util:json_data(<<"success">>)};
{error, Reason} ->
lager:warning("[endpoint_handler] stop endpoint id: ~p, get error: ~p", [Id, Reason]),
logger:warning("[endpoint_handler] stop endpoint id: ~p, get error: ~p", [Id, Reason]),
{ok, 200, iot_util:json_error(404, <<"stop endpoint error">>)}
end
end;
@ -75,7 +75,7 @@ handle_request("POST", "/endpoint/restart", _, #{<<"id">> := Id}) when is_intege
{ok, Pid} when is_pid(Pid) ->
{ok, 200, iot_util:json_data(<<"success">>)};
{error, Reason} ->
lager:warning("[endpoint_handler] start endpoint: ~p, get error: ~p", [Title, Reason]),
logger:warning("[endpoint_handler] start endpoint: ~p, get error: ~p", [Title, Reason]),
{ok, 200, iot_util:json_error(404, <<"restart endpoint error">>)}
end;
Pid when is_pid(Pid) ->
@ -85,11 +85,11 @@ handle_request("POST", "/endpoint/restart", _, #{<<"id">> := Id}) when is_intege
{ok, Pid0} when is_pid(Pid0) ->
{ok, 200, iot_util:json_data(<<"success">>)};
{error, Reason} ->
lager:warning("[endpoint_handler] start endpoint: ~p, get error: ~p", [Title, Reason]),
logger:warning("[endpoint_handler] start endpoint: ~p, get error: ~p", [Title, Reason]),
{ok, 200, iot_util:json_error(404, <<"restart endpoint error">>)}
end;
{error, Reason} ->
lager:warning("[endpoint_handler] start endpoint: ~p, get error: ~p", [Title, Reason]),
logger:warning("[endpoint_handler] start endpoint: ~p, get error: ~p", [Title, Reason]),
{ok, 200, iot_util:json_error(404, <<"stop endpoint error">>)}
end
end;
@ -106,7 +106,7 @@ handle_request("POST", "/endpoint/test", _, #{<<"protocol">> := <<"http">>, <<"c
{ok, _} ->
{ok, 200, iot_util:json_data(<<"ok">>)};
{error, Reason} ->
lager:debug("[endpint_handler] test http: ~p, error: ~p", [Url, Reason]),
logger:debug("[endpint_handler] test http: ~p, error: ~p", [Url, Reason]),
{ok, 200, iot_util:json_error(-1, <<"url failed">>)}
end;
@ -133,23 +133,23 @@ handle_request("POST", "/endpoint/test", _, #{<<"protocol">> := <<"mqtt">>, <<"c
case emqtt:start_link(Opts) of
{ok, ConnPid} ->
lager:debug("[endpint_handler] start connect, options: ~p", [Opts]),
logger:debug("[endpint_handler] start connect, options: ~p", [Opts]),
case catch emqtt:connect(ConnPid, 5000) of
{ok, _} ->
lager:debug("[endpint_handler] connect success, pid: ~p", [ConnPid]),
logger:debug("[endpint_handler] connect success, pid: ~p", [ConnPid]),
emqtt:stop(ConnPid),
{ok, 200, iot_util:json_data(<<"ok">>)};
{error, Reason} ->
lager:warning("[endpint_handler] connect get error: ~p", [Reason]),
logger:warning("[endpint_handler] connect get error: ~p", [Reason]),
emqtt:stop(ConnPid),
{ok, 200, iot_util:json_error(-1, <<"connect mqtt server failed">>)};
Error ->
lager:warning("[endpint_handler] connect get error: ~p", [Error]),
logger:warning("[endpint_handler] connect get error: ~p", [Error]),
emqtt:stop(ConnPid),
{ok, 200, iot_util:json_error(-1, <<"connect mqtt server failed">>)}
end;
Other ->
lager:warning("[endpint_handler] test connect mqtt with options: ~p, get error: ~p", [Opts, Other]),
logger:warning("[endpint_handler] test connect mqtt with options: ~p, get error: ~p", [Opts, Other]),
{ok, 200, iot_util:json_error(-1, <<"connect mqtt server failed">>)}
end;
{error, Errors} ->
@ -180,11 +180,11 @@ handle_request("POST", "/endpoint/test", _, #{<<"protocol">> := <<"kafka">>, <<"
ok = brod:stop_client(ClientId),
{ok, 200, iot_util:json_data(<<"ok">>)};
{error, Reason} ->
lager:debug("[endpint_handler] start_producer: ~p, get error: ~p", [ClientId, Reason]),
logger:debug("[endpint_handler] start_producer: ~p, get error: ~p", [ClientId, Reason]),
{ok, 200, iot_util:json_error(-1, <<"config kafka server failed">>)}
end;
Error ->
lager:debug("[endpint_handler] start_client: ~p, get error: ~p", [ClientId, Error]),
logger:debug("[endpint_handler] start_client: ~p, get error: ~p", [ClientId, Error]),
{ok, 200, iot_util:json_error(-1, <<"config kafka server failed">>)}
end;
{error, Errors} ->

2
src/transport/http/event_stream_handler.erl
View File

@ -21,7 +21,7 @@ init(Req0, Opts) ->
#{<<"task_id">> := TaskId0} = GetParams,
TaskId = binary_to_integer(TaskId0),
lager:debug("method: ~p, path: ~p, get: ~p", [Method, Path, GetParams]),
logger:debug("method: ~p, path: ~p, get: ~p", [Method, Path, GetParams]),
Req1 = cowboy_req:stream_reply(200, #{
<<"Content-Type">> => <<"text/event-stream">>,
<<"Cache-Control">> => <<"no-cache">>,

12
src/transport/http/host_handler.erl
View File

@ -18,7 +18,7 @@
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
handle_request("GET", "/host/metric", #{<<"uuid">> := UUID}, _) ->
lager:debug("[host_handler] get host metric uuid is: ~p", [UUID]),
logger:debug("[host_handler] get host metric uuid is: ~p", [UUID]),
case iot_host:get_pid(UUID) of
undefined ->
{ok, 200, iot_util:json_error(404, <<"host not found">>)};
@ -56,10 +56,10 @@ handle_request("POST", "/host/delete", _, #{<<"uuid">> := UUID}) when is_binary(
handle_request("POST", "/host/activate", _, #{<<"uuid">> := UUID, <<"auth">> := true}) when is_binary(UUID) ->
case iot_host_sup:ensured_host_started(UUID) of
{error, Reason} ->
lager:debug("[host_handler] activate host_id: ~p, failed with reason: ~p", [UUID, Reason]),
logger:debug("[host_handler] activate host_id: ~p, failed with reason: ~p", [UUID, Reason]),
{ok, 200, iot_util:json_error(400, <<"host not found">>)};
{ok, Pid} when is_pid(Pid) ->
lager:debug("[host_handler] activate host_id: ~p, start", [UUID]),
logger:debug("[host_handler] activate host_id: ~p, start", [UUID]),
ok = iot_host:activate(Pid, true),
{ok, 200, iot_util:json_data(<<"success">>)}
@ -69,10 +69,10 @@ handle_request("POST", "/host/activate", _, #{<<"uuid">> := UUID, <<"auth">> :=
handle_request("POST", "/host/activate", _, #{<<"uuid">> := UUID, <<"auth">> := false}) when is_binary(UUID) ->
case iot_host_sup:ensured_host_started(UUID) of
{error, Reason} ->
lager:debug("[host_handler] activate host_id: ~p, failed with reason: ~p", [UUID, Reason]),
logger:debug("[host_handler] activate host_id: ~p, failed with reason: ~p", [UUID, Reason]),
{ok, 200, iot_util:json_error(400, <<"host not found">>)};
{ok, Pid} when is_pid(Pid) ->
lager:debug("[host_handler] activate host_id: ~p, start", [UUID]),
logger:debug("[host_handler] activate host_id: ~p, start", [UUID]),
ok = iot_host:activate(Pid, false),
{ok, 200, iot_util:json_data(<<"success">>)}
@ -85,7 +85,7 @@ handle_request("POST", "/host/pub", _, #{<<"uuid">> := UUID, <<"topic">> := Topi
Qos = case Qos0 > 0 of true -> 1; false -> 0 end,
case iot_host_sup:ensured_host_started(UUID) of
{error, Reason} ->
lager:debug("[host_handler] pub host_id: ~p, topic: ~p, failed with reason: ~p", [UUID, Topic, Reason]),
logger:debug("[host_handler] pub host_id: ~p, topic: ~p, failed with reason: ~p", [UUID, Topic, Reason]),
{ok, 200, iot_util:json_error(400, <<"host not found">>)};
{ok, Pid} when is_pid(Pid) ->
ok = iot_host:pub(Pid, Topic, Qos, Content),

4
src/transport/http/http_protocol.erl
View File

@ -21,7 +21,7 @@ init(Req0, Opts = [Mod|_]) ->
try Mod:handle_request(Method, Path, GetParams, PostParams) of
{ok, StatusCode, Resp} ->
lager:debug("[http_protocol] request path: ~p, get_params: ~p, post_params: ~p, response: ~ts",
logger:debug("[http_protocol] request path: ~p, get_params: ~p, post_params: ~p, response: ~ts",
[Path, GetParams, PostParams, Resp]),
AcceptEncoding = cowboy_req:header(<<"accept-encoding">>, Req1, <<>>),
Req2 = case iolist_size(Resp) >= 1024 andalso supported_gzip(AcceptEncoding) of
@ -45,7 +45,7 @@ init(Req0, Opts = [Mod|_]) ->
}, ErrResp, Req1),
{ok, Req2, Opts};
_:Error:Stack ->
lager:warning("[http_handler] get error: ~p, stack: ~p", [Error, Stack]),
logger:warning("[http_handler] get error: ~p, stack: ~p", [Error, Stack]),
Req2 = cowboy_req:reply(500, #{
<<"Content-Type">> => <<"text/html;charset=utf-8">>
}, <<"Internal Server Error">>, Req1),

26
src/transport/tcp/tcp_channel.erl
View File

@ -65,7 +65,7 @@ start_link(Ref, Transport, Opts) ->
init(Ref, Transport, _Opts = []) ->
{ok, Socket} = ranch:handshake(Ref),
lager:debug("[sdlan_channel] get a new connection: ~p", [Socket]),
logger:debug("[sdlan_channel] get a new connection: ~p", [Socket]),
Transport:setopts(Socket, [binary, {active, true}, {packet, 4}]),
% erlang:start_timer(?PING_TICKER, self(), ping_ticker),
gen_server:enter_loop(?MODULE, [], #state{transport = Transport, socket = Socket}).
@ -93,12 +93,12 @@ handle_cast({jsonrpc_call, ReceiverPid, Ref, CallBin}, State = #state{transport
%% auth验证
handle_info({tcp, Socket, <<?PACKET_REQUEST, PacketId:32, RequestBin/binary>>}, State = #state{transport = Transport, socket = Socket}) ->
{ok, #auth_request{uuid = UUID, username = Username, token = Token, salt = Salt, timestamp = Timestamp}} = message_codec:decode(RequestBin),
lager:debug("[ws_channel] auth uuid: ~p", [UUID]),
logger:debug("[ws_channel] auth uuid: ~p", [UUID]),
case iot_auth:check(Username, Token, UUID, Salt, Timestamp) of
true ->
case iot_api_client:get_host_by_uuid(UUID) of
undefined ->
lager:warning("[ws_channel] uuid: ~p, user: ~p, host not found", [UUID, Username]),
logger:warning("[ws_channel] uuid: ~p, user: ~p, host not found", [UUID, Username]),
{stop, State};
{ok, _} ->
%% 尝试启动主机的服务进程
@ -115,20 +115,20 @@ handle_info({tcp, Socket, <<?PACKET_REQUEST, PacketId:32, RequestBin/binary>>},
erlang:monitor(process, HostPid),
AuthReplyBin = message_codec:encode(?MESSAGE_AUTH_REPLY, #auth_reply{code = 1, payload = Reason}),
Transport:send(Socket, <<?PACKET_RESPONSE, PacketId:32, AuthReplyBin/binary>>),
lager:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]),
logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]),
{noreply, State#state{uuid = UUID, host_pid = HostPid}};
{error, Reason} when is_binary(Reason) ->
AuthReplyBin = message_codec:encode(?MESSAGE_AUTH_REPLY, #auth_reply{code = 2, payload = Reason}),
Transport:send(Socket, <<?PACKET_RESPONSE, PacketId:32, AuthReplyBin/binary>>),
lager:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]),
logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]),
{stop, State}
end
end;
false ->
lager:warning("[ws_channel] uuid: ~p, user: ~p, auth failed", [UUID, Username]),
logger:warning("[ws_channel] uuid: ~p, user: ~p, auth failed", [UUID, Username]),
{stop, State}
end;
@ -140,7 +140,7 @@ handle_info({tcp, Socket, <<?PACKET_CAST, CastBin/binary>>}, State = #state{sock
#task_event_stream{task_id = TaskId, type = <<"close">>, stream = Reason} ->
iot_event_stream_observer:stream_close(TaskId, Reason);
#task_event_stream{task_id = TaskId, type = Type, stream = Stream} ->
lager:debug("[tcp_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream]),
logger:debug("[tcp_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream]),
iot_event_stream_observer:stream_data(TaskId, Type, Stream)
end,
{noreply, State};
@ -161,17 +161,17 @@ handle_info({tcp, Socket, <<?PACKET_RESPONSE, PacketId:32, ResponseBin/binary>>}
true ->
ReceiverPid ! {jsonrpc_reply, Ref, RpcReply};
false ->
lager:warning("[ws_channel] get async_call_reply message: ~p, packet_id: ~p, but receiver_pid is deaded", [RpcReply, PacketId])
logger:warning("[ws_channel] get async_call_reply message: ~p, packet_id: ~p, but receiver_pid is deaded", [RpcReply, PacketId])
end,
{noreply, State#state{inflight = NInflight}}
end;
handle_info({tcp_error, Sock, Reason}, State = #state{socket = Sock}) ->
lager:notice("[sdlan_channel] tcp_error: ~p", [Reason]),
logger:notice("[sdlan_channel] tcp_error: ~p", [Reason]),
{stop, normal, State};
handle_info({tcp_closed, Sock}, State = #state{socket = Sock}) ->
lager:notice("[sdlan_channel] tcp_closed"),
logger:notice("[sdlan_channel] tcp_closed"),
{stop, normal, State};
%% 关闭当前通道
@ -180,15 +180,15 @@ handle_info({stop, Reason}, State) ->
%% 主机进程挂掉的时候，需要关闭掉链接
handle_info({'DOWN', _, process, HostPid, Reason}, State = #state{uuid = UUID, host_pid = HostPid}) ->
lager:debug("[ws_channel] uuid: ~p, channel will close because host exited with reason: ~p", [UUID, Reason]),
logger:debug("[ws_channel] uuid: ~p, channel will close because host exited with reason: ~p", [UUID, Reason]),
{stop, State};
handle_info(Info, State) ->
lager:warning("[sdlan_channel] get a unknown message: ~p, channel will closed, state: ~p", [Info, State]),
logger:warning("[sdlan_channel] get a unknown message: ~p, channel will closed, state: ~p", [Info, State]),
{noreply, State}.
terminate(Reason, #state{}) ->
lager:warning("[sdlan_channel] stop with reason: ~p", [Reason]),
logger:warning("[sdlan_channel] stop with reason: ~p", [Reason]),
ok.
code_change(_OldVsn, State, _Extra) ->

2
src/transport/udp/iot_udp_handler.erl → src/transport/udp/udp_server.erl
View File

@ -6,7 +6,7 @@
%%% @end
%%% Created : 31. 8月 2023 13:48
%%%-------------------------------------------------------------------
-module(iot_udp_handler).
-module(udp_server).
-author("aresei").
%% API