fix endpoint

apps/iot/include/endpoint.hrl

@@ -24,10 +24,8 @@
 }).
 
 -record(kafka_endpoint, {
-    enable_sasl = false :: boolean(),
+    % {mechanism :: atom(), username = <<>> :: binary(), password = <<>> :: binary()}
-    username = <<>> :: binary(),
+    sasl_config :: undefined | tuple(),
-    password = <<>> :: binary(),
-    mechanism :: atom(),
     bootstrap_servers = [] :: [{string(), integer()}],
     topic = <<>> :: binary()
 }).

apps/iot/src/database/endpoint_bo.erl

@@ -60,15 +60,7 @@ parse_config(<<"http">>, #{<<"url">> := Url, <<"pool_size">> := PoolSize}) ->
         url = Url,
         pool_size = PoolSize
     };
-parse_config(<<"kafka">>, #{<<"enable_sasl">> := EnableSasl, <<"username">> := Username, <<"password">> := Password, <<"mechanism">> := Mechanism0, <<"bootstrap_servers">> := BootstrapServers0, <<"topic">> := Topic}) ->
+parse_config(<<"kafka">>, #{<<"sasl_config">> := #{<<"username">> := Username, <<"password">> := Password, <<"mechanism">> := Mechanism0}, <<"bootstrap_servers">> := BootstrapServers, <<"topic">> := Topic}) ->
-    BootstrapServers = lists:filtermap(fun(S) ->
-        case binary:split(S, <<":">>) of
-            [Host0, Port0] ->
-                {true, {binary_to_list(Host0), binary_to_integer(Port0)}};
-            _ ->
-                false
-        end
-                                       end, BootstrapServers0),
     Mechanism = case Mechanism0 of
                     <<"sha_256">> ->
                         scram_sha_256;
@@ -81,12 +73,25 @@ parse_config(<<"kafka">>, #{<<"enable_sasl">> := EnableSasl, <<"username">> := U
                 end,
 
     #kafka_endpoint{
-        enable_sasl = EnableSasl,
+        sasl_config = {Mechanism, Username, Password},
-        username = Username,
+        bootstrap_servers = parse_bootstrap_servers(BootstrapServers),
-        password = Password,
+        topic = Topic
-        mechanism = Mechanism,
+    };
-        bootstrap_servers = BootstrapServers,
+parse_config(<<"kafka">>, #{<<"bootstrap_servers">> := BootstrapServers, <<"topic">> := Topic}) ->
+    #kafka_endpoint{
+        sasl_config = undefined,
+        bootstrap_servers = parse_bootstrap_servers(BootstrapServers),
         topic = Topic
     };
 parse_config(_, _) ->
-    throw(invalid_config).
+    throw(invalid_config).
+
+parse_bootstrap_servers(BootstrapServers) when is_list(BootstrapServers) ->
+    lists:filtermap(fun(S) ->
+        case binary:split(S, <<":">>) of
+            [Host0, Port0] ->
+                {true, {binary_to_list(Host0), binary_to_integer(Port0)}};
+            _ ->
+                false
+        end
+                    end, BootstrapServers).

apps/iot/src/endpoint/endpoint_kafka.erl

@@ -92,23 +92,20 @@ handle_cast({forward, ServiceId, Metric}, State = #state{buffer = Buffer}) ->
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
 handle_info({timeout, _, connect}, State = #state{buffer = Buffer, status = ?DISCONNECTED, client_id = ClientId,
-    endpoint = #endpoint{title = Title, config = #kafka_endpoint{enable_sasl = EnableSasl, username = Username, password = Password, mechanism = Mechanism,  bootstrap_servers = BootstrapServers, topic = Topic}}}) ->
+    endpoint = #endpoint{title = Title, config = #kafka_endpoint{sasl_config = SaslConfig, bootstrap_servers = BootstrapServers, topic = Topic}}}) ->
     lager:debug("[endpoint_kafka] endpoint: ~p, create postman", [Title]),
 
-    ClientConfig0 = [
+    BaseConfig = [
         {reconnect_cool_down_seconds, 5},
-        {socket_options, [
+        {socket_options, [{keepalive, true}]}
-            {keepalive, true}
-        ]}
     ],
 
-    ClientConfig = case EnableSasl of
+    ClientConfig = case SaslConfig of
-                       true ->
+                     {Mechanism, Username, Password} ->
-                           SaslConfig = {sasl, {Mechanism, Username, Password}},
+                         [{sasl, {Mechanism, Username, Password}}|BaseConfig];
-                           [SaslConfig|ClientConfig0];
+                     undefined ->
-                       false ->
+                         BaseConfig
-                           ClientConfig0
+                 end,
-                   end,
 
     case brod:start_link_client(BootstrapServers, ClientId, ClientConfig) of
         {ok, ClientPid} ->

docs/router.md

@@ -1,3 +0,0 @@
-# 数据路由设计文档
-
-

docs/zhongdian_mqtt.md

@@ -1,58 +0,0 @@
-# 中电mqtt通讯规约 
-## 服务器地址
-    MQTT服务器IP：172.30.6.161
-    MQTT服务器端口：1883
-    MQTT服务器账号：admin
-    MQTT服务器密码：public
-
-## topic
-    南向：
-    MQTT ClientID: CET/NX
-    发布Topic:CET/NX/upload
-    订阅Topic:CET/NX/downlod
-
-    中电：
-    MQTT ClientID:CET/NX
-    订阅Topic:CET/NX/upload
-    发布Topic:CET/NX/download
-
-## 数据格式
-```text
-{
-  "version": "1.0",
-  "location_code": "string",
-  "ts ": 1688606685,
-  "properties": [
-    {
-      "type": "AI",
-      "key": "A相电流",
-      "value": 0.25,
-      "unit": "A",
-      "timestamp": 1688354258
-    },
-    {
-      "type": "AI",
-      "key": "A相电压",
-      "value": 220.5,
-      "unit": "V",
-      "timestamp": 1688354258
-    },
-    {
-      "type": "SOE",
-      "key": "电压越限",
-      "value": 1,
-      "unit": "V",
-      "timestamp": 1688354258
-    }
-  ]
-}
-
-南向上送数据
-
-CET应答：
-{
-    "location_code": string, //(点位编码信息),
-    "ts ":1688606685 ,
-    "result":1(int)
-}
-```