test: rulesql select fields and select from event message

apps/emqx_rule_engine/test/emqx_rulesql_SUITE.erl

@@ -21,6 +21,7 @@
 
 -include_lib("emqx_rule_engine/include/rule_engine.hrl").
 -include_lib("emqx/include/emqx.hrl").
+-include_lib("emqx/include/emqx_mqtt.hrl").
 
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("common_test/include/ct.hrl").
@@ -51,15 +52,16 @@ groups() ->
       , t_sqlselect_02
       , t_sqlselect_1
       , t_sqlselect_2
-      , t_sqlselect_2_1
-      , t_sqlselect_2_2
-      , t_sqlselect_2_3
-      , t_sqlselect_3
       ]},
      {rulesql_select_events, [],
-      [ t_sqlparse_event_1
+      [ t_sqlparse_event_client_connected_01
-      , t_sqlparse_event_2
+      , t_sqlparse_event_client_connected_02
-      , t_sqlparse_event_3
+      , t_sqlparse_event_client_disconnected
+      , t_sqlparse_event_session_subscribed
+      , t_sqlparse_event_session_unsubscribed
+      , t_sqlparse_event_message_delivered
+      , t_sqlparse_event_message_acked
+      , t_sqlparse_event_message_dropped
       ]},
      {rulesql_select_metadata, [],
       [ t_sqlparse_select_matadata_1
@@ -159,16 +161,35 @@ end_per_testcase(_TestCase, _Config) ->
 
 t_sqlselect_0(_Config) ->
     %% Verify SELECT with and without 'AS'
-    Sql = "select * "
+
-          "from \"t/#\" "
+    Sql1 = "SELECT * "
-          "where payload.cmd.info = 'tt'",
+        "FROM \"t/#\" "
-    ?assertMatch({ok,#{payload := <<"{\"cmd\": {\"info\":\"tt\"}}">>}},
+        "WHERE payload.cmd.info = 'tt'",
-                 emqx_rule_sqltester:test(
+    %% all available fields by `select` from message publish, selecte other key will be `undefined`
-                    #{<<"rawsql">> => Sql,
+    Topic = <<"t/a">>,
+    Payload = <<"{\"cmd\": {\"info\":\"tt\"}}">>,
+    {ok, #{username := <<"u_emqx">>,
+           topic := Topic,
+           timestamp := TimeStamp,
+           qos := 1,
+           publish_received_at := TimeStamp,
+           peerhost := <<"127.0.0.1">>,
+           payload := Payload,
+           node := 'test@127.0.0.1',
+           metadata := #{rule_id := TestRuleId},
+           id := MsgId,
+           flags := #{sys := true, event := true},
+           clientid := <<"c_emqx">>
+          }
+    } = emqx_rule_sqltester:test(
+          #{<<"rawsql">> => Sql1,
             <<"ctx">> =>
-                        #{<<"payload">> =>
+                #{<<"payload">> => Payload,
-                            <<"{\"cmd\": {\"info\":\"tt\"}}">>,
+                  <<"topic">> => Topic}}),
-                          <<"topic">> => <<"t/a">>}})),
+    ?assert(is_binary(TestRuleId)),
+    ?assert(is_binary(MsgId)),
+    ?assert(is_integer(TimeStamp)),
+
     Sql2 = "select payload.cmd as cmd "
            "from \"t/#\" "
            "where cmd.info = 'tt'",
@@ -179,6 +200,7 @@ t_sqlselect_0(_Config) ->
                         #{<<"payload">> =>
                             <<"{\"cmd\": {\"info\":\"tt\"}}">>,
                           <<"topic">> => <<"t/a">>}})),
+
     Sql3 = "select payload.cmd as cmd, cmd.info as info "
            "from \"t/#\" "
            "where cmd.info = 'tt' and info = 'tt'",
@@ -205,14 +227,15 @@ t_sqlselect_0(_Config) ->
 
 t_sqlselect_00(_Config) ->
     %% Verify plus/subtract and unary_add_or_subtract
-    Sql = "select 1-1 as a "
+    Sql0 = "select 1 - 1 as a "
           "from \"t/#\" ",
     ?assertMatch({ok,#{<<"a">> := 0}},
                  emqx_rule_sqltester:test(
-                    #{<<"rawsql">> => Sql,
+                    #{<<"rawsql">> => Sql0,
                       <<"ctx">> =>
                         #{<<"payload">> => <<"">>,
                           <<"topic">> => <<"t/a">>}})),
