feat: add tcp connect info

CHANGELOG.md

@@ -7,6 +7,7 @@
 * channel_log ([1e37058](https://github.com/dgiot/dgiot/commit/1e3705876eb905cab215085e7341e050ce822170))
 * fix mqtt trace bug ([1a53ebe](https://github.com/dgiot/dgiot/commit/1a53ebec4ecf68ad2e19c14d43ed47fbcbed1754))
 * post trace failed ([544eed5](https://github.com/dgiot/dgiot/commit/544eed54b1798fe799f885f5af99a06582fa115b))
+* publish topic error ([b3ab5fc](https://github.com/dgiot/dgiot/commit/b3ab5fcc4ac52467a192a3ea685d1a155b314445))
 * task_worker bug ([0396163](https://github.com/dgiot/dgiot/commit/039616312126d24e36b03cf339e0e6ffb6f3aa13))
 
 

apps/dgiot/src/transport/dgiot_tcp_server.erl

@@ -45,7 +45,6 @@ child_spec(Mod, Port, State, Opts) ->
     MFArgs = {?MODULE, start_link, [Mod, NewOpts, State]},
     esockd:child_spec(Name, Port, TCPOpts, MFArgs).
 
-
 start_link(Transport, Sock, Mod, Opts, State) ->
     {ok, proc_lib:spawn_link(?MODULE, init, [Mod, Transport, Opts, Sock, State])}.
 
@@ -62,7 +61,7 @@ init(Mod, Transport, Opts, Sock0, State) ->
                         rate_limit = rate_limit(proplists:get_value(rate_limit, Opts)),
                         child = NewChildState
                     },
-                    dgiot_metrics:inc(dgiot_bridge,<<"tcp_server">>,1),
+                    dgiot_metrics:inc(dgiot_bridge, <<"tcp_server">>, 1),
                     ok = activate_socket(GState),
                     gen_server:enter_loop(?MODULE, [], GState);
                 {error, Reason} ->
@@ -115,6 +114,9 @@ handle_info({tcp, Sock, Data}, #state{mod = Mod, child = #tcp{register = false,
     case Mod:handle_info({tcp, <<Buff/binary, NewBin/binary>>}, NewChildState) of
         {noreply, #tcp{register = true, clientid = ClientId, buff = Buff, socket = Sock} = NewChild} ->
             dgiot_cm:register_channel(ClientId, self(), #{conn_mod => Mod}),
+            Ip = dgiot_utils:get_ip(Sock),
+            Port = dgiot_utils:get_port(Sock),
+            dgiot_cm:insert_channel_info(ClientId,#{ip => Ip, port => Port,online => dgiot_datetime:now_microsecs()},[{tcp_recv, 1}]),
             {noreply, State#state{child = NewChild, incoming_bytes = Cnt}, hibernate};
         {noreply,  NewChild} ->
             {noreply, State#state{child = NewChild, incoming_bytes = Cnt}, hibernate};
@@ -145,7 +147,7 @@ handle_info({tcp, Sock, Data}, #state{mod = Mod, child = #tcp{buff = Buff, socke
 
 handle_info({shutdown, Reason}, #state{child = #tcp{clientid = CliendId, register = true} = ChildState} = State) ->
     ?LOG(error,"shutdown, ~p, ~p~n", [Reason, ChildState#tcp.state]),
-    dgiot_cm:unregister_channel(CliendId),
+    dgiot_cm:connection_closed(CliendId),
     write_log(ChildState#tcp.log, <<"ERROR">>, list_to_binary(io_lib:format("~w", [Reason]))),
     {stop, normal, State#state{child = ChildState#tcp{socket = undefined}}};
 
@@ -179,8 +181,8 @@ handle_info(Info, #state{mod = Mod, child = ChildState} = State) ->
     end.
 
 terminate(Reason, #state{mod = Mod, child = #tcp{clientid = CliendId, register = true} = ChildState}) ->
-    dgiot_cm:unregister_channel(CliendId),
-    dgiot_metrics:dec(dgiot_bridge,<<"tcp_server">>,1),
+    dgiot_cm:connection_closed(CliendId),
+    dgiot_metrics:dec(dgiot_bridge,<<"tcp_server">>, 1),
     Mod:terminate(Reason, ChildState);
 
 terminate(Reason, #state{mod = Mod, child = ChildState}) ->

apps/dgiot/src/utils/dgiot_utils.erl

@@ -95,6 +95,7 @@
     , random/0
     , get_hostname/0
     , get_ip/1
+    , get_port/1
     , get_natip/0
     , get_wlanip/0
     , get_computerconfig/0
@@ -757,6 +758,10 @@ get_ip(Socket) ->
         to_list(D),
     to_binary(Ip).
 
+get_port(Socket) ->
+    {ok, {{_A, _B, _C, _D}, Port}} = esockd_transport:peername(Socket),
+    Port.
+
 %%re:run(os:cmd("chcp 65001 & arp -a"),
 %%<<"([\\d]{1,3}\\.[\\d]{1,3}\\.[\\d]{1,3}\\.[\\d]{1,3}).*?([\\S]{2}-[\\S]{2}-[\\S]{2}-[\\S]{2}-[\\S]{2}-[\\S]{2})">>,
 %%[global, {capture, all_but_first, binary}])

apps/dgiot_device/src/dgiot_product.erl

@@ -58,17 +58,21 @@ handle_message({load, ProductId, IsLoadDevice}, State) ->
     Reply =
         case dgiot_product:get(ProductId) of
             {ok, Product} ->
-                {ok, Product1} = dgiot_product:save(Product),
-                case IsLoadDevice of
-                    true ->
-                        case dgiot_product:synchronize_device(ProductId) of
-                            ok ->
-                                {ok, Product1};
-                            {error, Why} ->
-                                {error, Why}
+                case dgiot_product:save(Product) of
+                    {ok, Product1} ->
+                        case IsLoadDevice of
+                            true ->
+                                case dgiot_product:synchronize_device(ProductId) of
+                                    ok ->
+                                        {ok, Product1};
+                                    {error, Why} ->
+                                        {error, Why}
+                                end;
+                            false ->
+                                {ok, Product1}
                         end;
-                    false ->
-                        {ok, Product1}
+                    _ ->
+                        pass
                 end;
             {error, Reason} ->
                 {error, Reason}