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

fix

Browse Source
This commit is contained in:
anlicheng 2023-08-08 10:27:00 +08:00
parent 1122d679a2
commit 8a898192a9
2 changed files with 20 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
apps/iot/src/iot_endpoint.erl
View File

@ -142,19 +142,28 @@ handle_event(cast, {reload, NEndpoint = #endpoint{name = Name}}, connected, Stat
handle_event(cast, {forward, LocationCode, Fields, Timestamp}, StateName, State = #state{tab_name = TabName, window_size = WindowSize, flight_num = FlightNum, endpoint = #endpoint{name = Name, mapper_fun = MapperFun}}) -> handle_event(cast, {forward, LocationCode, Fields, Timestamp}, StateName, State = #state{tab_name = TabName, window_size = WindowSize, flight_num = FlightNum, endpoint = #endpoint{name = Name, mapper_fun = MapperFun}}) ->
try try
Body = if Result = if
is_function(MapperFun, 2) -> is_function(MapperFun, 2) ->
MapperFun(LocationCode, Fields); MapperFun(LocationCode, Fields);
is_function(MapperFun, 3) -> is_function(MapperFun, 3) ->
MapperFun(LocationCode, Fields, Timestamp) MapperFun(LocationCode, Fields, Timestamp)
end, end,
mnesia_queue:insert(TabName, #north_data{location_code = LocationCode, body = Body}), case Result of
%% 避免不必要的内部消息 {ok, Body} ->
Actions = case StateName =:= connected andalso FlightNum < WindowSize of mnesia_queue:insert(TabName, #north_data{location_code = LocationCode, body = Body}),
true -> [{next_event, info, fetch_next}]; %% 避免不必要的内部消息
false -> [] Actions = case StateName =:= connected andalso FlightNum < WindowSize of
end, true -> [{next_event, info, fetch_next}];
{keep_state, State, Actions} false -> []
end,
{keep_state, State, Actions};
{error, Error} ->
lager:debug("[iot_endpoint] forward endpoint: ~p, mapper get error: ~p", [Name, Error]),
{keep_state, State};
ignore ->
lager:debug("[iot_endpoint] forward endpoint: ~p, mapper ignore", [Name]),
{keep_state, State}
end
catch _:Reason -> catch _:Reason ->
lager:debug("[iot_endpoint] forward endpoint: ~p, mapper data get error: ~p", [Name, Reason]), lager:debug("[iot_endpoint] forward endpoint: ~p, mapper data get error: ~p", [Name, Reason]),
{keep_state, State} {keep_state, State}
@ -259,7 +268,8 @@ handle_event(EventType, Event, StateName, State) ->
%% terminate. It should be the opposite of Module:init/1 and do any %% terminate. It should be the opposite of Module:init/1 and do any
%% necessary cleaning up. When it returns, the gen_statem terminates with %% necessary cleaning up. When it returns, the gen_statem terminates with
%% Reason. The return value is ignored. %% Reason. The return value is ignored.
terminate(_Reason, _StateName, _State) -> terminate(Reason, _StateName, _State) ->
lager:debug("[iot_endpoint] terminate with reason: ~p", [Reason]),
ok. ok.
%% @private %% @private

2
apps/iot/src/postman/mqtt_postman.erl
View File

@ -100,7 +100,7 @@ handle_info({puback, Packet = #{packet_id := PacketId}}, State = #state{parent_p
end; end;
%% 转发信息 %% 转发信息
handle_info({post, ReceiverPid, #north_data{id = Id, location_code = LocationCode, body = Message}}, State = #state{parent_pid = ParentPid, conn_pid = ConnPid, inflight = InFlight, topic = Topic0, qos = Qos}) -> handle_info({post, ReceiverPid, #north_data{id = Id, location_code = LocationCode, body = Message}}, State = #state{conn_pid = ConnPid, inflight = InFlight, topic = Topic0, qos = Qos}) ->
Topic = re:replace(Topic0, <<"\\${location_code}">>, LocationCode, [global, {return, binary}]), Topic = re:replace(Topic0, <<"\\${location_code}">>, LocationCode, [global, {return, binary}]),
lager:debug("[mqtt_postman] will publish topic: ~p, message: ~p, qos: ~p", [Topic, Message, Qos]), lager:debug("[mqtt_postman] will publish topic: ~p, message: ~p, qos: ~p", [Topic, Message, Qos]),
case emqtt:publish(ConnPid, Topic, #{}, Message, [{qos, Qos}, {retain, true}]) of case emqtt:publish(ConnPid, Topic, #{}, Message, [{qos, Qos}, {retain, true}]) of