fix Maintenance bug

apps/dgiot/src/channel/dgiot_channelx.erl

@@ -28,7 +28,7 @@
     spec/0, spec/1,
     delete/2, delete/3,
     status/1,
-    do_event/4, do_event/5,
+    do_event/3, do_event/4,  do_event/5,
     do_message/2, do_message/3, do_message/4,
     call/3, call/4, call2/3, call2/4,
     start_link/1]).
@@ -84,6 +84,14 @@ status(Id) ->
 
 name(Id) -> list_to_atom(lists:concat([?MODULE, "_", Id])).
 
+do_event(ChannelId, EventId, Event) ->
+    case dgiot_data:get({channeltype, ChannelId}) of
+        not_find ->
+            not_find;
+        ChannelType ->
+            do_event(ChannelType, ChannelId, EventId, Event, 5000)
+    end.
+
 do_event(ChannelType, ChannelId, EventId, Event) ->
     do_event(ChannelType, ChannelId, EventId, Event, 5000).
 do_event(ChannelType, ChannelId, EventId, Event, Timeout) ->
@@ -92,6 +100,7 @@ do_event(ChannelType, ChannelId, EventId, Event, Timeout) ->
         fun(Worker) ->
             Worker ! {event, Pool, EventId, Event}, ok
         end,
+    ?LOG(error, "EventId ~p", [EventId]),
     poolboy:transaction(Pool, Fun, Timeout).
 
 do_message(ChannelId, Message) ->

apps/dgiot_bridge/src/dgiot_bridge_actions.erl

@@ -151,12 +151,29 @@ on_action_create_dgiot(_Id, Params = #{
     ?LOG(error, " msg topic: ~p, payload: ~p", [TopicTks, PayloadTks]),
     Params.
 
+%% mqtt事件
+%%[ 'client.connected'
+%%, 'client.disconnected'
+%%, 'session.subscribed'
+%%, 'session.unsubscribed'
+%%, 'message.publish'
+%%, 'message.delivered'
+%%, 'message.acked'
+%%, 'message.dropped'
+%%]
 -spec on_action_dgiot(selected_data(), env_vars()) -> any().
-on_action_dgiot(Selected, Envs) ->
+on_action_dgiot(Selected, #{event := Event} = Envs) ->
     ChannelId = dgiot_mqtt:get_channel(Envs),
     Msg = dgiot_mqtt:get_message(Selected, Envs),
-    case dgiot_channelx:do_message(ChannelId, {rule, Msg, Selected}) of
-        not_find -> dgiot_mqtt:republish(Selected, Envs);
-        _ -> pass
+    case Event of
+        'message.publish' ->
+            case dgiot_channelx:do_message(ChannelId, {rule, Msg, Selected}) of
+                not_find -> dgiot_mqtt:republish(Selected, Envs);
+                _ -> pass
+            end;
+        EventId ->
+            dgiot_channelx:do_event(ChannelId, EventId, {rule, Msg, Selected})
     end.
 
+
+

apps/dgiot_niisten/src/dgiot_niisten_channel.erl

@@ -166,7 +166,8 @@ handle_init(State) ->
     {ok, State}.
 
 %% 通道消息处理,注意：进程池调用
-handle_event(_EventId, _Event, State) ->
+handle_event(EventId, Event, State) ->
+    ?LOG(error,"EventId ~p Event ~p",[EventId, Event] ),
     {ok, State}.
 
 

apps/dgiot_parse/src/dgiot_parse.erl

@@ -160,7 +160,7 @@ get_objectid(Class, Map) ->
         <<"post_classes_maintenance">> ->
             get_objectid(<<"Maintenance">>, Map);
         <<"Maintenance">> ->
-            Deviceid = maps:get(<<"device">>, Map, <<"">>),
+            #{<<"objectId">> := Deviceid} = maps:get(<<"device">>, Map, <<"">>),
             Number = maps:get(<<"number">>, Map, <<"">>),
             <<Pid:10/binary, _/binary>> = dgiot_utils:to_md5(<<"Maintenance", Deviceid/binary, Number/binary>>),
             Map#{