feat: channel log

apps/dgiot_bridge/src/dgiot_bridge.erl

@@ -26,7 +26,7 @@
 -export([start/0, start_channel/2, start_channel/3, register_channel/2, get_behaviour/1, do_global_message/1]).
 -export([get_product_info/1, get_products/1, get_acl/1, apply_channel/5, apply_product/3, parse_frame/3, to_frame/3]).
 -export([get_data/2, send_log/3, send_log/4, send_log/5]).
--export([get_all_channel/0, control_channel/2, list/0, get_proctol_channel/1, control_uniapp/2, uniapp_report/1]).
+-export([get_all_channel/0, control_channel/3, list/0, get_proctol_channel/1, control_uniapp/2, uniapp_report/1]).
 
 init_ets() ->
     dgiot_data:init(?DGIOT_BRIDGE),
@@ -332,7 +332,7 @@ list() ->
         end,
     lists:sort(dgiot_plugin:check_module(Fun, [])).
 
-control_channel(ChannelId, Action) ->
+control_channel(ChannelId, Action, SessionToken) ->
     {IsEnable, Result} =
         case Action of
             <<"disable">> ->
@@ -365,11 +365,13 @@ control_channel(ChannelId, Action) ->
                 dgiot_mqtt:publish(ChannelId, Topic, Payload),
                 {true, <<"success">>};
             <<"start_logger">> ->
+                dgiot_mqtt:subscribe_route_key([<<"$dg/user/channel/", ChannelId/binary, "/#">>], SessionToken),
                 Topic = <<"channel/", ChannelId/binary>>,
                 Payload = jsx:encode(#{<<"channelId">> => ChannelId, <<"action">> => <<"start_logger">>}),
                 dgiot_mqtt:publish(ChannelId, Topic, Payload),
                 {true, <<"success">>};
             <<"stop_logger">> ->
+                dgiot_mqtt:unsubscribe_route_key(SessionToken),
                 Topic = <<"channel/", ChannelId/binary>>,
                 Payload = jsx:encode(#{<<"channelId">> => ChannelId, <<"action">> => <<"stop_logger">>}),
                 dgiot_mqtt:publish(ChannelId, Topic, Payload),

apps/dgiot_bridge/src/handler/dgiot_bridge_handler.erl

@@ -80,8 +80,7 @@ handle(OperationID, Args, Context, Req) ->
 %%%===================================================================
 do_request(post_control_channel, #{<<"id">> := ChannelId, <<"action">> := Action}, #{<<"sessionToken">> := SessionToken} = _Context, _Req)
     when Action == <<"enable">>; Action == <<"disable">>; Action == <<"update">>; Action == <<"start_logger">>; Action == <<"stop_logger">> ->
-    dgiot_mqtt:subscribe_route_key([<<"$dg/user/channel/", ChannelId/binary, "/#">>], SessionToken),
-    dgiot_bridge:control_channel(ChannelId, Action);
+    dgiot_bridge:control_channel(ChannelId, Action, SessionToken);
 
 %% Decoder 概要: 获取指令集 描述:根据产品ID关联的解码器获取指令集
 %% OperationId:get_cmd_productid

apps/dgiot_device/src/dgiot_device_channel.erl

@@ -275,7 +275,7 @@ handle_message({sync_parse, _Pid, 'before', delete, _Token, <<"Channel">>, Objec
 %%    io:format("~s ~p ~p ~n", [?FILE, ?LINE, ObjectId]),
     case dgiot_parse:get_object(<<"Channel">>, ObjectId) of
         {ok, #{<<"isEnable">> := true}} ->
-            dgiot_bridge:control_channel(ObjectId, <<"disable">>);
+            dgiot_bridge:control_channel(ObjectId, <<"disable">>, <<>>);
         _ -> pass
     end,
     {ok, State};

apps/dgiot_device/src/utils/dgiot_product_csv.erl

@@ -33,7 +33,7 @@ read_csv(ChannelId, FilePath) ->
     timer:sleep(1000),
     dgiot_product_csv:post_thing(FileName, ProductIds),
     timer:sleep(1000),
-    dgiot_bridge:control_channel(TdChannelId, <<"update">>),
+    dgiot_bridge:control_channel(TdChannelId, <<"update">>, <<>>),
     get_max_addrs(FileName).
 
 %% dgiot_product_csv:read_from_csv(<<"/dgiot_file/product/csv/modbustcp.csv">>)
@@ -120,7 +120,7 @@ create_device(FileName, Devicemap, ProductIds) ->
                 {ok, #{<<"ACL">> := Acl}} ->
                     Acl;
                 _ ->
-                    #{<<114,111,108,101,58,229,188,128,229,143,145,232,128,133>> => #{<<"read">> => true, <<"write">> => true}}
+                    #{<<114, 111, 108, 101, 58, 229, 188, 128, 229, 143, 145, 232, 128, 133>> => #{<<"read">> => true, <<"write">> => true}}
             end,
         lists:foldl(fun(Devaddr, _Acc1) ->
             dgiot_device:create_device(#{

apps/dgiot_dlink/src/handler/dgiot_dlink_handler.erl

@@ -177,19 +177,8 @@ do_request(get_dlinkjson, #{<<"type">> := Type}, _Context, _Req) ->
     DlinkJson = dgiot_dlink:get_json(Type),
     {200, DlinkJson};
 
-do_request(post_topic, #{<<"topic">> := Topic} = _Args, #{<<"sessionToken">> := _SessionToken} = _Context, _Req) ->
-%%    TopicKey =
-%%        case is_list(Topic) of
-%%            true ->
-%%                <<"dev_states">>;
-%%            false ->
-%%                [_Head, _User, Key | _] = re:split(Topic, "/"),
-%%                <<"dg_user_", Key/binary>>
-%%        end,
-%%    PubTopic = <<"dgiot_topics/", SessionToken/binary>>,
-%%    Payload = jsx:encode(#{<<"topic">> => Topic, <<"topickey">> => TopicKey}),
-%%    dgiot_mqtt:publish(self(), PubTopic, Payload),
-%%    timer:sleep(100),
+do_request(post_topic, #{<<"topic">> := Topic} = _Args, #{<<"sessionToken">> := SessionToken} = _Context, _Req) ->
+    dgiot_mqtt:subscribe_route_key([Topic], SessionToken),
     {200, #{<<"message">> => <<"订阅成功"/utf8>>, <<"Topic">> => Topic, <<"TopicKey">> => <<"TopicKey">>}};
 
 do_request(get_thingecho, _Args, _Context, _Req) ->

apps/dgiot_meter/src/dgiot_meter_tcp.erl

@@ -113,7 +113,7 @@ handle_info({tcp, Buff}, #tcp{socket = Socket, state = #state{id = ChannelId, dt
             case DtuAddr of
                 <<>> ->
                     {noreply, TCPState#tcp{buff = <<>>}};
-                _->
+                _ ->
                     dgiot_meter:create_dtu(DtuAddr, ChannelId, DTUIP),
                     {DtuProductId, _, _} = dgiot_data:get({dtu, ChannelId}),
                     %%                    $dg/device/{productId}/{deviceAddr}/profile