feat: mqttssl

apps/dgiot_http/src/aliyun/dgiot_aliyun_auth.erl

@@ -225,7 +225,7 @@ jwtlogin(Idtoken) ->
                         <<"sex">> => "男"
                     },
                     <<"jwt">> => TokenData}},
-            SessionToken = dgiot_parse_handler:get_token(<<228, 186, 167, 228, 184, 154, 229, 164, 167, 232, 132, 145, 231, 148, 168, 230, 136, 183>>),
+            _SessionToken = dgiot_parse_handler:get_token(<<228, 186, 167, 228, 184, 154, 229, 164, 167, 232, 132, 145, 231, 148, 168, 230, 136, 183>>),
             case dgiot_parse:query_object(<<"_User">>, #{<<"where">> => #{<<"username">> => Username}}) of
                 {ok, #{<<"results">> := Results}} when length(Results) == 0 ->
                     case dgiot_parse:get_object(<<"_Role">>, <<"f897518198">>) of
@@ -249,13 +249,13 @@ jwtlogin(Idtoken) ->
                     dgiot_parse:update_object(<<"_User">>, UserId, #{<<"tag">> => Tag#{<<"jwt">> => TokenData}})
             end,
             UserInfo =
-                case dgiot_parse:login(Username, UdAccountUuid) of
+                case dgiot_parse_handler:login_by_account(Username, UdAccountUuid) of
                     {ok, #{<<"objectId">> := _UserId} = UserInfo1} ->
                         UserInfo1;
                     {error, _Msg} ->
                         #{}
                 end,
-            {ok, #{<<"code">> => 200, <<"username">> => Username, <<"userinfo">> => UserInfo, <<"state">> => TokenData, <<"msg">> => <<"operation success">>, <<"sessionToken">> => SessionToken}};
+            {ok, UserInfo#{<<"code">> => 200, <<"username">> => Username, <<"state">> => TokenData, <<"msg">> => <<"operation success">>}};
         _Error ->
             {ok, #{<<"code">> => 500, <<"msg">> => <<"operation error">>}}
     end.

apps/dgiot_mqtt/src/dgiot_mqtt_channel.erl

@@ -82,13 +82,16 @@ init(?TYPE, ChannelId, #{
                 dgiot_data:insert({mqttd, ProductId}, {Acl, maps:get(<<"properties">>, Thing, [])}),
 %%              创建连接规则
                 ConRawsql = <<"SELECT clientid, connected_at FROM \"$events/client_connected\" WHERE username = '", ProductId/binary, "'">>,
-                create_rules(<<"rule:connected_", ProductId/binary>>, ChannelId, <<"创建连接规则"/utf8>>, ConRawsql),
+                create_rules(<<"rule:connected_", ProductId/binary>>, ChannelId, <<"创建连接规则"/utf8>>, ConRawsql, <<"/${productid}/#">>),
 %%              创建断开连接规则
                 DisRawsql = <<"SELECT clientid, disconnected_at FROM \"$events/client_disconnected\" WHERE username = '", ProductId/binary, "'">>,
-                create_rules(<<"rule:disconnected_", ProductId/binary>>, ChannelId, <<"断开连接规则"/utf8>>, DisRawsql),
+                create_rules(<<"rule:disconnected_", ProductId/binary>>, ChannelId, <<"断开连接规则"/utf8>>, DisRawsql, <<"/${productid}/#">>),
 %%              创建上传数据规则
                 RepRawsql = <<"SELECT payload.msg as msg,clientid,'", ProductId/binary, "' as productid FROM \"/thing/", ProductId/binary, "/#\" WHERE username = '", ProductId/binary, "'">>,
-                create_rules(<<"rule:thingreport_", ProductId/binary>>, ChannelId, <<"派生物模型上报规则"/utf8>>, RepRawsql);
+                create_rules(<<"rule:thingreport_", ProductId/binary>>, ChannelId, <<"物模型上报规则"/utf8>>, RepRawsql, <<"/thing/${productid}/#">>),
+                %%              创建上传数据规则
+                MetaRawsql = <<"SELECT payload.msg as msg,clientid,'", ProductId/binary, "' as productid FROM \"/", ProductId/binary, "/#\" WHERE username = '", ProductId/binary, "'">>,
+                create_rules(<<"rule:metadata_", ProductId/binary>>, ChannelId, <<"派生物模型上报规则"/utf8>>, MetaRawsql, <<"/${productid}/#">>);
             _ ->
                 io:format("~s ~p X = ~p.~n", [?FILE, ?LINE, X]),
                 pass
@@ -157,6 +160,21 @@ handle_message({rule, #{clientid := _DeviceId, username := ProductId, payload :=
                             io:format("~s ~p error: ~p~n", [?FILE, ?LINE, _Other]),
                             pass
                     end;
+                [<<>>, ProductId, DtuAddr, <<"topo">>, _GroupName, <<"post">>] ->
+                    create_device(ProductId, DtuAddr, <<"MATLAB_", DtuAddr/binary>>, Peerhost),
+                    case jsx:decode(Payload, [{labels, binary}, return_maps]) of
+                        #{<<"timestamp">> := _Timestamp,
+                            <<"deviceAddr">> := _DeviceAddr,
+                            <<"properties">> := Properties} when is_map(Properties) ->
+                            io:format("~s ~p Metadata = ~p.~n", [?FILE, ?LINE, Properties]),
+                            NewProperties = get_properties(ProductId, Properties),
+                            io:format("~s ~p NewProperties: ~p~n", [?FILE, ?LINE, NewProperties]),
+                            dgiot_tdengine_adapter:save(ProductId, DtuAddr, NewProperties);
+                        _Other1 ->
+                            io:format("~s ~p error: ~p~n", [?FILE, ?LINE, _Other1]),
+                            pass
+                    end;
+
 %%%%                扫描子设备: /{productId}/{deviceAddr}/scan/{protocol}
 %%                [<<>>, ProductId, DtuAddr, <<"scan">>, <<"OPC_DA">>] ->
 %%                    case jsx:decode(Payload, [{labels, binary}, return_maps]) of
@@ -224,7 +242,7 @@ create_device(ProductId, DtuAddr, Name, DTUIP) ->
     },
     dgiot_device:create_device(Requests).
 
-create_rules(RuleID, ChannelId, Description, Rawsql) ->
+create_rules(RuleID, ChannelId, Description, Rawsql, Target_topic) ->
     emqx_rule_engine_api:create_resource(#{},
         [
             {<<"id">>, <<"resource:", ChannelId/binary>>},
@@ -239,7 +257,7 @@ create_rules(RuleID, ChannelId, Description, Rawsql) ->
                 <<"channel">> => <<"数蛙物联网通道"/utf8>>,
                 <<"payload_tmpl">> => <<"${payload}">>,
                 <<"target_qos">> => 0,
-                <<"target_topic">> => <<"/thing/${productid}/#">>
+                <<"target_topic">> => Target_topic
             }}],
         <<"enabled">> => true,
         <<"ctx">> => #{
@@ -267,3 +285,27 @@ create_rules(RuleID, ChannelId, Description, Rawsql) ->
                 _ -> pass
             end
     end.
+
+
+get_properties(ProductId, Properties) ->
+    case dgiot_product:lookup_prod(ProductId) of
+        {ok, #{<<"thing">> := #{<<"properties">> := Props}}} ->
+            lists:foldl(fun(X, Acc) ->
+                case X of
+                    #{<<"identifier">> := Identifier,
+                        <<"dataType">> := #{<<"type">> := _Type, <<"das">> := Das}} ->
+                        maps:fold(fun(PK, PV, _Acc3) ->
+                            case lists:member(PK, Das) of
+                                true ->
+                                    Acc#{Identifier => PV};
+                                _ ->
+                                    Acc
+                            end
+                                  end, Acc, Properties);
+                    _ ->
+                        Acc
+                end
+                        end, #{}, Props);
+        _Error ->
+            Properties
+    end.

apps/dgiot_parse/src/dgiot_parse.erl

@@ -932,7 +932,7 @@ import(Name, Class, [Data | Other], Count, Requests, Fun, Acc) when length(Reque
         NewRequests = [#{
             <<"method">> => <<"POST">>,
             <<"path">> => <<"classes/", Class/binary>>,
-            <<"body">> => format_data(Class, Data)
+            <<"body">> => Data
         } | Requests],
         import(Name, Class, Other, Count, NewRequests, Fun, Acc)
     catch

apps/dgiot_topo/src/dgiot_topo.erl

@@ -76,18 +76,18 @@ get_topo(Arg, _Context) ->
     Type = maps:get(<<"type">>, Arg, <<"web">>),
     case ViewId of
         undefined ->
-            case dgiot_parse:query_object(<<"View">>, #{<<"where">> => #{<<"key">> => ProductId, <<"type">> => <<"topo">>, <<"class">> => <<"Product">>}}) of
-                {ok, #{<<"results">> := Views}} ->
+            case dgiot_parse:query_object(<<"View">>, #{<<"limit">> => 1, <<"where">> => #{<<"key">> => ProductId, <<"type">> => <<"topo">>, <<"class">> => <<"Product">>}}) of
+                {ok, #{<<"results">> := Views}} when length(Views) > 0 ->
                     NewStage =
                         lists:foldl(fun(View, Acc) ->
                             case View of
-                                #{<<"data">> := #{<<"konva">> := #{<<"Stage">> := #{<<"children">> := Children}}}} when length(Children) > 0 ->
+                                #{<<"objectId">> := _ViewId1, <<"data">> := #{<<"konva">> := #{<<"Stage">> := #{<<"children">> := Children}}}} when length(Children) > 0 ->
                                     NewView = get_view(View, Devaddr, ProductId, Type),
-                                    Acc ++ [NewView];
+                                    NewView;
                                 _ ->
                                     Acc
                             end
-                                    end, [], Views),
+                                    end, #{}, Views),
                     {ok, #{<<"code">> => 200, <<"message">> => <<"SUCCESS">>, <<"data">> => NewStage}};
                 _ ->
                     {ok, #{<<"code">> => 204, <<"message">> => <<"没有组态"/utf8>>}}

apps/dgiot_topo/src/dgiot_topo_channel.erl

@@ -148,7 +148,7 @@ handle_message({sync_parse, Args}, State) ->
                     pass
             end,
             case dgiot_parse:query_object(<<"View">>, #{<<"where">> => #{<<"key">> => ProducttempletId, <<"class">> => <<"ProductTemplet">>}}) of
-                {ok, #{<<"results">> := Views}} ->
+                {ok, #{<<"results">> := Views}} when length(Views) > 0 ->
                     ViewRequests =
                         lists:foldl(fun(View, Acc) ->
                             NewDict = maps:without([<<"createdAt">>, <<"objectId">>, <<"updatedAt">>], View),
@@ -166,7 +166,32 @@ handle_message({sync_parse, Args}, State) ->
                                     end, [], Views),
                     dgiot_parse:batch(ViewRequests);
                 _ ->
-                    pass
+                    NewConfig = #{
+                        <<"konva">> => #{
+                            <<"Stage">> => #{
+                                <<"attrs">> => #{
+                                    <<"width">> => <<"1200">>,
+                                    <<"height">> => <<"700">>},
+                                <<"className">> => <<"Stage">>,
+                                <<"children">> => [#{
+                                    <<"attrs">> => #{
+                                        <<"id">> => <<"Layer_Thing">>},
+                                    <<"className">> => <<"Layer">>,
+                                    <<"children">> => [#{
+                                        <<"attrs">> => #{
+                                            <<"id">> => <<"bg">>,
+                                            <<"type">> => <<"bg-image">>,
+                                            <<"width">> => <<"1200">>,
+                                            <<"height">> => <<"700">>,
+                                            <<"src">> => <<"//img7.ddove.com/upload/20181127/134600237598.jpg?timestamp=1635422987361">>},
+                                        <<"className">> => <<"Image">>}]}]}}},
+                    dgiot_parse:create_object(<<"View">>, #{
+                        <<"title">> => ObjectId,
+                        <<"key">> => ObjectId,
+                        <<"type">> => <<"topo">>,
+                        <<"class">> => <<"Product">>,
+                        <<"data">> => NewConfig
+                    })
             end;
         _ ->
             pass

dgiot_install.sh

@@ -747,6 +747,7 @@ function update_dgiot() {
 }
 
 function install_dgiot() {
+  make_ssl
   if [ ! -f ${script_dir}/${dgiot}.tar.gz ]; then
     wget $fileserver/${dgiot}.tar.gz -O ${script_dir}/${dgiot}.tar.gz &> /dev/null
   fi
@@ -960,6 +961,29 @@ function install_nginx() {
     systemctl enable nginx.service &> /dev/null
 }
 
+function make_ssl() {
+    if [ ! -d /etc/ssl/dgiot/ ]; then
+      mkdir -p /etc/ssl/dgiot/
+      cd /etc/ssl/dgiot/
+
+      # 生成自签名的CA key和证书
+      openssl genrsa -out ca.key 2048
+      openssl req -x509 -new -nodes -key ca.key -sha256 -days 3650 -subj "/CN=${wlanip}" -out ca.pem
+
+      # 生成服务器端的key和证书
+      openssl genrsa -out server.key 2048
+      openssl req -new -key ./server.key -out server.csr -subj "/CN=0.0.0.0"
+      openssl x509 -req -in ./server.csr -CA ca.pem -CAkey ca.key -CAcreateserial -out server.pem -days 3650 -sha256
+
+      # 生成客户端key和证书
+      openssl genrsa -out client.key 2048
+      openssl req -new -key ./client.key -out client.csr -subj "/CN=0.0.0.0"
+      openssl x509 -req -in ./client.csr -CA ca.pem -CAkey ca.key -CAcreateserial -out client.pem -days 3650 -sha256
+
+      cd ${script_dir}/
+    fi
+}
+
 ## ==============================Main program starts from here============================
 
 set -e

etc/emqx.conf

@@ -1385,20 +1385,20 @@ listener.ssl.external.handshake_timeout = 15s
 ## See: http://erlang.org/doc/man/ssl.html
 ##
 ## Value: File
-listener.ssl.external.keyfile = /etc/ssl/certs/domain_name.key
+listener.ssl.external.keyfile = /etc/ssl/dgiot/server.key
 
 ## Path to a file containing the user certificate.
 ##
 ## See: http://erlang.org/doc/man/ssl.html
 ##
 ## Value: File
-listener.ssl.external.certfile = /etc/ssl/certs/domain_name.pem
+listener.ssl.external.certfile = /etc/ssl/dgiot/server.pem
 
 ## Path to the file containing PEM-encoded CA certificates. The CA certificates
 ## are used during server authentication and when building the client certificate chain.
 ##
 ## Value: File
-## listener.ssl.external.cacertfile = {{ platform_etc_dir }}/certs/cacert.pem
+listener.ssl.external.cacertfile = /etc/ssl/dgiot/ca.pem
 
 ## The Ephemeral Diffie-Helman key exchange is a very effective way of
 ## ensuring Forward Secrecy by exchanging a set of keys that never hit