From 0d799a5bc67b4cc4c0829cbe8151a620670f858e Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Thu, 11 Jul 2024 14:28:37 +0800 Subject: [PATCH] change iot_device gen_statem -> gen_server --- apps/iot/src/iot_device.erl | 261 ++++++++++++++++++------------------ 1 file changed, 131 insertions(+), 130 deletions(-) diff --git a/apps/iot/src/iot_device.erl b/apps/iot/src/iot_device.erl index f6ed8d6..0ef05f6 100644 --- a/apps/iot/src/iot_device.erl +++ b/apps/iot/src/iot_device.erl @@ -1,38 +1,35 @@ %%%------------------------------------------------------------------- -%%% @copyright (C) 2023, +%%% @author anlicheng +%%% @copyright (C) 2024, %%% @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). \ No newline at end of file