diff --git a/apps/iot/src/endpoint/iot_jinzhi_endpoint.erl b/apps/iot/src/endpoint/iot_jinzhi_endpoint.erl index c2609a6..674ac37 100644 --- a/apps/iot/src/endpoint/iot_jinzhi_endpoint.erl +++ b/apps/iot/src/endpoint/iot_jinzhi_endpoint.erl @@ -175,37 +175,34 @@ do_post(Url, Id, Body) when is_list(Url), is_integer(Id), is_binary(Body) -> %% 格式话要发送的数据,避免多次格式化处理 -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 - }, - +format_events(LocationCode, DynamicLocationCode, Events, PriKey) when is_binary(LocationCode), is_binary(DynamicLocationCode), is_list(Events) -> + DeviceInfo = format_event(LocationCode, hd(Events)), ReqData = #{ <<"sign">> => sign(DeviceInfo, PriKey), <<"sysId">> => ?SYS_ID, + <<"taskId">> => <<"">>, + <<"count">> => length(Events), <<"deviceInfo">> => DeviceInfo }, iolist_to_binary(jiffy:encode(ReqData, [force_utf8])). +format_event(LocationCode, {_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), + + % <<"occurrenceTime">> => <<"2023-06-10 12:00:00">>, + + #{ + <<"location">> => LocationCode, + <<"category">> => EventCode, + <<"description">> => Description, + <<"occurrenceTime">> => Datetime, + <<"attachments">> => Attachments + }. + -spec generate_private_key(PriFile :: string()) -> public_key:private_key(). generate_private_key(PriFile) when is_list(PriFile) -> PriKeyFile = code:priv_dir(iot) ++ "/" ++ PriFile, diff --git a/apps/iot/src/iot_event_publisher.erl b/apps/iot/src/iot_event_publisher.erl index 4e19835..0c4a777 100644 --- a/apps/iot/src/iot_event_publisher.erl +++ b/apps/iot/src/iot_event_publisher.erl @@ -19,16 +19,9 @@ -define(SERVER, ?MODULE). -%% 单个任务 --record(task, { - counter = 0, - %% 缓存周期内的数据 - buffer = [] -}). - -record(state, { device_uuid :: binary(), - group_tasks = #{} :: map() + group_buffers = #{} :: map() }). %%%=================================================================== @@ -77,20 +70,19 @@ 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_tasks = GroupTasks}) -> +handle_cast({publish, EventType, Params}, State = #state{device_uuid = DeviceUUID, group_buffers = GroupBuffers}) -> GroupKey = group_by(EventType, Params), - 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)}}; + case maps:find(GroupKey, GroupBuffers) of + {ok, Buffer} -> + NBuffer = [{EventType, Params}|Buffer], + {noreply, State#state{group_buffers = maps:put(GroupKey, NBuffer, GroupBuffers)}}; error -> ThrottleTime = iot_event_period_settings:get_throttle(GroupKey), erlang:start_timer(ThrottleTime * 1000, self(), {throttle_ticker, GroupKey}), %% 发送消息 iot_ai_router:batch_route_uuid(DeviceUUID, [{EventType, Params}]), - Task = #task{buffer = [], counter = 1}, - {noreply, State#state{group_tasks = maps:put(GroupKey, Task, GroupTasks)}} + {noreply, State#state{group_buffers = maps:put(GroupKey, [], GroupBuffers)}} end. %% @private @@ -99,21 +91,19 @@ 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, GroupKey}}, State = #state{device_uuid = DeviceUUID, group_tasks = GroupTasks}) -> +handle_info({timeout, _, {throttle_ticker, GroupKey}}, State = #state{device_uuid = DeviceUUID, group_buffers = GroupBuffers}) -> ThrottleTime = iot_event_period_settings:get_throttle(GroupKey), erlang:start_timer(ThrottleTime * 1000, self(), {throttle_ticker, GroupKey}), - case maps:find(GroupKey, GroupTasks) of - {ok, Task0 = #task{buffer = Buffer}} -> + case maps:find(GroupKey, GroupBuffers) of + {ok, Buffer} -> case length(Buffer) > 0 of true -> - Events = lists:reverse(Buffer), - iot_ai_router:batch_route_uuid(DeviceUUID, Events); + iot_ai_router:batch_route_uuid(DeviceUUID, Buffer); false -> ok end, - Task = Task0#task{buffer = []}, - {noreply, State#state{group_tasks = maps:put(GroupKey, Task, GroupTasks)}}; + {noreply, State#state{group_buffers = maps:put(GroupKey, [], GroupBuffers)}}; error -> {noreply, State} end.