diff --git a/apps/iot/src/endpoint/iot_jinzhi_endpoint.erl b/apps/iot/src/endpoint/iot_jinzhi_endpoint.erl index e11fd00..5f5fc8c 100644 --- a/apps/iot/src/endpoint/iot_jinzhi_endpoint.erl +++ b/apps/iot/src/endpoint/iot_jinzhi_endpoint.erl @@ -1,41 +1,45 @@ %%%------------------------------------------------------------------- -%%% @author aresei -%%% @copyright (C) 2023, +%%% @author anlicheng +%%% @copyright (C) 2024, %%% @doc %%% %%% @end -%%% Created : 06. 7月 2023 12:02 +%%% Created : 23. 7月 2024 14:51 %%%------------------------------------------------------------------- -module(iot_jinzhi_endpoint). --author("aresei"). --include("iot.hrl"). +-author("anlicheng"). --behaviour(gen_statem). +-behaviour(gen_server). %% API -export([start_link/0]). --export([get_pid/0, forward/4, get_stat/0]). -%% gen_statem callbacks --export([init/1, handle_event/4, terminate/3, code_change/4, callback_mode/0]). +-export([get_pid/0, batch_forward/3]). + +%% gen_server callbacks +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). + +-define(SERVER, ?MODULE). %% 消息重发间隔 -define(RETRY_INTERVAL, 5000). %% 系统id -define(SYS_ID, <<"ZNWLZJJKXT">>). +-record(task, { + id :: integer(), + location_code :: binary(), + dynamic_location_code :: binary(), + events =[] :: list() +}). + -record(state, { + id = 1, url :: string(), logger_pid :: pid(), - pool_size = 0, - flight_num = 0, pri_key :: public_key:private_key(), - id = 1, - queue :: queue:queue(), %% 定时器对应关系 - timer_map = #{}, - %% 记录成功处理的消息数 - acc_num = 0 + timer_map = #{} }). %%%=================================================================== @@ -46,166 +50,116 @@ get_pid() -> whereis(?MODULE). --spec forward(LocationCode :: binary(), DynamicLocationCode :: binary(), EventType :: integer(), Params :: map()) -> no_return(). -forward(LocationCode, DynamicLocationCode, EventType, Params) when is_binary(LocationCode), is_binary(DynamicLocationCode), is_integer(EventType), is_map(Params) -> - gen_statem:cast(?MODULE, {forward, LocationCode, DynamicLocationCode, EventType, Params}). +-spec batch_forward(LocationCode :: binary(), DynamicLocationCode :: binary(), Events :: list()) -> no_return(). +batch_forward(LocationCode, DynamicLocationCode, Events) when is_binary(LocationCode), is_binary(DynamicLocationCode), is_list(Events) -> + gen_server:cast(?MODULE, {batch_forward, LocationCode, DynamicLocationCode, Events}). --spec get_stat() -> {ok, Stat :: #{}}. -get_stat() -> - gen_statem:call(?MODULE, get_stat, 5000). - -%% @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() -> + {ok, Pid :: pid()} | ignore | {error, Reason :: term()}). start_link() -> - gen_statem:start_link({local, ?MODULE}, ?MODULE, [], []). + gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). %%%=================================================================== -%%% 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([]) -> {ok, Opts} = application:get_env(iot, jinzhi), - PoolSize = proplists:get_value(pool_size, Opts), PriFile = proplists:get_value(pri_key, Opts), Url = proplists:get_value(url, Opts), {ok, LoggerPid} = iot_logger:start_link("ai_event_data"), PriKey = generate_private_key(PriFile), - {ok, connected, #state{url = Url, logger_pid = LoggerPid, pri_key = PriKey, pool_size = PoolSize, queue = queue:new()}}. + {ok, #state{url = Url, logger_pid = LoggerPid, pri_key = PriKey}}. %% @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(_Request, _From, State = #state{}) -> + {reply, ok, State}. %% @private -%% @doc There should be one instance of this function for each possible -%% state name. If callback_mode is state_functions, one of these -%% functions is called when gen_statem receives and event from -%% call/2, cast/2, or as a normal process message. +%% @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({batch_forward, LocationCode, DynamicLocationCode, Events}, State = #state{id = Id, timer_map = TimerMap, pri_key = PriKey, url = Url}) -> + ReqBody = format_events(LocationCode, DynamicLocationCode, Events, PriKey), + catch do_post(Url, Id, ReqBody), + TimerRef = erlang:start_timer(?RETRY_INTERVAL, self(), {repost_ticker, Id, ReqBody}), -handle_event(cast, {forward, LocationCode, DynamicLocationCode, EventType, Params}, _, State = #state{id = Id, flight_num = FlightNum, pool_size = PoolSize, queue = Q}) -> - EventData = #event_data{id = Id, location_code = LocationCode, dynamic_location_code = DynamicLocationCode, event_type = EventType, params = Params}, - %% 避免不必要的内部消息 - Actions = case FlightNum < PoolSize of - true -> [{next_event, info, fetch_next}]; - false -> [] - end, - {keep_state, State#state{queue = queue:in(EventData, Q), id = Id + 1}, Actions}; + {noreply, State#state{id = Id + 1, timer_map = maps:put(Id, TimerRef, TimerMap)}}. -%% 触发读取下一条数据 -handle_event(info, fetch_next, _, State = #state{queue = Q, flight_num = FlightNum, timer_map = TimerMap}) -> - case queue:out(Q) of - {{value, EventData = #event_data{id = Id}}, Q1} -> - lager:debug("[iot_http_endpoint] fetch_next success, event data is: ~p", [EventData]), - catch do_post(EventData, State), - TimerRef = erlang:start_timer(?RETRY_INTERVAL, self(), {repost_ticker, EventData}), - - {keep_state, State#state{timer_map = maps:put(Id, TimerRef, TimerMap), queue = Q1, flight_num = FlightNum + 1}}; - {empty, Q1} -> - {keep_state, State#state{queue = Q1}} - end; - -%% 收到确认消息 -handle_event(info, {ack, Id}, _, State = #state{timer_map = TimerMap, acc_num = AccNum, flight_num = FlightNum}) -> +%% @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({ack, Id, Body, RespBody}, State = #state{timer_map = TimerMap, logger_pid = LoggerPid}) -> + %% 记录日志 + iot_logger:write(LoggerPid, [Body, RespBody]), case maps:take(Id, TimerMap) of error -> - {keep_state, State#state{acc_num = AccNum + 1, flight_num = FlightNum - 1}, [{next_event, info, fetch_next}]}; + {noreply, State}; {TimerRef, NTimerMap} -> is_reference(TimerRef) andalso erlang:cancel_timer(TimerRef), - {keep_state, State#state{timer_map = NTimerMap, acc_num = AccNum + 1, flight_num = FlightNum - 1}, [{next_event, info, fetch_next}]} + {noreply, State#state{timer_map = NTimerMap}} end; -%% 收到重发过期请求 -handle_event(info, {timeout, _, {repost_ticker, EventData = #event_data{id = Id}}}, _, State = #state{timer_map = TimerMap}) -> - lager:debug("[iot_jinzhi_endpoint] repost data: ~p", [EventData]), - catch do_post(EventData, State), - TimerRef = erlang:start_timer(?RETRY_INTERVAL, self(), {repost_ticker, EventData}), +handle_info({timeout, _, {repost_ticker, Id, ReqBody}}, State = #state{timer_map = TimerMap, url = Url}) -> + lager:debug("[iot_jinzhi_endpoint] repost data: ~p", [ReqBody]), + catch do_post(Url, Id, ReqBody), + TimerRef = erlang:start_timer(?RETRY_INTERVAL, self(), {repost_ticker, Id, ReqBody}), {keep_state, State#state{timer_map = maps:put(Id, TimerRef, TimerMap)}}; -%% Task进程挂掉 -handle_event(info, {'DOWN', _MRef, process, _Pid, normal}, _, State) -> - {keep_state, State}; +handle_info({'DOWN', _MRef, process, _Pid, normal}, State) -> + {noreply, State}; -handle_event(info, {'DOWN', _MRef, process, _Pid, Reason}, _, State) -> +handle_info({'DOWN', _MRef, process, _Pid, Reason}, State) -> lager:notice("[iot_jinzhi_endpoint] task process down with reason: ~p", [Reason]), - {keep_state, State}; - -%% 获取当前统计信息 -handle_event({call, From}, get_stat, StateName, State = #state{acc_num = AccNum}) -> - Stat = #{ - <<"acc_num">> => AccNum, - <<"queue_num">> => mnesia_queue:table_size(), - <<"state_name">> => atom_to_binary(StateName) - }, - {keep_state, State, [{reply, From, Stat}]}; + {noreply, 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(EventType, Event, StateName, State) -> - lager:warning("[iot_jinzhi_endpoint] unknown message, event_type: ~p, event: ~p, state_name: ~p, state: ~p", [EventType, Event, StateName, State]), - {keep_state, State}. - -%% @private -%% @doc This function is called by a gen_statem when it is about to +%% @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{}) -> - lager:debug("[iot_jinzhi_endpoint] terminate with reason: ~p", [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 = #state{}) -> 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 %%%=================================================================== - --spec do_post(EventData :: #event_data{}, State :: #state{}) -> no_return(). -do_post(#event_data{id = Id, location_code = LocationCode, dynamic_location_code = DynamicLocationCode, event_type = EventType, - params = Params = #{<<"event_code">> := EventCode, <<"description">> := Description, <<"datetime">> := Datetime, <<"attachments">> := Attachments0}}, - #state{pri_key = PriKey, url = Url, logger_pid = LoggerPid}) -> - - % <<"occurrenceTime">> => <<"2023-06-10 12:00:00">>, - - Attachments = lists:map(fun(#{<<"filename">> := Filename}) -> - {ok, FileUrl} = iot_util:file_uri(Filename), - Name = filename:basename(FileUrl), - #{<<"name">> => Name, <<"url">> => FileUrl} - end, Attachments0), - - DeviceInfo = #{ - <<"location">> => LocationCode, - <<"dynamic_location">> => DynamicLocationCode, - <<"category">> => EventCode, - <<"description">> => Description, - <<"occurrenceTime">> => Datetime, - <<"attachments">> => Attachments - }, - - ReqData = #{ - <<"sign">> => sign(DeviceInfo, PriKey), - <<"sysId">> => ?SYS_ID, - <<"deviceInfo">> => DeviceInfo - }, - Body = iolist_to_binary(jiffy:encode(ReqData, [force_utf8])), - - lager:debug("[iot_jinzhi_endpoint] do_post url: ~p, event_type: ~p, params: ~p, location_code: ~p, body: ~p", [Url, EventType, Params, LocationCode, Body]), - +-spec do_post(Url :: string(), Id :: integer(), Body :: binary()) -> no_return(). +do_post(Url, Id, Body) when is_list(Url), is_integer(Id), is_binary(Body) -> ReceiverPid = self(), %% 异步提交任务 spawn_monitor(fun() -> @@ -216,8 +170,7 @@ do_post(#event_data{id = Id, location_code = LocationCode, dynamic_location_code {ok, 200, _, ClientRef} -> {ok, RespBody} = hackney:body(ClientRef), hackney:close(ClientRef), - iot_logger:write(LoggerPid, [Body, RespBody]), - ReceiverPid ! {ack, Id}; + ReceiverPid ! {ack, Id, Body, RespBody}; {ok, HttpCode, _, ClientRef} -> {ok, RespBody} = hackney:body(ClientRef), hackney:close(ClientRef), @@ -227,6 +180,39 @@ do_post(#event_data{id = Id, location_code = LocationCode, dynamic_location_code end end). +%% 格式话要发送的数据，避免多次格式化处理 +-spec format_events(LocationCode :: binary(), DynamicLocationCode :: binary(), Events :: list(), PriKey :: public_key:private_key()) -> binary(). +format_events(LocationCode, DynamicLocationCode, Events, PriKey) -> + % <<"occurrenceTime">> => <<"2023-06-10 12:00:00">>, + %% 格式转换 + TextEvents = lists:map(fun({_EventType, #{<<"event_code">> := EventCode, <<"description">> := Description, <<"datetime">> := Datetime, <<"attachments">> := Attachments0}}) -> + Attachments = lists:map(fun(#{<<"filename">> := Filename}) -> + {ok, FileUrl} = iot_util:file_uri(Filename), + Name = filename:basename(FileUrl), + #{<<"name">> => Name, <<"url">> => FileUrl} + end, Attachments0), + + #{ + <<"category">> => EventCode, + <<"description">> => Description, + <<"occurrenceTime">> => Datetime, + <<"attachments">> => Attachments + } + end, Events), + + DeviceInfo = #{ + <<"location">> => LocationCode, + <<"dynamic_location">> => DynamicLocationCode, + <<"events">> => TextEvents + }, + + ReqData = #{ + <<"sign">> => sign(DeviceInfo, PriKey), + <<"sysId">> => ?SYS_ID, + <<"deviceInfo">> => DeviceInfo + }, + iolist_to_binary(jiffy:encode(ReqData, [force_utf8])). + -spec generate_private_key(PriFile :: string()) -> public_key:private_key(). generate_private_key(PriFile) when is_list(PriFile) -> PriKeyFile = code:priv_dir(iot) ++ "/" ++ PriFile, @@ -271,4 +257,4 @@ serialize([{K, V}|T], Target) -> <<$[, V0/binary, $]>> end, Item = <<$", K/binary, $", $:, V1/binary>>, - serialize(T, [Item|Target]). + serialize(T, [Item|Target]). \ No newline at end of file diff --git a/apps/iot/src/iot_ai_router.erl b/apps/iot/src/iot_ai_router.erl index 42075e1..2bcbdd1 100644 --- a/apps/iot/src/iot_ai_router.erl +++ b/apps/iot/src/iot_ai_router.erl @@ -23,4 +23,16 @@ route_uuid(RouterUUID, EventType, Params) when is_binary(RouterUUID), is_integer lager:debug("[iot_ai_router] the event_data hget location_code, uuid: ~p, not found", [RouterUUID]); {error, Reason} -> lager:debug("[iot_ai_router] the event_data hget location_code uuid: ~p, get error: ~p", [RouterUUID, Reason]) + end. + +-spec batch_route_uuid(RouterUUID :: binary(), Events :: list()) -> no_return(). +batch_route_uuid(RouterUUID, Events) when is_binary(RouterUUID), is_list(Events) -> + %% 查找终端设备对应的点位信息 + case redis_client:hgetall(RouterUUID) of + {ok, #{<<"location_code">> := LocationCode, <<"dynamic_location_code">> := DynamicLocationCode}} when is_binary(LocationCode), is_binary(DynamicLocationCode) -> + iot_jinzhi_endpoint:batch_forward(LocationCode, DynamicLocationCode, Events); + {ok, _} -> + lager:debug("[iot_ai_router] the event_data hget location_code, uuid: ~p, not found", [RouterUUID]); + {error, Reason} -> + lager:debug("[iot_ai_router] the event_data hget location_code uuid: ~p, get error: ~p", [RouterUUID, Reason]) end. \ No newline at end of file diff --git a/apps/iot/src/iot_device.erl b/apps/iot/src/iot_device.erl index 95d5374..4269228 100644 --- a/apps/iot/src/iot_device.erl +++ b/apps/iot/src/iot_device.erl @@ -32,7 +32,7 @@ auth_status :: integer(), status = ?DEVICE_OFFLINE, %% 事件分组 - ai_groups = #{} + publisher_pid :: pid() }). %%%=================================================================== @@ -141,7 +141,8 @@ init([DeviceUUID]) when is_binary(DeviceUUID) -> ignore end; init([#{<<"device_uuid">> := DeviceUUID, <<"authorize_status">> := AuthorizeStatus, <<"status">> := Status}]) -> - {ok, #state{device_uuid = DeviceUUID, status = as_integer(Status), auth_status = as_integer(AuthorizeStatus)}}. + {ok, PublisherPid} = iot_event_publisher:start_link(DeviceUUID), + {ok, #state{device_uuid = DeviceUUID, publisher_pid = PublisherPid, status = as_integer(Status), auth_status = as_integer(AuthorizeStatus)}}. %% @private %% @doc Handling call messages @@ -235,17 +236,9 @@ handle_cast({auth, false}, State = #state{device_uuid = DeviceUUID}) -> {noreply, State#state{auth_status = 0}}; %% ai事件的延迟整流逻辑 -handle_cast({ai_event, EventType, Params}, State = #state{ai_groups = Groups, device_uuid = DeviceUUID}) -> - GroupKey = group_by(EventType, Params), - case maps:find(GroupKey, Groups) of - {ok, PublisherPid} -> - iot_event_publisher:publish(PublisherPid, EventType, Params), - {noreply, State}; - error -> - {ok, PublisherPid} = iot_event_publisher:start_link(DeviceUUID, GroupKey), - iot_event_publisher:publish(PublisherPid, EventType, Params), - {noreply, State#state{ai_groups = maps:put(GroupKey, PublisherPid, Groups)}} - end. +handle_cast({ai_event, EventType, Params}, State = #state{publisher_pid = PublisherPid}) -> + iot_event_publisher:publish(PublisherPid, EventType, Params), + {noreply, State}. %% @private %% @doc Handling all non call/cast messages @@ -325,8 +318,4 @@ filter_tags(L, Tags) when map_size(Tags) =:= 0 -> as_integer(Val) when is_integer(Val) -> Val; as_integer(Val) when is_binary(Val) -> - binary_to_integer(Val). - -%% 事件分组函数 -group_by(EventType, _Params) -> - EventType. \ No newline at end of file + binary_to_integer(Val). \ No newline at end of file diff --git a/apps/iot/src/iot_event_publisher.erl b/apps/iot/src/iot_event_publisher.erl index 0b65c4c..ba41522 100644 --- a/apps/iot/src/iot_event_publisher.erl +++ b/apps/iot/src/iot_event_publisher.erl @@ -12,22 +12,23 @@ -behaviour(gen_server). %% API --export([start_link/2, publish/3, set_throttle/2]). +-export([start_link/1, publish/3]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -define(SERVER, ?MODULE). +%% 单个任务 +-record(task, { + counter = 0, + %% 缓存周期内的数据 + buffer = [] +}). + -record(state, { device_uuid :: binary(), - timer_ref = undefined :: undefined | reference(), - group_key :: any(), - %% 已经发送的事件的计数器 - counter = 0, - last_event = undefined :: any(), - %% 最后一次发送数据的时间戳, 如果数据跨越了多个时间跨度时也需要立即发送出去 - last_timestamp = 0 :: integer() + group_tasks = #{} :: map() }). %%%=================================================================== @@ -38,15 +39,11 @@ publish(Pid, EventType, Params) when is_pid(Pid), is_integer(EventType), is_map(Params) -> gen_server:cast(Pid, {publish, EventType, Params}). --spec set_throttle(Pid :: pid(), ThrottleTime :: integer()) -> no_return(). -set_throttle(Pid, ThrottleTime) when is_pid(Pid), is_integer(ThrottleTime), ThrottleTime > 0 -> - gen_server:cast(Pid, {set_throttle, ThrottleTime}). - %% @doc Spawns the server and registers the local name (unique) --spec(start_link(DeviceUUID :: binary(), GroupKey :: any()) -> +-spec(start_link(DeviceUUID :: binary()) -> {ok, Pid :: pid()} | ignore | {error, Reason :: term()}). -start_link(DeviceUUID, GroupKey) when is_binary(DeviceUUID) -> - gen_server:start_link(?MODULE, [DeviceUUID, GroupKey], []). +start_link(DeviceUUID) when is_binary(DeviceUUID) -> + gen_server:start_link(?MODULE, [DeviceUUID], []). %%%=================================================================== %%% gen_server callbacks @@ -57,8 +54,8 @@ start_link(DeviceUUID, GroupKey) when is_binary(DeviceUUID) -> -spec(init(Args :: term()) -> {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} | {stop, Reason :: term()} | ignore). -init([DeviceUUID, GroupKey]) -> - {ok, #state{device_uuid = DeviceUUID, group_key = GroupKey, last_timestamp = iot_util:timestamp_of_seconds(), last_event = undefined}}. +init([DeviceUUID]) -> + {ok, #state{device_uuid = DeviceUUID}}. %% @private %% @doc Handling call messages @@ -80,29 +77,21 @@ handle_call(_Request, _From, State = #state{}) -> {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), NewState :: #state{}}). %% 第一次收到消息，则立即发送; 并且重置定时器 -handle_cast({publish, EventType, Params}, State = #state{device_uuid = DeviceUUID, group_key = GroupKey, counter = 0}) -> - %% 重置定时器 - ThrottleTime = iot_event_period_settings:get_throttle(GroupKey), - TimerRef = erlang:start_timer(ThrottleTime * 1000, self(), throttle_ticker), - %% 发送消息 - iot_ai_router:route_uuid(DeviceUUID, EventType, Params), - {noreply, State#state{counter = 1, timer_ref = TimerRef, last_timestamp = iot_util:timestamp_of_seconds()}}; +handle_cast({publish, EventType, Params}, State = #state{device_uuid = DeviceUUID, group_tasks = GroupTasks}) -> + GroupKey = group_by(EventType, Params), -%% 缓冲数据，等到时间到了再发送 -handle_cast({publish, EventType, Params}, State = #state{device_uuid = DeviceUUID, timer_ref = TimerRef, counter = Counter, last_timestamp = LastTimestamp, group_key = GroupKey}) -> - ThrottleTime = iot_event_period_settings:get_throttle(GroupKey), - Timestamp = iot_util:timestamp_of_seconds(), - %% 如果数据发送间隔已经超过了一个时间跨度，也需要立即发送 - case Timestamp > LastTimestamp + ThrottleTime of - true -> + case maps:find(GroupKey, GroupTasks) of + {ok, Task0 = #task{counter = Counter, buffer = Buffer}} -> + Task = Task0#task{counter = Counter + 1, buffer = [{EventType, Params}|Buffer]}, + {noreply, State#state{group_tasks = maps:put(GroupKey, Task, GroupTasks)}}; + error -> %% 重置定时器 - is_reference(TimerRef) andalso erlang:cancel_timer(TimerRef), - NTimerRef = erlang:start_timer(ThrottleTime * 1000, self(), throttle_ticker), + ThrottleTime = iot_event_period_settings:get_throttle(GroupKey), + erlang:start_timer(ThrottleTime * 1000, self(), {throttle_ticker, GroupKey}), %% 发送消息 iot_ai_router:route_uuid(DeviceUUID, EventType, Params), - {noreply, State#state{counter = Counter + 1, timer_ref = NTimerRef, last_timestamp = iot_util:timestamp_of_seconds()}}; - false -> - {noreply, State#state{last_event = {EventType, Params}}} + Task = #task{buffer = [], counter = 1}, + {noreply, State#state{group_tasks = maps:put(GroupKey, Task, GroupTasks)}} end. %% @private @@ -111,18 +100,21 @@ handle_cast({publish, EventType, Params}, State = #state{device_uuid = DeviceUUI {noreply, NewState :: #state{}} | {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), NewState :: #state{}}). -%% 周期时间内没有任何数据 -handle_info({timeout, _, throttle_ticker}, State = #state{group_key = GroupKey, last_event = undefined}) -> - ThrottleTime = iot_event_period_settings:get_throttle(GroupKey), - TimerRef = erlang:start_timer(ThrottleTime * 1000, self(), throttle_ticker), - {noreply, State#state{timer_ref = TimerRef}}; -%% 已经发送的数据也需要重置，避免推送重复的数据 -handle_info({timeout, _, throttle_ticker}, State = #state{device_uuid = DeviceUUID, group_key = GroupKey, last_event = {EventType, Params}}) -> +handle_info({timeout, _, {throttle_ticker, GroupKey}}, State = #state{device_uuid = DeviceUUID, group_tasks = GroupTasks}) -> ThrottleTime = iot_event_period_settings:get_throttle(GroupKey), - TimerRef = erlang:start_timer(ThrottleTime * 1000, self(), throttle_ticker), - iot_ai_router:route_uuid(DeviceUUID, EventType, Params), - {noreply, State#state{timer_ref = TimerRef, last_event = undefined, last_timestamp = iot_util:timestamp_of_seconds()}}. + erlang:start_timer(ThrottleTime * 1000, self(), throttle_ticker), + + case maps:find(GroupKey, GroupTasks) of + {ok, Task0 = #task{buffer = Buffer}} -> + Events = lists:reverse(Buffer), + iot_ai_router:route_uuid(DeviceUUID, Events), + + Task = Task0#task{buffer = []}, + {noreply, State#state{group_tasks = maps:put(GroupKey, Task, GroupTasks)}}; + error -> + {noreply, State} + end. %% @private %% @doc This function is called by a gen_server when it is about to @@ -145,3 +137,7 @@ code_change(_OldVsn, State = #state{}, _Extra) -> %%%=================================================================== %%% Internal functions %%%=================================================================== + +%% 事件分组函数 +group_by(EventType, _Params) -> + EventType. \ No newline at end of file