change iot_device gen_statem -> gen_server
This commit is contained in:
parent
2362c70997
commit
0d799a5bc6
@ -1,38 +1,35 @@
|
||||
%%%-------------------------------------------------------------------
|
||||
%%% @copyright (C) 2023, <COMPANY>
|
||||
%%% @author anlicheng
|
||||
%%% @copyright (C) 2024, <COMPANY>
|
||||
%%% @doc
|
||||
%% 1. 终端是否授权 => 1: 授权，0: 未授权
|
||||
%%%
|
||||
%%% @end
|
||||
%%% Created : 14. 8月 2023 11:40
|
||||
%%% Created : 11. 7月 2024 11:33
|
||||
%%%-------------------------------------------------------------------
|
||||
-module(iot_device).
|
||||
-author("aresei").
|
||||
-module(iot_device1).
|
||||
-author("anlicheng").
|
||||
-include("iot.hrl").
|
||||
|
||||
-behaviour(gen_statem).
|
||||
-behaviour(gen_server).
|
||||
|
||||
%% API
|
||||
-export([get_name/1, get_pid/1, serialize/1]).
|
||||
-export([start_link/2, is_activated/1, is_alive/1, change_status/2, reload/1, auth/2, data/2, query/5]).
|
||||
|
||||
%% gen_statem callbacks
|
||||
-export([init/1, handle_event/4, terminate/3, code_change/4, callback_mode/0]).
|
||||
%% gen_server callbacks
|
||||
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
|
||||
|
||||
%% 终端是否授权
|
||||
-define(DEVICE_AUTH_DENIED, 0).
|
||||
-define(DEVICE_AUTH_AUTHED, 1).
|
||||
|
||||
%% 存储数据的上限
|
||||
-define(MAX_SIZE, 2000).
|
||||
|
||||
%% 状态
|
||||
-define(STATE_DENIED, denied).
|
||||
-define(STATE_ACTIVATED, activated).
|
||||
|
||||
-record(state, {
|
||||
device_uuid :: binary(),
|
||||
%% 用来保存数据，作为存储在influxdb里面的数据的备份
|
||||
queue = queue:new(),
|
||||
%% 设备是否授权
|
||||
auth_status :: integer(),
|
||||
status = ?DEVICE_OFFLINE
|
||||
}).
|
||||
|
||||
@ -50,6 +47,7 @@
|
||||
% "timestamp": int
|
||||
%}
|
||||
%],
|
||||
%% 用来保存在内存中的格式，不需要序列话处理 !!!, 放入到influxdb的数据是基于base64的
|
||||
-spec serialize(FieldsList :: [map()]) -> [Val :: map()].
|
||||
serialize(FieldsList) when is_list(FieldsList) ->
|
||||
lists:flatmap(fun serialize0/1, FieldsList).
|
||||
@ -87,46 +85,47 @@ get_name(DeviceUUID) when is_binary(DeviceUUID) ->
|
||||
is_activated(undefined) ->
|
||||
false;
|
||||
is_activated(Pid) when is_pid(Pid) ->
|
||||
gen_statem:call(Pid, is_activated).
|
||||
gen_server:call(Pid, is_activated).
|
||||
|
||||
-spec change_status(Pid :: pid() | undefined, NewStatus :: integer()) -> no_return().
|
||||
change_status(undefined, _) ->
|
||||
ok;
|
||||
change_status(Pid, NewStatus) when is_pid(Pid), is_integer(NewStatus) ->
|
||||
gen_statem:cast(Pid, {change_status, NewStatus}).
|
||||
gen_server:cast(Pid, {change_status, NewStatus}).
|
||||
|
||||
-spec reload(Pid :: pid()) -> no_return().
|
||||
reload(Pid) when is_pid(Pid) ->
|
||||
gen_statem:cast(Pid, reload).
|
||||
gen_server:cast(Pid, reload).
|
||||
|
||||
-spec auth(Pid :: pid(), Auth :: boolean()) -> no_return().
|
||||
auth(Pid, Auth) when is_pid(Pid), is_boolean(Auth) ->
|
||||
gen_statem:cast(Pid, {auth, Auth}).
|
||||
gen_server:cast(Pid, {auth, Auth}).
|
||||
|
||||
-spec data(Pid :: pid(), DataList :: [#device_data{}]) -> no_return().
|
||||
data(Pid, DataList) when is_pid(Pid), is_list(DataList) ->
|
||||
gen_statem:cast(Pid, {data, DataList}).
|
||||
gen_server:cast(Pid, {data, DataList}).
|
||||
|
||||
-spec query(Pid :: pid(), Tags :: map(), StartTs :: integer(), StopTs :: integer(), Limit :: integer()) -> {ok, [#device_data{}]}.
|
||||
query(Pid, Tags, StartTs, StopTs, Limit) when is_pid(Pid), is_map(Tags), is_integer(StartTs), is_integer(StopTs), is_integer(Limit), StartTs >= 0, StopTs >= 0, Limit >= 0 ->
|
||||
gen_statem:call(Pid, {query, Tags, StartTs, StopTs, Limit}).
|
||||
gen_server:call(Pid, {query, Tags, StartTs, StopTs, Limit}).
|
||||
|
||||
%% @doc Creates a gen_statem process which calls Module:init/1 to
|
||||
%% initialize. To ensure a synchronized start-up procedure, this
|
||||
%% function does not return until Module:init/1 has returned.
|
||||
%% @doc Spawns the server and registers the local name (unique)
|
||||
-spec(start_link(Name :: atom(), DeviceUUID :: binary()) ->
|
||||
{ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
|
||||
start_link(Name, DeviceUUID) when is_atom(Name), is_binary(DeviceUUID) ->
|
||||
gen_statem:start_link({local, Name}, ?MODULE, [DeviceUUID], []);
|
||||
gen_server:start_link({local, Name}, ?MODULE, [DeviceUUID], []);
|
||||
start_link(Name, DeviceInfo) when is_atom(Name), is_map(DeviceInfo) ->
|
||||
gen_statem:start_link({local, Name}, ?MODULE, [DeviceInfo], []).
|
||||
gen_server:start_link({local, Name}, ?MODULE, [DeviceInfo], []).
|
||||
|
||||
%%%===================================================================
|
||||
%%% gen_statem callbacks
|
||||
%%% gen_server callbacks
|
||||
%%%===================================================================
|
||||
|
||||
%% @private
|
||||
%% @doc Whenever a gen_statem is started using gen_statem:start/[3,4] or
|
||||
%% gen_statem:start_link/[3,4], this function is called by the new
|
||||
%% process to initialize.
|
||||
%% @doc Initializes the server
|
||||
-spec(init(Args :: term()) ->
|
||||
{ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
|
||||
{stop, Reason :: term()} | ignore).
|
||||
init([DeviceUUID]) when is_binary(DeviceUUID) ->
|
||||
case device_bo:get_device_by_uuid(DeviceUUID) of
|
||||
{ok, DeviceInfo} ->
|
||||
@ -135,85 +134,25 @@ init([DeviceUUID]) when is_binary(DeviceUUID) ->
|
||||
lager:warning("[iot_device] device uuid: ~p, loaded from mysql failed", [DeviceUUID]),
|
||||
ignore
|
||||
end;
|
||||
init([DeviceInfo = #{<<"device_uuid">> := DeviceUUID, <<"authorize_status">> := AuthorizeStatus, <<"status">> := Status}]) when is_map(DeviceInfo) ->
|
||||
case AuthorizeStatus =:= ?DEVICE_AUTH_AUTHED of
|
||||
true ->
|
||||
{ok, ?STATE_ACTIVATED, #state{device_uuid = DeviceUUID, status = Status}};
|
||||
false ->
|
||||
{ok, ?STATE_DENIED, #state{device_uuid = DeviceUUID, status = Status}}
|
||||
end.
|
||||
init([#{<<"device_uuid">> := DeviceUUID, <<"authorize_status">> := AuthorizeStatus, <<"status">> := Status}]) ->
|
||||
{ok, #state{device_uuid = DeviceUUID, status = as_integer(Status), auth_status = as_integer(AuthorizeStatus)}}.
|
||||
|
||||
%% @private
|
||||
%% @doc This function is called by a gen_statem when it needs to find out
|
||||
%% the callback mode of the callback module.
|
||||
callback_mode() ->
|
||||
handle_event_function.
|
||||
%% @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(is_activated, _From, State = #state{auth_status = AuthStatus}) ->
|
||||
{reply, AuthStatus =:= 1, State};
|
||||
|
||||
%% @private
|
||||
%% @doc If callback_mode is handle_event_function, then whenever a
|
||||
%% gen_statem receives an event from call/2, cast/2, or as a normal
|
||||
%% process message, this function is called.
|
||||
|
||||
%% 判断是否是激活状态
|
||||
handle_event({call, From}, is_activated, StateName, State = #state{}) ->
|
||||
{keep_state, State, [{reply, From, StateName =:= ?STATE_ACTIVATED}]};
|
||||
|
||||
%% 改变为在线状态，但是数据库中的状态已经是在线状态，忽略
|
||||
handle_event(cast, {change_status, ?DEVICE_ONLINE}, _, State = #state{status = ?DEVICE_ONLINE}) ->
|
||||
{keep_state, State};
|
||||
%% 改变数据库的状态, 其他情况下执行次数都很少
|
||||
handle_event(cast, {change_status, ?DEVICE_ONLINE}, _, State = #state{device_uuid = DeviceUUID}) ->
|
||||
{ok, _} = device_bo:change_status(DeviceUUID, ?DEVICE_ONLINE),
|
||||
report_event(DeviceUUID, ?DEVICE_ONLINE),
|
||||
{keep_state, State#state{status = ?DEVICE_ONLINE}};
|
||||
|
||||
handle_event(cast, {change_status, ?DEVICE_OFFLINE}, _, State = #state{device_uuid = DeviceUUID}) ->
|
||||
{ok, #{<<"status">> := Status}} = device_bo:get_device_by_uuid(DeviceUUID),
|
||||
case Status of
|
||||
?DEVICE_NOT_JOINED ->
|
||||
lager:debug("[iot_device] device: ~p, device_maybe_offline, not joined, can not change to offline", [DeviceUUID]),
|
||||
{keep_state, State#state{status = ?DEVICE_NOT_JOINED}};
|
||||
?DEVICE_OFFLINE ->
|
||||
lager:debug("[iot_device] device: ~p, device_maybe_offline, is offline, do nothing", [DeviceUUID]),
|
||||
{keep_state, State#state{status = ?DEVICE_OFFLINE}};
|
||||
?DEVICE_ONLINE ->
|
||||
{ok, _} = device_bo:change_status(DeviceUUID, ?DEVICE_OFFLINE),
|
||||
report_event(DeviceUUID, ?DEVICE_OFFLINE),
|
||||
{keep_state, State#state{status = ?DEVICE_OFFLINE}}
|
||||
end;
|
||||
|
||||
%% 重新加载数据库数据
|
||||
handle_event(cast, reload, _, State = #state{device_uuid = DeviceUUID}) ->
|
||||
lager:debug("[iot_device] will reload: ~p", [DeviceUUID]),
|
||||
case device_bo:get_device_by_uuid(DeviceUUID) of
|
||||
{ok, #{<<"authorize_status">> := AuthorizeStatus, <<"status">> := Status}} ->
|
||||
case AuthorizeStatus =:= ?DEVICE_AUTH_AUTHED of
|
||||
true ->
|
||||
{next_state, ?STATE_ACTIVATED, State#state{status = Status}};
|
||||
false ->
|
||||
{next_state, ?STATE_DENIED, State#state{status = Status}}
|
||||
end;
|
||||
undefined ->
|
||||
lager:warning("[iot_device] device uuid: ~p, loaded from mysql failed", [DeviceUUID]),
|
||||
{stop, normal, State}
|
||||
end;
|
||||
|
||||
%% 向设备中追加数据
|
||||
handle_event(cast, {data, DataList}, _, State = #state{queue = Q}) ->
|
||||
NQ = lists:foldl(fun(Data, Q0) ->
|
||||
case queue:len(Q0) > ?MAX_SIZE of
|
||||
true ->
|
||||
{_, Q1} = queue:out(Q0),
|
||||
queue:in(Data, Q1);
|
||||
false ->
|
||||
queue:in(Data, Q0)
|
||||
end
|
||||
end, Q, DataList),
|
||||
|
||||
{keep_state, State#state{queue = NQ}};
|
||||
|
||||
%% 查询device里面的数据
|
||||
handle_event({call, From}, {query, Tags, StartTs, StopTs, Limit}, _StateName, State = #state{queue = Q}) ->
|
||||
%% 查询当前设备中产生的数据，缓存在内存中
|
||||
handle_call({query, Tags, StartTs, StopTs, Limit}, _From, State = #state{queue = Q}) ->
|
||||
L = queue:to_list(Q),
|
||||
|
||||
%% 过滤
|
||||
@ -228,38 +167,94 @@ handle_event({call, From}, {query, Tags, StartTs, StopTs, Limit}, _StateName, St
|
||||
false ->
|
||||
lists:sublist(L3, 1, Limit)
|
||||
end,
|
||||
{keep_state, State, [{reply, From, {ok, L4}}]};
|
||||
|
||||
%% 处理授权
|
||||
handle_event(cast, {auth, Auth}, StateName, State = #state{device_uuid = DeviceUUID}) ->
|
||||
case {StateName, Auth} of
|
||||
{?STATE_DENIED, false} ->
|
||||
lager:debug("[iot_device] device_uuid: ~p, auth: false, will keep state_name: ~p", [DeviceUUID, ?STATE_DENIED]),
|
||||
{keep_state, State};
|
||||
{?STATE_DENIED, true} ->
|
||||
{next_state, ?STATE_ACTIVATED, State};
|
||||
|
||||
{?STATE_ACTIVATED, false} ->
|
||||
lager:debug("[iot_device] device_uuid: ~p, auth: false, state_name from: ~p, to: ~p", [DeviceUUID, ?STATE_ACTIVATED, ?STATE_DENIED]),
|
||||
{next_state, ?STATE_DENIED, State};
|
||||
{?STATE_ACTIVATED, true} ->
|
||||
lager:debug("[iot_device] device_uuid: ~p, auth: true, will keep state_name: ~p", [DeviceUUID, ?STATE_ACTIVATED]),
|
||||
{keep_state, State}
|
||||
end.
|
||||
{reply, {ok, L4}, State}.
|
||||
|
||||
%% @private
|
||||
%% @doc This function is called by a gen_statem when it is about to
|
||||
%% @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({change_status, ?DEVICE_ONLINE}, State = #state{status = ?DEVICE_ONLINE}) ->
|
||||
{noreply, State};
|
||||
handle_cast({change_status, ?DEVICE_ONLINE}, State = #state{device_uuid = DeviceUUID}) ->
|
||||
{ok, _} = device_bo:change_status(DeviceUUID, ?DEVICE_ONLINE),
|
||||
report_event(DeviceUUID, ?DEVICE_ONLINE),
|
||||
{noreply, State#state{status = ?DEVICE_ONLINE}};
|
||||
handle_cast({change_status, ?DEVICE_OFFLINE}, State = #state{device_uuid = DeviceUUID}) ->
|
||||
{ok, #{<<"status">> := Status}} = device_bo:get_device_by_uuid(DeviceUUID),
|
||||
case Status of
|
||||
?DEVICE_NOT_JOINED ->
|
||||
lager:debug("[iot_device] device: ~p, device_maybe_offline, not joined, can not change to offline", [DeviceUUID]),
|
||||
{noreply, State#state{status = ?DEVICE_NOT_JOINED}};
|
||||
?DEVICE_OFFLINE ->
|
||||
lager:debug("[iot_device] device: ~p, device_maybe_offline, is offline, do nothing", [DeviceUUID]),
|
||||
{noreply, State#state{status = ?DEVICE_OFFLINE}};
|
||||
?DEVICE_ONLINE ->
|
||||
{ok, _} = device_bo:change_status(DeviceUUID, ?DEVICE_OFFLINE),
|
||||
report_event(DeviceUUID, ?DEVICE_OFFLINE),
|
||||
{noreply, State#state{status = ?DEVICE_OFFLINE}}
|
||||
end;
|
||||
|
||||
%% 重新加载数据库数据
|
||||
handle_cast(reload, State = #state{device_uuid = DeviceUUID}) ->
|
||||
lager:debug("[iot_device] will reload: ~p", [DeviceUUID]),
|
||||
case device_bo:get_device_by_uuid(DeviceUUID) of
|
||||
{ok, #{<<"authorize_status">> := AuthorizeStatus, <<"status">> := Status}} ->
|
||||
{noreply, State#state{status = as_integer(Status), auth_status = as_integer(AuthorizeStatus)}};
|
||||
undefined ->
|
||||
lager:warning("[iot_device] device uuid: ~p, loaded from mysql failed", [DeviceUUID]),
|
||||
{stop, normal, State}
|
||||
end;
|
||||
|
||||
%% 向设备中追加数据
|
||||
handle_cast({data, DataList}, State = #state{queue = Q}) ->
|
||||
NQ = lists:foldl(fun(Data, Q0) ->
|
||||
case queue:len(Q0) > ?MAX_SIZE of
|
||||
true ->
|
||||
{_, Q1} = queue:out(Q0),
|
||||
queue:in(Data, Q1);
|
||||
false ->
|
||||
queue:in(Data, Q0)
|
||||
end
|
||||
end, Q, DataList),
|
||||
{noreply, State#state{queue = NQ}};
|
||||
|
||||
%% 处理授权
|
||||
handle_cast({auth, true}, State = #state{device_uuid = DeviceUUID}) ->
|
||||
lager:debug("[iot_device] device_uuid: ~p, auth: true", [DeviceUUID]),
|
||||
{noreply, State#state{auth_status = 1}};
|
||||
handle_cast({auth, false}, State = #state{device_uuid = DeviceUUID}) ->
|
||||
lager:debug("[iot_device] device_uuid: ~p, auth: false", [DeviceUUID]),
|
||||
{noreply, State#state{auth_status = 0}}.
|
||||
|
||||
%% @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_statem terminates with
|
||||
%% Reason. The return value is ignored.
|
||||
terminate(Reason, StateName, #state{device_uuid = DeviceUUID}) ->
|
||||
lager:notice("[iot_device] device_uuid: ~p, state_name: ~p, terminate with reason: ~p", [DeviceUUID, StateName, Reason]),
|
||||
%% 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{device_uuid = DeviceUUID}) ->
|
||||
lager:notice("[iot_device] device_uuid: ~p, terminate with reason: ~p", [DeviceUUID, Reason]),
|
||||
ok.
|
||||
|
||||
%% @private
|
||||
%% @doc Convert process state when code is changed
|
||||
code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
|
||||
{ok, StateName, State}.
|
||||
-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
|
||||
@ -306,3 +301,9 @@ filter_tags(L, Tags) when map_size(Tags) =:= 0 ->
|
||||
lists:all(fun({TagName, TagVal}) -> maps:is_key(TagName, Tags0) andalso maps:get(TagName, Tags0) =:= TagVal end, maps:to_list(Tags))
|
||||
end, L).
|
||||
|
||||
%% 转换成整数，从数据读取的数据有时候不一定都是整数
|
||||
-spec as_integer(Val :: integer() | binary()) -> integer().
|
||||
as_integer(Val) when is_integer(Val) ->
|
||||
Val;
|
||||
as_integer(Val) when is_binary(Val) ->
|
||||
binary_to_integer(Val).
|
||||
Loading…
x
Reference in New Issue
Block a user