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

Merge branch 'master' of http://g.s5s8.com/alc/iot

Browse Source
This commit is contained in:
anlicheng 2023-03-12 20:51:16 +08:00
commit 7044345c0b
30 changed files with 1682 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
apps/iot/include/iot.hrl
View File

@ -30,6 +30,8 @@
-record(memory_metric, {
%% 使用量
used = 0,
%% 空余
free = 0,
%% 总量
total = 0
}).
@ -38,6 +40,8 @@
-record(disk_metric, {
%% 使用量
used = 0,
%% 空余
free = 0,
%% 总量
total = 0
}).
@ -84,6 +88,12 @@
interfaces = []
}).
%% id生成器
-record(id_generator, {
name :: string(),
seq = 0 :: integer()
}).
%% 主机定义
-record(host, {
%% ID
@ -94,6 +104,8 @@
model :: binary(),
%% 单元网格编号
cell_id :: integer(),
%% rsa公钥
public_rsa :: binary(),
%% aes的key, 后续通讯需要基于这个加密
aes :: binary(),
metric = #host_metric{},
@ -154,6 +166,61 @@
status = 0
}).
%% scenario 应用场景
-record(scenario, {
%% 场景id，自增id
scenario_id :: integer(),
%% 名称
name :: binary(),
%% 场景描述
desc :: binary(),
%% 场景规则数据
rule :: binary(),
%% 最后更新时间
update_ts = 0 :: integer(),
%% 状态
status :: integer()
}).
%% 场景部署关系表
-record(scenario_deploy, {
%% id，自增id
deploy_id :: integer(),
%% 场景id
scenario_id :: integer(),
%% 主机id
host_id :: binary(),
%% 创建时间
create_ts = 0 :: integer(),
%% 更新时间
update_ts = 0 :: integer(),
%% 状态
status = 0:: integer()
}).
%% 工单管理
-record(issue, {
issue_id :: integer(),
%% 工单名称
name :: binary(),
%% 创建用户
uid :: integer(),
%% 部署类型
deploy_type :: integer(),
%% 关联的id, 微服务id或者场景id
assoc_id :: any(),
%% 主机的id列表
hosts = [] :: list(),
%% 超时时间
timeout = 0 :: integer(),
%% 部署结果
results = [],
%% 创建时间
create_ts = 0 :: integer(),
%% 工单状态
status = 0 :: integer()
}).
-record(http_endpoint, {
url = <<>>
}).
@ -164,6 +231,7 @@
%% 数据转换规则表
-record(router, {
router_id :: integer(),
%% 名称
name = <<>>,
%% 数据过滤规则

3
apps/iot/src/http_api_handler.erl → apps/iot/src/http_handler/http_api_handler.erl
View File

@ -18,5 +18,6 @@
handle_request("GET", "/api/booking", _, _Params) ->
{ok, 200, iot_util:json_data(<<"success">>)};
handle_request("POST", "/api/booking", _, _Params) ->
handle_request("POST", _, _, Params) ->
lager:debug("body is: ~p", [Params]),
{ok, 200, iot_util:json_data(<<"success">>)}.

19
apps/iot/src/http_host_handler.erl → apps/iot/src/http_handler/http_host_handler.erl
View File

@ -72,7 +72,6 @@ handle_request(_, "/host/list", Params, PostParams) ->
handle_request("GET", "/host/detail", #{<<"id">> := HostId}, _) ->
lager:debug("[host_handler] detail id is: ~p", [HostId]),
timer:tc(),
case host_model:get_host(HostId) of
undefined ->
{ok, 200, iot_util:json_error(404, <<"host not found">>)};
@ -87,7 +86,23 @@ handle_request("GET", "/host/detail", #{<<"id">> := HostId}, _) ->
Services = lists:map(fun(S) -> service_model:to_map(S) end, Services0),
HostInfo2 = maps:put(<<"services">>, Services, HostInfo1),
{ok, 200, iot_util:json_data(HostInfo2)}
%% 获取部署应用场景
{ok, DeployList0} = scenario_deploy_model:get_host_deploy_list(HostId),
DeployList = lists:map(fun(E) -> scenario_deploy_model:to_map(E) end, DeployList0),
%% 获取部署的场景信息
DeployList1 = lists:map(fun(DeployInfo = #{<<"scenario_id">> := ScenarioId}) ->
case scenario_model:get_scenario(ScenarioId) of
{ok, Scenario} ->
ScenarioInfo = scenario_model:to_map(Scenario),
DeployInfo#{<<"scenario_info">> => ScenarioInfo};
undefined ->
DeployInfo#{<<"scenario_info">> => #{}}
end
end, DeployList),
HostInfo3 = maps:put(<<"deploy_list">>, HostInfo2, DeployList1),
{ok, 200, iot_util:json_data(HostInfo3)}
end;
handle_request("POST", "/host/update", _, _Params) ->

163
apps/iot/src/http_handler/http_iot_handler.erl Normal file
View File

@ -0,0 +1,163 @@
%%%-------------------------------------------------------------------
%%% @author licheng5
%%% @copyright (C) 2023, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 06. 3月 2023 14:29
%%%-------------------------------------------------------------------
-module(http_iot_handler).
-author("licheng5").
-include("iot.hrl").
%% API
-export([handle_request/4]).
handle_request("GET", "/api/booking", _, _Params) ->
{ok, 200, iot_util:json_data(<<"success">>)};
%% 下发参数
handle_request("POST", "/iot/send_params", _, PostParams = #{<<"host_id">> := HostId, <<"service_name">> := ServiceName, <<"params">> := Params}) ->
lager:debug("body is: ~p", [PostParams]),
case host_model:activate(HostId) of
{error, Reason} when is_binary(Reason) ->
{ok, 200, iot_util:json_error(404, Reason)};
{ok, #host{aes = Aes}} ->
Reply = #{
<<"t">> => 1,
<<"b">> => #{
<<"t_id">> => iot_util:uuid() ,
<<"to">> => ServiceName,
<<"t">> => 10,
<<"m">> => Params
}
},
EncMsg = iot_cipher_aes:encrypt(Aes, Aes, Reply),
iot_emqtt_client:publish(<<"clients.cmd.", HostId/binary>>, EncMsg, 1),
lager:debug("enc_reply is: ~p", [EncMsg]),
{ok, 200, iot_util:json_data(<<"success">>)}
end;
%% 下发采集项
handle_request("POST", "/iot/send_metrics", _,
PostParams = #{<<"host_id">> := HostId, <<"service_name">> := ServiceName, <<"metrics">> := Metrics}) ->
lager:debug("body is: ~p", [PostParams]),
case host_model:activate(HostId) of
{error, Reason} when is_binary(Reason) ->
{ok, 200, iot_util:json_error(404, Reason)};
{ok, #host{aes = Aes}} ->
Reply = #{
<<"t">> => 2,
<<"b">> => #{
<<"t_id">> => iot_util:uuid() ,
<<"to">> => ServiceName,
<<"t">> => 10,
<<"m">> => Metrics
}
},
EncMsg = iot_cipher_aes:encrypt(Aes, Aes, Reply),
iot_emqtt_client:publish(<<"clients.cmd.", HostId/binary>>, EncMsg, 1),
lager:debug("enc_reply is: ~p", [EncMsg]),
{ok, 200, iot_util:json_data(<<"success">>)}
end;
%% 下发微服务
handle_request("POST", "/iot/send_mirco_service", _,
PostParams = #{<<"host_id">> := HostId, <<"args">> := Args}) ->
lager:debug("body is: ~p", [PostParams]),
case host_model:activate(HostId) of
{error, Reason} when is_binary(Reason) ->
{ok, 200, iot_util:json_error(404, Reason)};
{ok, #host{aes = Aes}} ->
Reply = #{
<<"t">> => 3,
<<"b">> => #{
<<"t_id">> => iot_util:uuid() ,
<<"t">> => 10,
<<"m">> => Args
}
},
EncMsg = iot_cipher_aes:encrypt(Aes, Aes, Reply),
iot_emqtt_client:publish(<<"clients.cmd.", HostId/binary>>, EncMsg, 1),
lager:debug("enc_reply is: ~p", [EncMsg]),
{ok, 200, iot_util:json_data(<<"success">>)}
end;
%% 下发数据流图
handle_request("POST", "/iot/send_data_flow", _,
PostParams = #{<<"host_id">> := HostId, <<"args">> := Args}) ->
lager:debug("body is: ~p", [PostParams]),
case host_model:activate(HostId) of
{error, Reason} when is_binary(Reason) ->
{ok, 200, iot_util:json_error(404, Reason)};
{ok, #host{aes = Aes}} ->
Reply = #{
<<"t">> => 4,
<<"b">> => #{
<<"t_id">> => iot_util:uuid() ,
<<"t">> => 10,
<<"m">> => Args
}
},
EncMsg = iot_cipher_aes:encrypt(Aes, Aes, Reply),
iot_emqtt_client:publish(<<"clients.cmd.", HostId/binary>>, EncMsg, 1),
lager:debug("enc_reply is: ~p", [EncMsg]),
{ok, 200, iot_util:json_data(<<"success">>)}
end;
%% 处理命令下发
handle_request("POST", "/iot/send_command", _, Params = #{<<"host_id">> := HostId}) ->
lager:debug("body is: ~p", [Params]),
case host_model:activate(HostId) of
{error, Reason} when is_binary(Reason) ->
{ok, 200, iot_util:json_error(404, Reason)};
{ok, #host{aes = Aes, public_rsa = PubKey}} ->
Reply = #{
<<"a">> => true,
<<"aes">> => Aes,
<<"reply">> => <<"client.reply.", HostId/binary>>
},
EncReply = iot_cipher_rsa:encode(Reply, PubKey),
%% TODO 发送消息到终端
lager:debug("enc_reply is: ~p", [EncReply]),
{ok, 200, iot_util:json_data(<<"success">>)}
end;
%% 处理主机的授权
handle_request("POST", "/iot/host_auth", _, Params = #{<<"host_id">> := HostId}) ->
lager:debug("body is: ~p", [Params]),
case host_model:activate(HostId) of
{error, Reason} when is_binary(Reason) ->
{ok, 200, iot_util:json_error(404, Reason)};
{ok, #host{aes = Aes, public_rsa = PubKey}} ->
Reply = #{
<<"a">> => true,
<<"aes">> => Aes,
<<"reply">> => <<"client.reply.", HostId/binary>>
},
EncReply = iot_cipher_rsa:encode(Reply, PubKey),
%% TODO 发送消息到终端
lager:debug("enc_reply is: ~p", [EncReply]),
{ok, 200, iot_util:json_data(<<"success">>)}
end.

74
apps/iot/src/http_handler/http_router_handler.erl Normal file
View File

@ -0,0 +1,74 @@
%%%-------------------------------------------------------------------
%%% @author licheng5
%%% @copyright (C) 2020, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 26. 4月 2020 3:36 下午
%%%-------------------------------------------------------------------
-module(http_router_handler).
-author("licheng5").
-include("iot.hrl").
%% API
-export([handle_request/4]).
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% helper methods
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
handle_request(_, "/router/list", Params, PostParams) ->
Page0 = maps:get(<<"page">>, Params, <<"1">>),
Size0 = maps:get(<<"size">>, Params, <<"10">>),
Page = binary_to_integer(Page0),
Size = binary_to_integer(Size0),
true = Page > 0 andalso Size > 0,
Start = (Page - 1) * Size,
%% 处理查询条件
Name = maps:get(<<"name">>, PostParams, <<"">>),
MatchHead = #router{name = '$1', _ = '_'},
Guard = [],
Guard1 = case Name =/= <<"">> of
true ->
[{'=:=', '$1', Name}|Guard];
false ->
Guard
end,
Result = ['$_'],
case router_model:get_routers({MatchHead, Guard1, Result}, Start, Size) of
{ok, Routers, TotalNum} ->
Response = #{
<<"routers">> => lists:map(fun(R) -> router_model:to_map(R) end, Routers),
<<"total_num">> => TotalNum
},
{ok, 200, iot_util:json_data(Response)};
{error, Reason} ->
lager:warning("[host_handler] get a error: ~p", [Reason]),
{ok, 200, iot_util:json_error(404, <<"database error">>)}
end;
handle_request("GET", "/router/detail", #{<<"router_id">> := RouterId}, _) ->
case router_model:get_router(RouterId) of
undefined ->
{ok, 200, iot_util:json_error(404, <<"router not found">>)};
{ok, Router} ->
RouterInfo = router_model:to_map(Router),
{ok, 200, iot_util:json_data(RouterInfo)}
end;
handle_request("POST", "/router/changer_status", _, Params = #{<<"router_id">> := RouterId, <<"status">> := NStatus}) ->
lager:debug("[router_handler] post params is: ~p", [Params]),
case router_model:change_status(RouterId, NStatus) of
ok ->
{ok, 200, iot_util:json_data(<<"success">>)};
{error, _} ->
{ok, 200, iot_util:json_error(404, <<"error">>)}
end.

106
apps/iot/src/http_handler/http_scenario_handler.erl Normal file
View File

@ -0,0 +1,106 @@
%%%-------------------------------------------------------------------
%%% @author licheng5
%%% @copyright (C) 2020, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 26. 4月 2020 3:36 下午
%%%-------------------------------------------------------------------
-module(http_scenario_handler).
-author("licheng5").
-include("iot.hrl").
%% API
-export([handle_request/4]).
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% helper methods
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
handle_request(_, "/scenario/list", Params, PostParams) ->
Page0 = maps:get(<<"page">>, Params, <<"1">>),
Size0 = maps:get(<<"size">>, Params, <<"10">>),
Page = binary_to_integer(Page0),
Size = binary_to_integer(Size0),
true = Page > 0 andalso Size > 0,
Start = (Page - 1) * Size,
%% 处理查询条件
Name = maps:get(<<"name">>, PostParams, <<"">>),
MatchHead = #scenario{name = '$1', _ = '_'},
Guard = [],
Guard1 = case Name =/= <<"">> of
true ->
[{'=:=', '$1', Name}|Guard];
false ->
Guard
end,
Result = ['$_'],
case scenario_model:get_scenario_list({MatchHead, Guard1, Result}, Start, Size) of
{ok, ScenarioList, TotalNum} ->
Response = #{
<<"scenarios">> => lists:map(fun(Host) -> scenario_model:to_map(Host) end, ScenarioList),
<<"total_num">> => TotalNum
},
{ok, 200, iot_util:json_data(Response)};
{error, Reason} ->
lager:warning("[http_scenario_handler] get a error: ~p", [Reason]),
{ok, 200, iot_util:json_error(404, <<"database error">>)}
end;
handle_request("GET", "/scenario/detail", #{<<"id">> := ScenarioId0}, _) ->
ScenarioId = binary_to_integer(ScenarioId0),
case scenario_model:get_scenario(ScenarioId) of
undefined ->
{ok, 200, iot_util:json_error(404, <<"scenario not found">>)};
{ok, Scenario} ->
ScenarioInfo = scenario_model:to_map(Scenario),
%% 获取场景下部署的主机列表
{ok, DeployList0} = scenario_deploy_model:get_scenario_deploy_list(ScenarioId),
DeployList = lists:map(fun(E) -> scenario_deploy_model:to_map(E) end, DeployList0),
ScenarioInfo1 = maps:put(<<"deploy_list">>, ScenarioInfo, DeployList),
%% 获取部署的主机信息
ScenarioInfo2 = lists:map(fun(Info = #{<<"host_id">> := HostId}) ->
case host_model:get_host(HostId) of
{ok, Host} ->
HostInfo = host_model:to_map(Host),
Info#{<<"host_info">> => HostInfo};
undefined ->
Info#{<<"host_info">> => #{}}
end
end, ScenarioInfo1),
{ok, 200, iot_util:json_data(ScenarioInfo2)}
end;
handle_request("POST", "/scenario/change_status", _, #{<<"scenario_id">> := ScenarioId, <<"status">> := Status}) when is_integer(ScenarioId), is_integer(Status) ->
case scenario_model:change_status(ScenarioId, Status) of
ok ->
{ok, 200, iot_util:json_data(<<"success">>)};
{error, Reason} when is_binary(Reason) ->
lager:warning("[http_scenario_handler] change_status get a error: ~p", [Reason]),
{ok, 200, iot_util:json_error(404, Reason)}
end;
%% 部署
handle_request("POST", "/scenario/deploy", _, #{<<"scenario_id">> := ScenarioId, <<"status">> := Status}) when is_integer(ScenarioId), is_integer(Status) ->
case scenario_model:change_status(ScenarioId, Status) of
ok ->
{ok, 200, iot_util:json_data(<<"success">>)};
{error, Reason} when is_binary(Reason) ->
lager:warning("[http_scenario_handler] change_status get a error: ~p", [Reason]),
{ok, 200, iot_util:json_error(404, Reason)}
end;
%% 删除??
handle_request("POST", "/scenario/change_status", _, #{<<"scenario_id">> := ScenarioId, <<"status">> := Status}) when is_integer(ScenarioId), is_integer(Status) ->
case scenario_model:change_status(ScenarioId, Status) of
ok ->
{ok, 200, iot_util:json_data(<<"success">>)};
{error, Reason} when is_binary(Reason) ->
lager:warning("[http_scenario_handler] change_status get a error: ~p", [Reason]),
{ok, 200, iot_util:json_error(404, Reason)}
end.

3
apps/iot/src/iot.app.src
View File

@ -13,8 +13,11 @@
parse_trans,
hackney,
poolboy,
mnesia,
crypto,
public_key,
ssl,
kernel,
stdlib
]},

10
apps/iot/src/iot_app.erl
View File

@ -8,11 +8,15 @@
-behaviour(application).
-export([start/2, stop/1]).
-export([start_http_server/0]).
start(_StartType, _StartArgs) ->
io:setopts([{encoding, unicode}]),
%% 启动数据库
mnesia:start(),
Tables = mnesia:system_info(tables),
iot_util:assert_exec(lists:member(router, [Tables]), fun() -> mnesia:wait_for_tables([router], infinity) end),
%% 加速内存的回收
erlang:system_flag(fullsweep_after, 16),
@ -35,8 +39,10 @@ start_http_server() ->
Dispatcher = cowboy_router:compile([
{'_', [
{"/host/[...]", http_protocol, [http_host_handler]}
]}
{"/host/[...]", http_protocol, [http_host_handler]},
{"/api/[...]", http_protocol, [http_api_handler]},
{"/router/[...]", http_protocol, [http_router_handler]}
]}
]),
TransOpts = [
{port, Port},

119
apps/iot/src/iot_emqtt_client.erl Normal file
View File

@ -0,0 +1,119 @@
%%%-------------------------------------------------------------------
%%% @author licheng5
%%% @copyright (C) 2023, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 06. 3月 2023 15:42
%%%-------------------------------------------------------------------
-module(iot_emqtt_client).
-author("licheng5").
-behaviour(gen_server).
%% API
-export([start_link/0]).
%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
-export([publish/3]).
-define(SERVER, ?MODULE).
-record(state, {
conn_pid :: pid()
}).
%%%===================================================================
%%% API
%%%===================================================================
publish(Topic, Message, Qos) when is_binary(Topic), is_binary(Message), is_integer(Qos), Qos == 0; Qos == 1 ->
gen_server:call(?SERVER, {publish, Topic, Message, Qos}).
%% @doc Spawns the server and registers the local name (unique)
-spec(start_link() ->
{ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
start_link() ->
gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
%%%===================================================================
%%% gen_server callbacks
%%%===================================================================
%% @private
%% @doc Initializes the server
-spec(init(Args :: term()) ->
{ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
{stop, Reason :: term()} | ignore).
init([]) ->
%{ok, ServerOpts} = application:get_env(iot, emqx_server),
%{ok, ConnPid} = emqtt:start_link([{clientid, iot_emqtt_client}, {owner, self()}|ServerOpts]),
%{ok, _Props} = emqtt:connect(ConnPid),
%SubOpts = [{qos, 1}],
%{ok, _Props, _ReasonCodes} = emqtt:subscribe(ConnPid, #{}, [{<<"hello">>, SubOpts}]),
{ok, #state{}}.
%% @private
%% @doc Handling call messages
-spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()},
State :: #state{}) ->
{reply, Reply :: term(), NewState :: #state{}} |
{reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} |
{noreply, NewState :: #state{}} |
{noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
{stop, Reason :: term(), NewState :: #state{}}).
handle_call({publish, Topic, Message, Qos}, _From, State = #state{conn_pid = ConnPid}) ->
if
Qos == 0 ->
ok = emqtt:publish(ConnPid, Topic, #{}, Message, [{qos, 0}]);
Qos == 1 ->
{ok, _PktId} = emqtt:publish(ConnPid, Topic, #{}, Message, [{qos, 1}]);
true ->
ok
end,
{reply, ok, State}.
%% @private
%% @doc Handling cast messages
-spec(handle_cast(Request :: term(), State :: #state{}) ->
{noreply, NewState :: #state{}} |
{noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), NewState :: #state{}}).
handle_cast(_Request, State = #state{}) ->
{noreply, State}.
%% @private
%% @doc Handling all non call/cast messages
-spec(handle_info(Info :: timeout() | term(), State :: #state{}) ->
{noreply, NewState :: #state{}} |
{noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), NewState :: #state{}}).
handle_info(_Info, State = #state{}) ->
{noreply, State}.
%% @private
%% @doc This function is called by a gen_server when it is about to
%% terminate. It should be the opposite of Module:init/1 and do any
%% necessary cleaning up. When it returns, the gen_server terminates
%% with Reason. The return value is ignored.
-spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
State :: #state{}) -> term()).
terminate(_Reason, _State = #state{}) ->
ok.
%% @private
%% @doc Convert process state when code is changed
-spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{},
Extra :: term()) ->
{ok, NewState :: #state{}} | {error, Reason :: term()}).
code_change(_OldVsn, State = #state{}, _Extra) ->
{ok, State}.
%%%===================================================================
%%% Internal functions
%%%===================================================================

40
apps/iot/src/iot_http_client.erl Normal file
View File

@ -0,0 +1,40 @@
%%%-------------------------------------------------------------------
%%% @author licheng5
%%% @copyright (C) 2023, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 03. 3月 2023 11:48
%%%-------------------------------------------------------------------
-module(iot_http_client).
-author("licheng5").
%% API
-export([post/2]).
post(Url, Body) when is_list(Url), is_binary(Body) ->
case hackney:request(post, Url, [], Body) of
{ok, 200, _, ClientRef} ->
case hackney:body(ClientRef) of
{ok, RespBody} ->
lager: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]),
{error, failed}
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]),
ok;
{error, Reason} ->
lager:warning("[iot_http_client] url: ~p, http_code: ~p, get error: ~p", [Url, HttpCode, Reason]),
{error, failed}
end;
{error, Reason} ->
lager:warning("[iot_http_client] url: ~p, get error: ~p", [Url, Reason]),
{error, failed}
end.

119
apps/iot/src/iot_issue.erl Normal file
View File

@ -0,0 +1,119 @@
%%%-------------------------------------------------------------------
%%% @author licheng5
%%% @copyright (C) 2023, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 10. 3月 2023 16:44
%%%-------------------------------------------------------------------
-module(iot_issue).
-author("licheng5").
-include("iot.hrl").
-behaviour(gen_server).
%% API
-export([start_link/1]).
-export([get_pid/1]).
%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
-define(SERVER, ?MODULE).
-record(state, {
issue :: #issue{},
timer_ref
}).
%%%===================================================================
%%% API
%%%===================================================================
get_pid(IssueId) when is_integer(IssueId) ->
whereis(get_name(IssueId)).
get_name(IssueId) when is_integer(IssueId) ->
list_to_atom("iot_issue:" ++ integer_to_list(IssueId)).
%% @doc Spawns the server and registers the local name (unique)
-spec(start_link(Issue :: #issue{}) ->
{ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
start_link(Issue = #issue{issue_id = IssueId}) ->
Name = get_name(IssueId),
gen_server:start_link({local, Name}, ?MODULE, [Issue], []).
%%%===================================================================
%%% gen_server callbacks
%%%===================================================================
%% @private
%% @doc Initializes the server
-spec(init(Args :: term()) ->
{ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
{stop, Reason :: term()} | ignore).
init([Issue = #issue{timeout = Timeout0, hosts = Hosts}]) ->
lager:debug("iot_issue started!!: ~p", [Issue]),
%% 启动任务定时器, 默认最长为1个小时
Timeout = if Timeout0 > 0 -> Timeout0; true -> 3600 end,
TimerRef = erlang:start_timer(Timeout * 1000, self(), issue_task_timeout),
%% TODO 下发数据到主机
lists:map(fun(Host) ->
iot_emqtt_client:publish(Host, x, 1)
end, Hosts),
{ok, #state{issue = Issue, timer_ref = TimerRef}}.
%% @private
%% @doc Handling call messages
-spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()},
State :: #state{}) ->
{reply, Reply :: term(), NewState :: #state{}} |
{reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} |
{noreply, NewState :: #state{}} |
{noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
{stop, Reason :: term(), NewState :: #state{}}).
handle_call(_Request, _From, State = #state{}) ->
{reply, ok, State}.
%% @private
%% @doc Handling cast messages
-spec(handle_cast(Request :: term(), State :: #state{}) ->
{noreply, NewState :: #state{}} |
{noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), NewState :: #state{}}).
handle_cast(_Request, State = #state{}) ->
{noreply, State}.
%% @private
%% @doc Handling all non call/cast messages
-spec(handle_info(Info :: timeout() | term(), State :: #state{}) ->
{noreply, NewState :: #state{}} |
{noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), NewState :: #state{}}).
handle_info(_Info, State = #state{}) ->
{noreply, State}.
%% @private
%% @doc This function is called by a gen_server when it is about to
%% terminate. It should be the opposite of Module:init/1 and do any
%% necessary cleaning up. When it returns, the gen_server terminates
%% with Reason. The return value is ignored.
-spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
State :: #state{}) -> term()).
terminate(_Reason, _State = #state{}) ->
ok.
%% @private
%% @doc Convert process state when code is changed
-spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{},
Extra :: term()) ->
{ok, NewState :: #state{}} | {error, Reason :: term()}).
code_change(_OldVsn, State = #state{}, _Extra) ->
{ok, State}.
%%%===================================================================
%%% Internal functions
%%%===================================================================

73
apps/iot/src/iot_issue_sup.erl Normal file
View File

@ -0,0 +1,73 @@
%%%-------------------------------------------------------------------
%%% @author licheng5
%%% @copyright (C) 2023, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 10. 3月 2023 16:44
%%%-------------------------------------------------------------------
-module(iot_issue_sup).
-author("licheng5").
-include("iot.hrl").
-behaviour(supervisor).
%% API
-export([start_link/0, start_issue/1]).
%% Supervisor callbacks
-export([init/1]).
-define(SERVER, ?MODULE).
%%%===================================================================
%%% API functions
%%%===================================================================
%% @doc Starts the supervisor
-spec(start_link() -> {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
start_link() ->
supervisor:start_link({local, ?SERVER}, ?MODULE, []).
%%%===================================================================
%%% Supervisor callbacks
%%%===================================================================
%% @private
%% @doc Whenever a supervisor is started using supervisor:start_link/[2,3],
%% this function is called by the new process to find out about
%% restart strategy, maximum restart frequency and child
%% specifications.
-spec(init(Args :: term()) ->
{ok, {SupFlags :: {RestartStrategy :: supervisor:strategy(),
MaxR :: non_neg_integer(), MaxT :: non_neg_integer()},
[ChildSpec :: supervisor:child_spec()]}}
| ignore | {error, Reason :: term()}).
init([]) ->
SupFlags = #{strategy => simple_one_for_one, intensity => 1000, period => 3600},
AChild = #{
id => 'iot_issue',
start => {'iot_issue', start_link, []},
restart => temporary,
shutdown => 2000,
type => worker,
modules => ['iot_issue']
},
{ok, {SupFlags, [AChild]}}.
%%%===================================================================
%%% Internal functions
%%%===================================================================
%% 启动一个任务
-spec start_issue(Issue :: #issue{}) -> {ok, Pid :: pid()} | {error, Reason :: any()}.
start_issue(Issue = #issue{}) ->
case supervisor:start_child(?MODULE, [Issue]) of
{ok, Pid} ->
{ok, Pid};
{error, {'already_started', Pid}} ->
{ok, Pid};
Error ->
Error
end.

43
apps/iot/src/iot_mnesia.erl
View File

@ -24,6 +24,14 @@ init_database() ->
ok = mnesia:start(),
%% 创建数据库表
%% id生成表
mnesia:create_table(id_generator, [
{attributes, record_info(fields, id_generator)},
{record_name, id_generator},
{disc_copies, [node()]},
{type, ordered_set}
]),
%% 主机表
mnesia:create_table(host, [
{attributes, record_info(fields, host)},
@ -48,6 +56,38 @@ init_database() ->
{type, ordered_set}
]),
%% 转发规则表
mnesia:create_table(router, [
{attributes, record_info(fields, router)},
{record_name, router},
{disc_copies, [node()]},
{type, ordered_set}
]),
%% 应用场景
mnesia:create_table(scenario, [
{attributes, record_info(fields, scenario)},
{record_name, scenario},
{disc_copies, [node()]},
{type, ordered_set}
]),
%% 应用场景部署关系表
mnesia:create_table(scenario_deploy, [
{attributes, record_info(fields, scenario_deploy)},
{record_name, scenario_deploy},
{disc_copies, [node()]},
{type, ordered_set}
]),
%% 工单表
mnesia:create_table(issue, [
{attributes, record_info(fields, issue)},
{record_name, issue},
{disc_copies, [node()]},
{type, ordered_set}
]),
ok.
%% 加入集群
@ -69,4 +109,7 @@ copy_database(MasterNode) when is_atom(MasterNode) ->
mnesia:add_table_copy(host, node(), ram_copies),
mnesia:add_table_copy(terminal, node(), ram_copies),
mnesia:add_table_copy(service, node(), ram_copies),
mnesia:add_table_copy(scenario, node(), ram_copies),
mnesia:add_table_copy(scenario_deploy, node(), ram_copies),
mnesia:add_table_copy(issue, node(), ram_copies),
ok.

67
apps/iot/src/iot_mock.erl
View File

@ -11,8 +11,23 @@
-include("iot.hrl").
%% API
-export([insert_hosts/0, insert_services/1, insert_terminals/1]).
-export([insert_hosts/0, insert_services/1, insert_terminals/1, insert_routers/0]).
-export([start_router/1]).
-export([rsa_encode/1]).
-export([start_issue/0]).
start_issue() ->
iot_issue_sup:start_issue(#issue{
issue_id = 1,
name = <<"issue 1">>,
uid = 1234,
deploy_type = 1,
assoc_id = 1,
hosts = [<<"host1">>, <<"host2">>],
timeout = 6,
create_ts = 0,
status = 0
}).
insert_hosts() ->
lists:foreach(fun(Id0) ->
@ -53,8 +68,6 @@ insert_services(HostId) ->
insert_terminals(HostId) ->
lists:foreach(fun(Id0) ->
Q0 = queue:new(),
Q = queue:in({1234, 21}, Q0),
Terminal = #terminal{
terminal_id = integer_to_binary(Id0),
host_id = HostId,
@ -71,6 +84,28 @@ insert_terminals(HostId) ->
terminal_model:add_terminal(Terminal)
end, lists:seq(1, 100)).
insert_routers() ->
lists:foreach(fun(Id0) ->
R = #router{
router_id = Id0,
name = <<"计费电表"/utf8>>,
rule = <<"测试规则"/utf8>>,
endpoint = #http_endpoint{url = <<"http://127.0.0.1:8080/data">>},
status = 1
},
router_model:add_router(R)
end, lists:seq(1, 100)).
start_router(Id0) when is_integer(Id0) ->
R = #router{
router_id = Id0,
name = <<"计费电表"/utf8>>,
rule = <<"测试规则"/utf8>>,
endpoint = #http_endpoint{url = <<"http://127.0.0.1:8080/data">>},
status = 1
},
router_model:add_router(R),
iot_router_sup:start_new_router(R).
rsa_encode(Data) when is_binary(Data) ->
%% 读取相关配置
@ -106,28 +141,4 @@ rsa_decode(EncData) when is_binary(EncData) ->
PlainData = public_key:decrypt_private(EncData, PubKey),
lager:debug("plain data is: ~p", [PlainData]),
ok.
scan_and(Tokens) ->
scan_and(Tokens, [], [], 0).
%% 扫描完成并且最后一个表达式为空
scan_and([], [], Acc, 0) ->
lists:reverse(Acc);
%% 扫描完成并且最后一个表达式不为空
scan_and([], Expr, Acc, 0) ->
lists:reverse([{simple, lists:reverse(Expr)}|Acc]);
%% 遇到OR关键词, 并且此时的层级为0
scan_and([32, $O, $R, 32|Tokens], Expr, Acc, 0) ->
scan_and(Tokens, [], [{simple, lists:reverse(Expr)}|Acc], 0);
%% 扫描到左括号 && Level > 0; 此时的Expr需要更多的字符
scan_and([Token|Tokens], Expr, Acc, Level) when Token == $( ->
scan_and(Tokens, [Token|Expr], Acc, Level + 1);
%% 扫描到右括号 && Level == 1; 此时的Expr表达式的内部嵌套的子串扫描完成
scan_and([Token|Tokens], Expr, Acc, 1) when Token == $) ->
scan_and(Tokens, [Token|Expr], Acc, 0);
%% 扫描到右括号 && Level > 1; 此时的Expr表达式的内部嵌套的子串扫描完成，Level的值减1
scan_and([Token|Tokens], Expr, Acc, Level) when Token == $) ->
scan_and(Tokens, [Token|Expr], Acc, Level - 1);
%% 普通字符
scan_and([Token|Tokens], Expr, Acc, Level) ->
scan_and(Tokens, [Token|Expr], Acc, Level).
ok.

28
apps/iot/src/iot_message_handler.erl → apps/iot/src/iot_mqtt_message_handler.erl
View File

@ -6,7 +6,7 @@
%%% @end
%%% Created : 18. 2月 2023 21:39
%%%-------------------------------------------------------------------
-module(iot_message_handler).
-module(iot_mqtt_message_handler).
-author("aresei").
-include("iot.hrl").
@ -81,7 +81,7 @@ handle(<<"server.register">>, Msg = #{<<"c_id">> := ClientId, <<"r">> := PubKey,
handle(<<"server.data">>, #{<<"c_id">> := HostId, <<"d">> := Data}) ->
case host_model:get_host(HostId) of
{ok, #host{aes = Aes}} ->
Services = service_model:get_services(HostId),
Services = service_model:get_host_services(HostId),
PlainData = iot_cipher_aes:decrypt(Aes, Aes, Data),
case jiffy:decode(PlainData, [return_maps]) of
Infos when is_list(Infos) ->
@ -107,7 +107,29 @@ handle(<<"server.data">>, #{<<"c_id">> := HostId, <<"d">> := Data}) ->
end;
undefined ->
lager:warning("[iot_message_handler] host_id: ~p, not exists", [HostId])
end.
end;
%% 处理服务器的ping
handle(<<"server.ping">>, #{<<"c">> := HostId, <<"at">> := At,
<<"h">> := #{<<"fm">> := FreeMemory, <<"fd">> := FreeDisk, <<"cp">> := CpuLoad}}) ->
case host_model:get_host(HostId) of
{ok, Host=#host{metric = Metric = #host_metric{disk = Disk, memory = Memory}}} ->
NMetric = Metric#host_metric{
disk = Disk#disk_metric{free = FreeDisk},
memory = Memory#memory_metric{free = FreeMemory}
},
case mnesia:transaction(fun() -> mnesia:write(host, Host#host{metric = NMetric}, write) end) of
{atomic, ok} ->
ok;
{error, Reason} ->
lager:warning("[iot_message_handler] host_id: ~p, ping get error: ~p", [HostId, Reason])
end;
undefined ->
lager:warning("[iot_message_handler] host_id: ~p, not exists", [HostId])
end.
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% helper methods

105
apps/iot/src/iot_router.erl Normal file
View File

@ -0,0 +1,105 @@
%%%-------------------------------------------------------------------
%%% @author licheng5
%%% @copyright (C) 2023, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 01. 3月 2023 16:03
%%%-------------------------------------------------------------------
-module(iot_router).
-author("licheng5").
-include("iot.hrl").
-behaviour(gen_server).
%% API
-export([start_link/2]).
-export([get_name/1]).
%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
-define(SERVER, ?MODULE).
-record(state, {
}).
get_name(Id) when is_integer(Id) ->
list_to_atom("iot_router:" ++ integer_to_list(Id)).
%%%===================================================================
%%% API
%%%===================================================================
%% @doc Spawns the server and registers the local name (unique)
-spec(start_link(Name :: atom(), Router :: #router{}) ->
{ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
start_link(Name, Router = #router{}) ->
gen_server:start_link({global, Name}, ?MODULE, [Router], []).
%%%===================================================================
%%% gen_server callbacks
%%%===================================================================
%% @private
%% @doc Initializes the server
-spec(init(Args :: term()) ->
{ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
{stop, Reason :: term()} | ignore).
init([Router]) ->
lager:debug("router is: ~p", [Router]),
{ok, #state{}}.
%% @private
%% @doc Handling call messages
-spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()},
State :: #state{}) ->
{reply, Reply :: term(), NewState :: #state{}} |
{reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} |
{noreply, NewState :: #state{}} |
{noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
{stop, Reason :: term(), NewState :: #state{}}).
handle_call(_Request, _From, State = #state{}) ->
{reply, ok, State}.
%% @private
%% @doc Handling cast messages
-spec(handle_cast(Request :: term(), State :: #state{}) ->
{noreply, NewState :: #state{}} |
{noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), NewState :: #state{}}).
handle_cast(_Request, State = #state{}) ->
{noreply, State}.
%% @private
%% @doc Handling all non call/cast messages
-spec(handle_info(Info :: timeout() | term(), State :: #state{}) ->
{noreply, NewState :: #state{}} |
{noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), NewState :: #state{}}).
handle_info(_Info, State = #state{}) ->
{noreply, State}.
%% @private
%% @doc This function is called by a gen_server when it is about to
%% terminate. It should be the opposite of Module:init/1 and do any
%% necessary cleaning up. When it returns, the gen_server terminates
%% with Reason. The return value is ignored.
-spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
State :: #state{}) -> term()).
terminate(_Reason, _State = #state{}) ->
ok.
%% @private
%% @doc Convert process state when code is changed
-spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{},
Extra :: term()) ->
{ok, NewState :: #state{}} | {error, Reason :: term()}).
code_change(_OldVsn, State = #state{}, _Extra) ->
{ok, State}.
%%%===================================================================
%%% Internal functions
%%%===================================================================

86
apps/iot/src/iot_router_sup.erl Normal file
View File

@ -0,0 +1,86 @@
%%%-------------------------------------------------------------------
%%% @author licheng5
%%% @copyright (C) 2023, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 01. 3月 2023 16:01
%%%-------------------------------------------------------------------
-module(iot_router_sup).
-author("licheng5").
-include("iot.hrl").
-behaviour(supervisor).
%% API
-export([start_link/0, start_new_router/1]).
%% Supervisor callbacks
-export([init/1]).
-define(SERVER, ?MODULE).
%%%===================================================================
%%% API functions
%%%===================================================================
%% @doc Starts the supervisor
-spec(start_link() -> {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
start_link() ->
supervisor:start_link({local, ?SERVER}, ?MODULE, []).
%%%===================================================================
%%% Supervisor callbacks
%%%===================================================================
%% @private
%% @doc Whenever a supervisor is started using supervisor:start_link/[2,3],
%% this function is called by the new process to find out about
%% restart strategy, maximum restart frequency and child
%% specifications.
-spec(init(Args :: term()) ->
{ok, {SupFlags :: {RestartStrategy :: supervisor:strategy(),
MaxR :: non_neg_integer(), MaxT :: non_neg_integer()},
[ChildSpec :: supervisor:child_spec()]}}
| ignore | {error, Reason :: term()}).
init([]) ->
SupFlags = #{strategy => one_for_one, intensity => 1000, period => 3600},
%% 启动目前生效的全部转发规则
Specs = case router_model:get_all_valid_routers() of
{ok, Routers} ->
lists:map(fun generate_router_spec/1, Routers);
error ->
[]
end,
{ok, {SupFlags, Specs}}.
%%%===================================================================
%%% Internal functions
%%%===================================================================
%% 动态启动一个转发规则
-spec start_new_router(Router :: #router{}) -> {ok, Pid :: pid()} | {error, Reason :: any()}.
start_new_router(Router = #router{}) ->
Spec = generate_router_spec(Router),
case supervisor:start_child(?MODULE, Spec) of
{ok, Pid} ->
{ok, Pid};
{error, {already_started, Pid}} ->
{ok, Pid};
{error, Error} ->
lager:debug("start router get a error: ~p", [Error]),
{error, Error}
end.
generate_router_spec(Router = #router{router_id = RouterId}) ->
Id = iot_router:get_name(RouterId),
#{
id => Id,
start => {'iot_router', start_link, [Id, Router]},
restart => permanent,
shutdown => 2000,
type => worker,
modules => ['iot_router']
}.

58
apps/iot/src/iot_rule_parser.erl
View File

@ -11,6 +11,8 @@
%% API
-export([parse/1, test/0]).
-export([parse_bracket/1]).
-record(or_expr, {
expr = []
@ -156,4 +158,60 @@ get_tag(0) ->
get_tag(_) ->
complex.
-record(bracket, {
items = []
}).
%% 括号解析
parse_bracket(Tokens) ->
parse_bracket(Tokens, [], []).
parse_bracket([], [], S) ->
#bracket{items = lists:reverse(S)};
parse_bracket([], Acc, S) ->
#bracket{items = lists:reverse([lists:reverse(Acc)|S])};
parse_bracket([$(|Tokens], Acc, S) ->
{ChildElements, RestTokens} = parse_bracket0(Tokens),
Bracket = parse_bracket(ChildElements),
parse_bracket(RestTokens, [], [Bracket, lists:reverse(Acc)|S]);
parse_bracket([H|Tokens], Acc, S) ->
parse_bracket(Tokens, [H|Acc], S).
%% 截取配对的括号
parse_bracket0(Tokens) ->
parse_bracket0(Tokens, [], 1).
parse_bracket0([$)|Tokens], Acc, Level) when Level - 1 == 0 ->
{lists:reverse(Acc), Tokens};
parse_bracket0([$)|Tokens], Acc, Level) ->
parse_bracket0(Tokens, [$)|Acc], Level - 1);
parse_bracket0([$(|Tokens], Acc, Level) ->
parse_bracket0(Tokens, [$(|Acc], Level + 1);
parse_bracket0([H|Tokens], Acc, Level) ->
parse_bracket0(Tokens, [H|Acc], Level).
%% 处理运算符号解析出表达式和操作符号
%{bracket,["id < 3 OR ",
% {bracket,["name = 'anlicheng' OR ",
% {bracket,["name = 'test' AND ",
% {bracket,["y = 1 OR x = 1"]}]}]},
% " OR ",
% {bracket,["1 = 1"]}]}
syntax(Bracket = #bracket{items = Items}) ->
lists:map(fun(E) ->
case E of
B = #bracket{} ->
syntax(B);
Expr ->
string:split(Expr, "AND", all)
end
end, Items),
ok.

33
apps/iot/src/iot_sup.erl
View File

@ -26,10 +26,35 @@ start_link() ->
%% type => worker(), % optional
%% modules => modules()} % optional
init([]) ->
SupFlags = #{strategy => one_for_all,
intensity => 0,
period => 1},
ChildSpecs = [],
SupFlags = #{strategy => one_for_one, intensity => 1000, period => 3600},
ChildSpecs = [
#{
id => 'iot_router_sup',
start => {'iot_router_sup', start_link, []},
restart => permanent,
shutdown => 2000,
type => supervisor,
modules => ['iot_router_sup']
},
#{
id => 'iot_issue_sup',
start => {'iot_issue_sup', start_link, []},
restart => permanent,
shutdown => 2000,
type => supervisor,
modules => ['iot_issue_sup']
},
#{
id => 'iot_emqtt_client',
start => {'iot_emqtt_client', start_link, []},
restart => permanent,
shutdown => 2000,
type => worker,
modules => ['iot_emqtt_client']
}
],
{ok, {SupFlags, ChildSpecs}}.
%% internal functions

10
apps/iot/src/iot_util.erl
View File

@ -13,7 +13,7 @@
-export([timestamp/0, number_format/2, current_time/0]).
-export([step/3, chunks/2, rand_bytes/1, uuid/0]).
-export([json_data/1, json_error/2]).
-export([queue_limited_in/3]).
-export([queue_limited_in/3, assert_exec/2]).
%% 时间，精确到毫秒
timestamp() ->
@ -80,4 +80,10 @@ queue_limited_in(Item, Q, Num) when is_integer(Num) ->
queue:in(Item, Q1);
false ->
queue:in(Item, Q)
end.
end.
-spec assert_exec(boolean(), any()) -> any().
assert_exec(true, Fun) when is_function(Fun, 0) ->
{ok, Fun()};
assert_exec(_, _) ->
aborted.

32
apps/iot/src/model/host_model.erl
View File

@ -12,7 +12,7 @@
-include_lib("stdlib/include/qlc.hrl").
%% API
-export([get_host/1, get_hosts/3, get_stat/0, add_host/1, change_status/2, delete/1, table_size/0, find_hosts/3]).
-export([get_host/1, get_hosts/3, get_stat/0, add_host/1, change_status/2, delete/1, table_size/0, find_hosts/3, activate/1]).
-export([to_map/1]).
get_host(HostId) when is_binary(HostId) ->
@ -97,6 +97,36 @@ change_status(HostId, Status) when is_binary(HostId), is_integer(Status) ->
{error, Reason}
end.
-spec activate(HostId :: binary()) -> ok | {error, Reason :: any()}.
activate(HostId) when is_binary(HostId) ->
Fun = fun() ->
case mnesia:read(host, HostId) of
[] ->
mnesia:abort(<<"host not found">>);
[Host = #host{status = Status}] ->
case Status =:= ?HOST_STATUS_INACTIVE of
true ->
Aes = iot_util:rand_bytes(16),
NHost = Host#host{
aes = Aes,
activated_ts = iot_util:current_time(),
update_ts = iot_util:current_time(),
status = ?HOST_STATUS_ONLINE
},
ok = mnesia:write(host, NHost, write),
NHost;
false ->
mnesia:abort(<<"host status invalid">>)
end
end
end,
case mnesia:transaction(Fun) of
{atomic, Host} ->
{ok, Host};
{aborted, Reason} ->
{error, Reason}
end.
-spec delete(HostId :: binary()) -> ok | {error, Reason :: any()}.
delete(HostId) when is_binary(HostId) ->
case mnesia:transaction(fun() -> mnesia:delete(host, HostId, write) end) of

22
apps/iot/src/model/id_generator_model.erl Normal file
View File

@ -0,0 +1,22 @@
%%%-------------------------------------------------------------------
%%% @author licheng5
%%% @copyright (C) 2021, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 27. 4月 2021 下午4:38
%%%-------------------------------------------------------------------
-module(id_generator_model).
-author("licheng5").
-include("iot.hrl").
%% API
-export([generate/1]).
generate(Name) when is_list(Name) ->
case mnesia:transaction(fun() -> mnesia:dirty_update_counter(id_generator, Name, 1) end) of
{atomic, Id} ->
{ok, Id};
{aborted, Reason} ->
{error, Reason}
end.

109
apps/iot/src/model/issue_model.erl Normal file
View File

@ -0,0 +1,109 @@
%%%-------------------------------------------------------------------
%%% @author licheng5
%%% @copyright (C) 2021, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 27. 4月 2021 下午4:38
%%%-------------------------------------------------------------------
-module(issue_model).
-author("licheng5").
-include("iot.hrl").
-include_lib("stdlib/include/qlc.hrl").
-define(TAB_NAME, issue).
%% API
-export([get_issue/1, get_issues/3, add_issue/1, change_status/2, delete/1, table_size/0, get_user_issues/3]).
-export([to_map/1]).
get_issue(IssueId) when is_integer(IssueId) ->
case mnesia:dirty_read(?TAB_NAME, IssueId) of
[Issue = #issue{}] ->
{ok, Issue};
_ ->
undefined
end.
%% 获取app信息
-spec get_issues(Filter :: any(), Start :: integer(), Limit :: integer()) ->
{ok, Items :: list(), TotalNum :: integer()} |
{error, Reason :: any()}.
get_issues(Spec, Start, Limit) when is_integer(Limit), is_integer(Start), Start >= 0, Limit > 0 ->
Items = mnesia:dirty_select(?TAB_NAME, [Spec]),
NItems = lists:sublist(Items, Start + 1, Limit),
{ok, NItems, length(Items)}.
-spec get_user_issues(UserId :: integer(), Start :: integer(), Limit :: integer()) ->
{ok, Items :: [#host{}], Num :: integer()} |
{error, Reason :: any()}.
get_user_issues(UserId, Start, Limit) when is_integer(UserId), is_integer(Limit), is_integer(Start), Start >= 0, Limit > 0 ->
Fun = fun() ->
Q = qlc:q([E || E <- mnesia:table(?TAB_NAME), E#issue.uid =:= UserId]),
qlc:e(Q)
end,
case mnesia:transaction(Fun) of
{atomic, Items} when is_list(Items) ->
{ok, lists:sublist(Items, Start + 1, Limit), length(Items)};
{aborted, Error} ->
{error, Error}
end.
-spec add_issue(Issue :: #issue{}) -> ok | {error, Reason :: any()}.
add_issue(Issue = #issue{}) ->
case mnesia:transaction(fun() -> mnesia:write(?TAB_NAME, Issue, write) end) of
{atomic, _} ->
ok;
{aborted, Error} ->
{error, Error}
end.
-spec change_status(IssueId :: integer(), Status :: integer()) -> ok | {error, Reason :: any()}.
change_status(IssueId, Status) when is_integer(IssueId), is_integer(Status) ->
Fun = fun() ->
case mnesia:read(?TAB_NAME, IssueId) of
[] ->
mnesia:abort(<<"issue not found">>);
[Issue] ->
mnesia:write(?TAB_NAME, Issue#issue{status = Status}, write)
end
end,
case mnesia:transaction(Fun) of
{atomic, ok} ->
ok;
{aborted, Reason} ->
{error, Reason}
end.
-spec delete(IssueId :: binary()) -> ok | {error, Reason :: any()}.
delete(IssueId) when is_integer(IssueId) ->
case mnesia:transaction(fun() -> mnesia:delete(?TAB_NAME, IssueId, write) end) of
{atomic, ok} ->
ok;
{aborted, Reason} ->
{error, Reason}
end.
%% 获取app表的数据大小
table_size() ->
mnesia:table_info(?TAB_NAME, size).
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% helper methods
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
to_map(#issue{issue_id = IssueId, name = Name, uid = Uid, deploy_type = DeployType, assoc_id = AssocId, hosts = Hosts, timeout = Timeout,
create_ts = CreateTs, results = Results, status = Status}) ->
#{
<<"issue_id">> => IssueId,
<<"name">> => Name,
<<"uid">> => Uid,
<<"deploy_type">> => DeployType,
<<"assoc_id">> => AssocId,
<<"hosts">> => Hosts,
<<"timeout">> => Timeout,
<<"status">> => Status,
<<"create_ts">> => CreateTs,
<<"results">> => Results
}.

124
apps/iot/src/model/router_model.erl Normal file
View File

@ -0,0 +1,124 @@
%%%-------------------------------------------------------------------
%%% @author licheng5
%%% @copyright (C) 2021, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 27. 4月 2021 下午4:38
%%%-------------------------------------------------------------------
-module(router_model).
-author("licheng5").
-include("iot.hrl").
-include_lib("stdlib/include/qlc.hrl").
%% API
-export([get_router/1, get_routers/3, add_router/1, change_status/2, delete/1, table_size/0, get_all_routers/0, get_all_valid_routers/0]).
-export([to_map/1]).
get_router(RouterId) when is_integer(RouterId) ->
case mnesia:dirty_read(router, RouterId) of
[Router = #router{}] ->
{ok, Router};
_ ->
undefined
end.
%% 获取app信息
-spec get_all_routers() -> {ok, Items :: list()} | {error, Reason :: any()}.
get_all_routers() ->
Fun = fun() ->
Q = qlc:q([E || E <- mnesia:table(router)]),
qlc:e(Q)
end,
case mnesia:transaction(Fun) of
{atomic, Routers} ->
{ok, Routers};
{aborted, _} ->
error
end.
%% 获取app信息
-spec get_all_valid_routers() -> {ok, Items :: list()} | {error, Reason :: any()}.
get_all_valid_routers() ->
Fun = fun() ->
Q = qlc:q([E || E <- mnesia:table(router), E#router.status =:= 1]),
qlc:e(Q)
end,
case mnesia:transaction(Fun) of
{atomic, Routers} ->
{ok, Routers};
{aborted, _} ->
error
end.
%% 获取app信息
-spec get_routers(Filter :: any(), Start :: integer(), Limit :: integer()) ->
{ok, Items :: list(), TotalNum :: integer()} |
{error, Reason :: any()}.
get_routers(Spec, Start, Limit) when is_integer(Limit), is_integer(Start), Start >= 0, Limit > 0 ->
Items0 = mnesia:dirty_select(router, [Spec]),
Items = lists:reverse(Items0),
NItems = lists:sublist(Items, Start + 1, Limit),
{ok, NItems, length(Items)}.
-spec add_router(Router :: #router{}) -> ok | {error, Reason :: any()}.
add_router(Router = #router{}) ->
case mnesia:transaction(fun() -> mnesia:write(router, Router, write) end) of
{atomic, ok} ->
ok;
{aborted, Error} ->
{error, Error}
end.
-spec change_status(RouterId :: integer(), Status :: integer()) -> ok | {error, Reason :: any()}.
change_status(RouterId, Status) when is_integer(RouterId), is_integer(Status) ->
Fun = fun() ->
case mnesia:read(router, RouterId) of
[] ->
mnesia:abort(<<"router not found">>);
[Router] ->
mnesia:write(host, Router#router{status = Status}, write)
end
end,
case mnesia:transaction(Fun) of
{atomic, ok} ->
ok;
{aborted, Reason} ->
{error, Reason}
end.
-spec delete(HostId :: integer()) -> ok | {error, Reason :: any()}.
delete(RouterId) when is_integer(RouterId) ->
case mnesia:transaction(fun() -> mnesia:delete(router, RouterId, write) end) of
{atomic, ok} ->
ok;
{aborted, Reason} ->
{error, Reason}
end.
%% 获取app表的数据大小
table_size() ->
mnesia:table_info(router, size).
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% helper methods
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
to_map(#router{router_id = RouterId, name = Name, status = Status, endpoint = Endpoint}) ->
EndpointInfo = case Endpoint of
undefined ->
#{};
#http_endpoint{url = Url} ->
#{
<<"type">> => <<"http">>,
<<"url">> => Url
};
#kafka_endpoint{} ->
#{}
end,
#{
<<"router_id">> => RouterId,
<<"name">> => Name,
<<"status">> => Status,
<<"endpoint">> => EndpointInfo
}.

97
apps/iot/src/model/scenario_deploy_model.erl Normal file
View File

@ -0,0 +1,97 @@
%%%-------------------------------------------------------------------
%%% @author licheng5
%%% @copyright (C) 2021, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 27. 4月 2021 下午4:38
%%%-------------------------------------------------------------------
-module(scenario_deploy_model).
-author("licheng5").
-include("iot.hrl").
-include_lib("stdlib/include/qlc.hrl").
-define(TAB_NAME, scenario_deploy).
%% API
-export([get_host_deploy_list/1, get_scenario_deploy_list/1, add_deploy/1, change_status/2, delete/1, table_size/0]).
-export([to_map/1]).
-spec get_host_deploy_list(HostId :: binary()) -> {ok, List :: [#scenario_deploy{}]} | error.
get_host_deploy_list(HostId) when is_binary(HostId) ->
Fun = fun() ->
Q = qlc:q([E || E <- mnesia:table(?TAB_NAME), E#scenario_deploy.host_id =:= HostId]),
qlc:e(Q)
end,
case mnesia:transaction(Fun) of
{atomic, Items} ->
{ok, Items};
{aborted, _} ->
error
end.
-spec get_scenario_deploy_list(ScenarioId :: integer()) -> {ok, List :: [#scenario_deploy{}]} | error.
get_scenario_deploy_list(ScenarioId) when is_integer(ScenarioId) ->
Fun = fun() ->
Q = qlc:q([E || E <- mnesia:table(?TAB_NAME), E#scenario_deploy.scenario_id =:= ScenarioId]),
qlc:e(Q)
end,
case mnesia:transaction(Fun) of
{atomic, Items} ->
{ok, Items};
{aborted, _} ->
error
end.
-spec add_deploy(Deploy :: #scenario_deploy{}) -> ok | {error, Reason :: any()}.
add_deploy(Deploy = #scenario_deploy{}) ->
case mnesia:transaction(fun() -> mnesia:write(?TAB_NAME, Deploy, write) end) of
{atomic, _} ->
ok;
{aborted, Error} ->
{error, Error}
end.
-spec change_status(DeployId :: integer(), Status :: integer()) -> ok | {error, Reason :: any()}.
change_status(DeployId, Status) when is_integer(DeployId), is_integer(Status) ->
Fun = fun() ->
case mnesia:read(?TAB_NAME, DeployId) of
[] ->
mnesia:abort(<<"deploy not found">>);
[Deploy] ->
mnesia:write(?TAB_NAME, Deploy#scenario_deploy{status = Status}, write)
end
end,
case mnesia:transaction(Fun) of
{atomic, ok} ->
ok;
{aborted, Reason} ->
{error, Reason}
end.
-spec delete(DeployId :: binary()) -> ok | {error, Reason :: any()}.
delete(DeployId) when is_integer(DeployId) ->
case mnesia:transaction(fun() -> mnesia:delete(?TAB_NAME, DeployId, write) end) of
{atomic, ok} ->
ok;
{aborted, Reason} ->
{error, Reason}
end.
%% 获取app表的数据大小
table_size() ->
mnesia:table_info(?TAB_NAME, size).
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% helper methods
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
to_map(#scenario_deploy{deploy_id = DeployId, scenario_id = ScenarioId, host_id = HostId, create_ts = CreateTs, update_ts = UpdateTs, status = Status}) ->
#{
<<"deploy_id">> => DeployId,
<<"scenario_id">> => ScenarioId,
<<"host_id">> => HostId,
<<"create_ts">> => CreateTs,
<<"update_ts">> => UpdateTs,
<<"status">> => Status
}.

87
apps/iot/src/model/scenario_model.erl Normal file
View File

@ -0,0 +1,87 @@
%%%-------------------------------------------------------------------
%%% @author licheng5
%%% @copyright (C) 2021, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 27. 4月 2021 下午4:38
%%%-------------------------------------------------------------------
-module(scenario_model).
-author("licheng5").
-include("iot.hrl").
-include_lib("stdlib/include/qlc.hrl").
%% API
-export([get_scenario/1, get_scenario_list/3, add_scenario/1, change_status/2, delete/1, table_size/0]).
-export([to_map/1]).
get_scenario(ScenarioId) when is_integer(ScenarioId) ->
case mnesia:dirty_read(scenario, ScenarioId) of
[Scenario = #scenario{}] ->
{ok, Scenario};
_ ->
undefined
end.
%% 获取app信息
-spec get_scenario_list(Filter :: any(), Start :: integer(), Limit :: integer()) ->
{ok, Items :: list(), TotalNum :: integer()} |
{error, Reason :: any()}.
get_scenario_list(Spec, Start, Limit) when is_integer(Limit), is_integer(Start), Start >= 0, Limit > 0 ->
Items0 = mnesia:dirty_select(scenario, [Spec]),
Items = lists:reverse(Items0),
NItems = lists:sublist(Items, Start + 1, Limit),
{ok, NItems, length(Items)}.
-spec add_scenario(Scenario :: #scenario{}) -> ok | {error, Reason :: any()}.
add_scenario(Scenario = #scenario{}) ->
case mnesia:transaction(fun() -> mnesia:write(scenario, Scenario, write) end) of
{atomic, _} ->
ok;
{aborted, Error} ->
{error, Error}
end.
-spec change_status(ScenarioId :: integer(), Status :: integer()) -> ok | {error, Reason :: any()}.
change_status(ScenarioId, Status) when is_integer(ScenarioId), is_integer(Status) ->
Fun = fun() ->
case mnesia:read(scenario, ScenarioId) of
[] ->
mnesia:abort(<<"scenario not found">>);
[Scenario] ->
mnesia:write(scenario, Scenario#scenario{status = Status}, write)
end
end,
case mnesia:transaction(Fun) of
{atomic, ok} ->
ok;
{aborted, Reason} ->
{error, Reason}
end.
-spec delete(ScenarioId :: binary()) -> ok | {error, Reason :: any()}.
delete(ScenarioId) when is_integer(ScenarioId) ->
case mnesia:transaction(fun() -> mnesia:delete(scenario, ScenarioId, write) end) of
{atomic, ok} ->
ok;
{aborted, Reason} ->
{error, Reason}
end.
%% 获取app表的数据大小
table_size() ->
mnesia:table_info(scenario, size).
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% helper methods
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
to_map(#scenario{scenario_id = ScenarioId, name = Name, desc = Desc, rule = Rule, update_ts = UpdateTs, status = Status}) ->
#{
<<"scenario_id">> => ScenarioId,
<<"name">> => Name,
<<"desc">> => Desc,
<<"rule">> => Rule,
<<"update_ts">> => UpdateTs,
<<"status">> => Status
}.

12
config/sys.config
View File

@ -5,7 +5,19 @@
{acceptors, 500},
{max_connections, 10240},
{backlog, 10240}
]},
%% 目标服务器地址
{emqx_server, [
{host, ""},
{port, 18080},
{tcp_opts, []},
{username, ""},
{password, ""},
{keepalive, 'Keepalive'},
{retry_interval, 5}
]}
]},
%% 系统日志配置，系统日志为lager, 支持日志按日期自动分割

2
docs/R.md Normal file
View File

@ -0,0 +1,2 @@
##
here

1
rebar.config
View File

@ -6,6 +6,7 @@
{cowboy, ".*", {git, "https://github.com/ninenines/cowboy.git", {tag, "2.5.0"}}},
{jiffy, ".*", {git, "https://github.com/davisp/jiffy.git", {tag, "1.1.1"}}},
{parse_trans, ".*", {git, "https://github.com/uwiger/parse_trans", {tag, "3.0.0"}}},
{emqtt, ".*", {git, "https://github.com/emqx/emqtt", {tag, "v1.2.0"}}},
{lager, ".*", {git,"https://github.com/erlang-lager/lager.git", {tag, "3.9.2"}}}
]}.

13
rebar.lock
View File

@ -8,8 +8,17 @@
{git,"https://github.com/ninenines/cowlib",
{ref,"106ba84bb04537879d8ce59321a04e0682110b91"}},
1},
{<<"emqtt">>,
{git,"https://github.com/emqx/emqtt",
{ref,"55e50041cc5b3416067c120eadb8774f1d3d1f4a"}},
0},
{<<"fs">>,{pkg,<<"fs">>,<<"6.1.1">>},1},
{<<"getopt">>,{pkg,<<"getopt">>,<<"1.0.1">>},1},
{<<"goldrush">>,{pkg,<<"goldrush">>,<<"0.1.9">>},1},
{<<"gun">>,
{git,"https://github.com/ninenines/gun",
{ref,"e7dd9f227e46979d8073e71c683395a809b78cb4"}},
1},
{<<"hackney">>,
{git,"https://github.com/benoitc/hackney.git",
{ref,"f3e9292db22c807e73f57a8422402d6b423ddf5f"}},
@ -40,13 +49,14 @@
{<<"ssl_verify_fun">>,{pkg,<<"ssl_verify_fun">>,<<"1.1.6">>},1},
{<<"sync">>,
{git,"https://github.com/rustyio/sync.git",
{ref,"3f0049e809ffe303ae2cd395217a025ce6e758ae"}},
{ref,"aa27b66ccfbfc798b57288af4cf7dc386e205d68"}},
0},
{<<"unicode_util_compat">>,{pkg,<<"unicode_util_compat">>,<<"0.5.0">>},2}]}.
[
{pkg_hash,[
{<<"certifi">>, <<"B7CFEAE9D2ED395695DD8201C57A2D019C0C43ECAF8B8BCB9320B40D6662F340">>},
{<<"fs">>, <<"9D147B944D60CFA48A349F12D06C8EE71128F610C90870BDF9A6773206452ED0">>},
{<<"getopt">>, <<"C73A9FA687B217F2FF79F68A3B637711BB1936E712B521D8CE466B29CBF7808A">>},
{<<"goldrush">>, <<"F06E5D5F1277DA5C413E84D5A2924174182FB108DABB39D5EC548B27424CD106">>},
{<<"idna">>, <<"1D038FB2E7668CE41FBF681D2C45902E52B3CB9E9C77B55334353B222C2EE50C">>},
{<<"metrics">>, <<"25F094DEA2CDA98213CECC3AEFF09E940299D950904393B2A29D191C346A8486">>},
@ -56,6 +66,7 @@
{pkg_hash_ext,[
{<<"certifi">>, <<"3B3B5F36493004AC3455966991EAF6E768CE9884693D9968055AEEEB1E575040">>},
{<<"fs">>, <<"EF94E95FFE79916860649FED80AC62B04C322B0BB70F5128144C026B4D171F8B">>},
{<<"getopt">>, <<"53E1AB83B9CEB65C9672D3E7A35B8092E9BDC9B3EE80721471A161C10C59959C">>},
{<<"goldrush">>, <<"99CB4128CFFCB3227581E5D4D803D5413FA643F4EB96523F77D9E6937D994CEB">>},
{<<"idna">>, <<"A02C8A1C4FD601215BB0B0324C8A6986749F807CE35F25449EC9E69758708122">>},
{<<"metrics">>, <<"69B09ADDDC4F74A40716AE54D140F93BEB0FB8978D8636EADED0C31B6F099F16">>},