+
     Sql1 = "select -1 + 1 as a "
            "from \"t/#\" ",
     ?assertMatch({ok,#{<<"a">> := 0}},
@@ -221,6 +244,7 @@ t_sqlselect_00(_Config) ->
                       <<"ctx">> =>
                         #{<<"payload">> => <<"">>,
                           <<"topic">> => <<"t/a">>}})),
+
     Sql2 = "select 1 + 1 as a "
            "from \"t/#\" ",
     ?assertMatch({ok,#{<<"a">> := 2}},
@@ -229,6 +253,7 @@ t_sqlselect_00(_Config) ->
                       <<"ctx">> =>
                         #{<<"payload">> => <<"">>,
                           <<"topic">> => <<"t/a">>}})),
+
     Sql3 = "select +1 as a "
            "from \"t/#\" ",
     ?assertMatch({ok,#{<<"a">> := 1}},
@@ -431,58 +456,93 @@ t_sqlselect_2(_Config) ->
     emqtt:stop(Client),
     emqx_rule_registry:remove_rule(TopicRule).
 
-t_sqlselect_2_1(_Config) ->
+%%------------------------------------------------------------------------------
+%% Test cases for events
+%%------------------------------------------------------------------------------
+
+%% FROM $events/client_connected
+t_sqlparse_event_client_connected_01(_Config) ->
+    Sql = "select *"
+          "from \"$events/client_connected\" ",
+
+    %% all available fields by `select` from message publish, selecte other key will be `undefined`
+    {ok, #{clientid := <<"c_emqx">>,
+           username := <<"u_emqx">>,
+           timestamp := TimeStamp,
+           connected_at := TimeStamp,
+           peername := <<"127.0.0.1:12345">>,
+           metadata := #{rule_id := RuleId},
+           %% default value
+           node := 'test@127.0.0.1',
+           sockname := <<"0.0.0.0:1883">>,
+           proto_name := <<"MQTT">>,
+           proto_ver := 5,
+           mountpoint := undefined,
+           keepalive := 60,
+           is_bridge := false,
+           expiry_interval := 3600,
+           event := 'client.connected',
+           clean_start := true
+          }
+    } = emqx_rule_sqltester:test(
+          #{<<"rawsql">> => Sql,
+            <<"ctx">> => #{<<"clientid">> => <<"c_emqx">>, <<"peername">> => <<"127.0.0.1:12345">>, <<"username">> => <<"u_emqx">>}}),
+    ?assert(is_binary(RuleId)),
+    ?assert(is_integer(TimeStamp)).
+
+t_sqlparse_event_client_connected_02(_Config) ->
     ok = emqx_rule_engine:load_providers(),
-    %% recursively republish to t2, if the msg dropped
+    %% republish the client.connected msg
     TopicRule = create_simple_repub_rule(
-                    <<"t2">>,
+                    <<"repub/to/connected">>,
                     "SELECT * "
-                    "FROM \"$events/message_dropped\" "),
+                    "FROM \"$events/client_connected\" "
-    {ok, Client} = emqtt:start_link([{username, <<"emqx">>}]),
+                    "WHERE username = 'emqx1'",
+                    <<"{clientid: ${clientid}}">>),
+    {ok, Client} = emqtt:start_link([{clientid, <<"emqx0">>}, {username, <<"emqx0">>}]),
     {ok, _} = emqtt:connect(Client),
-    emqtt:publish(Client, <<"t2">>, <<"{\"x\":1,\"y\":144}">>, 0),
+    {ok, _, _} = emqtt:subscribe(Client, <<"repub/to/connected">>, 0),
-    Fun = fun() ->
+    ct:sleep(200),
-            receive {publish, #{topic := <<"t2">>, payload := _}} ->
+    {ok, Client1} = emqtt:start_link([{clientid, <<"c_emqx1">>}, {username, <<"emqx1">>}]),
-                received_t2
+    {ok, _} = emqtt:connect(Client1),
-            after 500 ->
+    receive {publish, #{topic := T, payload := Payload}} ->
-                received_nothing
+        ?assertEqual(<<"repub/to/connected">>, T),
-            end
+        ?assertEqual(<<"{clientid: c_emqx1}">>, Payload)
+    after 1000 ->
+        ct:fail(wait_for_t2)
     end,
-    received_nothing = Fun(),
 
-    %% it should not keep republishing "t2"
+    emqtt:publish(Client, <<"t1">>, <<"{\"x\":1,\"y\":1}">>, 0),
-    {ok, _, _} = emqtt:subscribe(Client, <<"t2">>, 0),
+    receive {publish, #{topic := <<"t2">>, payload := _}} ->
-    received_nothing = Fun(),
+        ct:fail(unexpected_t2)
+    after 1000 ->
+        ok
+    end,
 
-    emqtt:stop(Client),
+    emqtt:stop(Client), emqtt:stop(Client1),
     emqx_rule_registry:remove_rule(TopicRule).
 
-t_sqlselect_2_2(_Config) ->
+%% FROM $events/client_disconnected
-    ok = emqx_rule_engine:load_providers(),
+t_sqlparse_event_client_disconnected(_Config) ->
-    %% recursively republish to t2, if the msg acked
+    %% TODO
-    TopicRule = create_simple_repub_rule(
+    ok.
-                    <<"t2">>,
-                    "SELECT * "
-                    "FROM \"$events/message_acked\" "),
-    {ok, Client} = emqtt:start_link([{username, <<"emqx">>}]),
-    {ok, _} = emqtt:connect(Client),
-    {ok, _, _} = emqtt:subscribe(Client, <<"t2">>, 1),
-    emqtt:publish(Client, <<"t2">>, <<"{\"x\":1,\"y\":144}">>, 1),
-    Fun = fun() ->
-            receive {publish, #{topic := <<"t2">>, payload := _}} ->
-                received_t2
-            after 500 ->
-                received_nothing
-            end
-          end,
-    received_t2 = Fun(),
-    received_t2 = Fun(),
-    received_nothing = Fun(),
 
-    emqtt:stop(Client),
+%% FROM $events/session_subscribed
-    emqx_rule_registry:remove_rule(TopicRule).
+t_sqlparse_event_session_subscribed(_Config) ->
+    Sql = "select topic as tp "
+          "from \"$events/session_subscribed\" ",
+    ?assertMatch({ok,#{<<"tp">> := <<"t/tt">>}},
+        emqx_rule_sqltester:test(
+        #{<<"rawsql">> => Sql,
+          <<"ctx">> => #{<<"topic">> => <<"t/tt">>}})).
 
-t_sqlselect_2_3(_Config) ->
+%% FROM $events/session_unsubscribed
+t_sqlparse_event_session_unsubscribed(_Config) ->
+    %% TODO
+    ok.
+
+%% FROM $events/message_delivered
+t_sqlparse_event_message_delivered(_Config) ->
     ok = emqx_rule_engine:load_providers(),
     %% recursively republish to t2, if the msg delivered
     TopicRule = create_simple_repub_rule(
@@ -507,65 +567,84 @@ t_sqlselect_2_3(_Config) ->
     emqtt:stop(Client),
     emqx_rule_registry:remove_rule(TopicRule).
 
-t_sqlselect_3(_Config) ->
+%% FROM $events/message_acked
+t_sqlparse_event_message_acked(_Config) ->
     ok = emqx_rule_engine:load_providers(),
-    %% republish the client.connected msg
+    %% republish to `repub/if/acked`, if the msg acked
     TopicRule = create_simple_repub_rule(
-                    <<"t2">>,
+                    <<"repub/if/acked">>,
                     "SELECT * "
-                    "FROM \"$events/client_connected\" "
+                    "FROM \"$events/message_acked\" "),
-                    "WHERE username = 'emqx1'",
+    {ok, Client} = emqtt:start_link([{username, <<"emqx">>}]),
-                    <<"clientid=${clientid}">>),
-    {ok, Client} = emqtt:start_link([{clientid, <<"emqx0">>}, {username, <<"emqx0">>}]),
     {ok, _} = emqtt:connect(Client),
-    {ok, _, _} = emqtt:subscribe(Client, <<"t2">>, 0),
+    {ok, _, _} = emqtt:subscribe(Client, <<"repub/if/acked">>, ?QOS_1),
-    ct:sleep(200),
+
-    {ok, Client1} = emqtt:start_link([{clientid, <<"c_emqx1">>}, {username, <<"emqx1">>}]),
+    Fun = fun() ->
-    {ok, _} = emqtt:connect(Client1),
+            receive {publish, #{topic := <<"repub/if/acked">>, payload := _}} ->
-    receive {publish, #{topic := T, payload := Payload}} ->
+                received_acked
-        ?assertEqual(<<"t2">>, T),
+            after 500 ->
-        ?assertEqual(<<"clientid=c_emqx1">>, Payload)
+                received_nothing
-    after 1000 ->
+            end
-        ct:fail(wait_for_t2)
           end,
 
-    emqtt:publish(Client, <<"t1">>, <<"{\"x\":1,\"y\":1}">>, 0),
+    %% sub with max qos1 to generate ack packet
-    receive {publish, #{topic := <<"t2">>, payload := _}} ->
+    {ok, _, _} = emqtt:subscribe(Client, <<"any/topic">>, ?QOS_1),
-        ct:fail(unexpected_t2)
-    after 1000 ->
-        ok
-    end,
 
-    emqtt:stop(Client), emqtt:stop(Client1),
+    Payload = <<"{\"x\":1,\"y\":144}">>,
+
+    %% even sub with qos1, but publish with qos0, no ack
+    emqtt:publish(Client, <<"any/topic">>, Payload, ?QOS_0),
+    received_nothing = Fun(),
+
+    %% sub and pub both are qos1, acked
+    emqtt:publish(Client, <<"any/topic">>, Payload, ?QOS_1),
+    received_acked = Fun(),
+    received_nothing = Fun(),
+
+    %% pub with qos2 but subscribed with qos1, acked
+    emqtt:publish(Client, <<"any/topic">>, Payload, ?QOS_2),
+    received_acked = Fun(),
+    received_nothing = Fun(),
+
+    emqtt:stop(Client),
     emqx_rule_registry:remove_rule(TopicRule).
 
-%%------------------------------------------------------------------------------
+%% FROM $events/message_dropped
-%% Test cases for events
+t_sqlparse_event_message_dropped(_Config) ->
-%%------------------------------------------------------------------------------
+    ok = emqx_rule_engine:load_providers(),
+    %% republish to `repub/if/dropped`, if any msg dropped
+    TopicRule = create_simple_repub_rule(
+                    <<"repub/if/dropped">>,
+                    "SELECT * "
+                    "FROM \"$events/message_dropped\" "),
+    {ok, Client} = emqtt:start_link([{username, <<"emqx">>}]),
+    {ok, _} = emqtt:connect(Client),
 
-t_sqlparse_event_1(_Config) ->
+    %% this message will be dropped and then repub to `repub/if/dropped`
-    Sql = "select topic as tp "
+    emqtt:publish(Client, <<"any/topic/1">>, <<"{\"x\":1,\"y\":144}">>, 0),
-          "from \"$events/session_subscribed\" ",
-    ?assertMatch({ok,#{<<"tp">> := <<"t/tt">>}},
-        emqx_rule_sqltester:test(
-        #{<<"rawsql">> => Sql,
-          <<"ctx">> => #{<<"topic">> => <<"t/tt">>}})).
 
-t_sqlparse_event_2(_Config) ->
+    Fun_t2 = fun() ->
-    Sql = "select clientid "
+               receive {publish, #{topic := <<"repub/if/dropped">>, payload := _}} ->
-          "from \"$events/client_connected\" ",
+                   received_repub
-    ?assertMatch({ok,#{<<"clientid">> := <<"abc">>}},
+               after 500 ->
-        emqx_rule_sqltester:test(
+                   received_nothing
-        #{<<"rawsql">> => Sql,
+               end
-          <<"ctx">> => #{<<"clientid">> => <<"abc">>}})).
+             end,
 
-t_sqlparse_event_3(_Config) ->
+    %% No client subscribed `any/topic`, triggered repub rule.
-    Sql = "select clientid, topic as tp "
+    %% But havn't sub `repub/if/dropped`, so the repub message will also be dropped with recursively republish.
-          "from \"t/tt\", \"$events/client_connected\" ",
+    received_nothing = Fun_t2(),
-    ?assertMatch({ok,#{<<"clientid">> := <<"abc">>, <<"tp">> := <<"t/tt">>}},
+
-        emqx_rule_sqltester:test(
+    {ok, _, _} = emqtt:subscribe(Client, <<"repub/if/dropped">>, 0),
-        #{<<"rawsql">> => Sql,
+
-          <<"ctx">> => #{<<"clientid">> => <<"abc">>, <<"topic">> => <<"t/tt">>}})).
+    %% this message will be dropped and then repub to `repub/to/t`
+    emqtt:publish(Client, <<"any/topic/2">>, <<"{\"x\":1,\"y\":144}">>, 0),
+
+    %% received subscribed `repub/to/t`
+    received_repub = Fun_t2(),
+
+    emqtt:stop(Client),
+    emqx_rule_registry:remove_rule(TopicRule).
 
 %%------------------------------------------------------------------------------
 %% Test cases for `foreach`