feat: add readme

Author: savonarola

.github/workflows/ci.yml

@@ -10,32 +10,32 @@ on:
   workflow_dispatch:
 
 jobs:
-  # test:
-  #   name: Run tests
-  #   runs-on: ubuntu-24.04
+  test:
+    name: Run tests
+    runs-on: ubuntu-24.04
 
-  #   steps:
-  #   - uses: actions/checkout@v4
+    steps:
+    - uses: actions/checkout@v4
 
-  #   - name: Setup Erlang
-  #     uses: erlef/setup-beam@v1
-  #     with:
-  #       otp-version: "26"
-  #       rebar3-version: "3"
+    - name: Setup Erlang
+      uses: erlef/setup-beam@v1
+      with:
+        otp-version: "26"
+        rebar3-version: "3"
 
-  #   - name: Start Dependencies
-  #     run: make up
+    - name: Start Dependencies
+      run: make up
 
-  #   - name: Run Tests
-  #     run: make ct
+    - name: Run Tests
+      run: make ct
 
-  #   - name: Cleanup
-  #     if: always()
-  #     run: make down
+    - name: Cleanup
+      if: always()
+      run: make down
 
   release:
     name: Publish release
-    # needs: test
+    needs: test
     if: startsWith(github.ref, 'refs/tags/')
     runs-on: ubuntu-24.04
 

.github/workflows/run_tests.yml

@@ -1,5 +1,85 @@
 # EMQX Offline Message Plugin
 
+## Usage
+
+Download the plugin:
+
+```bash
+wget https://github.com/savonarola/emqx-offline-message-plugin/releases/download/v0.0.4/emqx_offline_message_plugin-1.0.0.tar.gz
+```
+
+Install the plugin:
+
+```bash
+curl -u key:secret -X POST http://localhost:18083/api/v5/plugins/install \
+-H "Content-Type: multipart/form-data" \
+-F "plugin=@emqx_offline_message_plugin-1.0.0.tar.gz"
+```
+
+Check the plugin is installed:
+
+```bash
+curl -u key:secret http://localhost:18083/api/v5/plugins | jq
+```
+
+Add connector:
+
+```bash
+curl -u key:secret -X POST http://localhost:18083/api/v5/connectors \
+-H "Content-Type: application/json" \
+-d '{"type":"mysql","name":"omp","server":"mysql","database":"emqx","pool_size":8,"username":"emqx","password":"public","resource_opts":{"start_timeout":"5s","health_check_interval":"15s"}}'
+
+curl -s -u key:secret http://localhost:18083/api/v5/connectors | jq
+```
+
+Add publish action:
+
+```bash
+curl -u key:secret -X POST http://localhost:18083/api/v5/actions \
+-H "Content-Type: application/json" \
+-d '{"type":"mysql","parameters":{"undefined_vars_as_null":false,"sql":"insert into mqtt_msg(msgid, sender, topic, qos, retain, payload, arrived) values(${id}, ${clientid}, ${topic}, ${qos}, ${retain}, ${payload}, FROM_UNIXTIME(${timestamp}/1000))"},"name":"omp_publish_action","enable":true,"connector":"omp"}'
+
+curl -s -u key:secret http://localhost:18083/api/v5/actions | jq
+```
+
+Add publish rule:
+
+```bash
+curl -u key:secret -X POST http://localhost:18083/api/v5/rules \
+-H "Content-Type: application/json" \
+-d '{"sql":"SELECT id, clientid, topic, qos, payload, timestamp, int(coalesce(flags.retain, 0)) as retain FROM \"t/#\" WHERE qos = 1 or qos = 2","name":"omp_publish_rule","description":"Offline Message Plugin Publish Action","actions":["mysql:omp_publish_action"]}'
+
+curl -s -u key:secret http://localhost:18083/api/v5/rules | jq
+```
+
+Add subscribe/ack rule.
+This the only configuration change that is unavailable from the Dashboard.
+
+```bash
+curl -u key:secret -X POST http://localhost:18083/api/v5/rules \
+-H "Content-Type: application/json" \
+-d '{"sql":"SELECT * FROM \"$events/session_subscribed\", \"$events/message_acked\" WHERE topic =~ '"'"'t/#'"'"'","name":"omp_subscribe_ack_rule","description":"Offline Message Plugin Subscribe/Ack Action","actions":[{"function":"emqx_omp:action","args":{"opts":{},"connector_name":"mysql:omp"}}]}'
+
+curl -s -u key:secret http://localhost:18083/api/v5/rules | jq
+```
+
+Verify:
+
+```bash
+mosquitto_pub -d -q 1 -t 't/2' -m 'hello-from-offline1'
+mosquitto_pub -d -q 1 -t 't/2' -m 'hello-from-offline2'
+mosquitto_pub -d -q 1 -t 't/2' -m 'hello-from-offline3'
+
+mosquitto_sub -d -q 1 -t 't/2' -i $(pwgen 20 -1)
+```
+
+No messages should be received:
+
+```bash
+mosquitto_sub -d -q 1 -t 't/2' -i $(pwgen 20 -1)
+```
+
+
 ## Release
 
 An EMQX plugin release is a tar file including including a subdirectory of this plugin's name and it's version, that contains:

README.md

@@ -27,7 +27,6 @@
 -spec on_message_acked(Envs :: map(), ConnectorName :: binary(), Opts :: map()) -> ok.
 on_message_acked(#{id := MsgId} = Envs, ConnectorName, Opts) ->
     ResourceId = resource_id(ConnectorName),
-
     case emqx_resource:simple_sync_query(ResourceId, {sql, ?DELETE_SQL, [MsgId]}) of
         ok ->
             emqx_metrics_worker:inc(emqx_omp_metrics_worker, message_acked, success),

src/emqx_omp_mysql.erl

@@ -48,7 +48,6 @@ init_per_suite(Config) ->
     [{plugin_id, PluginId}, {plugin_filename, Filename} | Config].
 
 end_per_suite(_Config) ->
-    % cleanup
     ok = emqx_omp_test_api_helpers:delete_all_plugins(),
     ok = emqx_omp_test_helpers:stop(),
     ok.
@@ -90,20 +89,12 @@ end_per_testcase(_Case, _Config) ->
 %%--------------------------------------------------------------------
 
 t_ok(_Config) ->
-    %% create actions and rules
-    % PublishAction = publish_mysql_action("omp_publish_action", "omp"),
-    % ok = emqx_omp_test_api_helpers:create_action(PublishAction),
-    % PublishRule = publish_mysql_rule("omp_publish_rule", "t/#", "mysql:omp_publish_action"),
-    % ok = emqx_omp_test_api_helpers:create_rule(PublishRule),
-    % SubscribeAckRule = subscribe_ack_rule("omp_subscribe_ack_rule", "t/#", "mysql:omp", #{}),
-    % ok = emqx_omp_test_api_helpers:create_rule(SubscribeAckRule),
-
     %% publish message
     Payload = emqx_guid:to_hexstr(emqx_guid:gen()),
     ClientPub = emqtt_connect(),
     _ = emqtt:publish(ClientPub, <<"t/1">>, Payload, 1),
     ok = emqtt:stop(ClientPub),
-    ct:sleep(1000),
+    ct:sleep(500),
 
     %% A new subscriber should receive the message
     ClientSub0 = emqtt_connect(),
@@ -115,6 +106,7 @@ t_ok(_Config) ->
         ct:fail("Message not received")
     end,
     ok = emqtt:stop(ClientSub0),
+    ct:sleep(500),
 
     %% Another subscriber should NOT receive the message:
     %% it should be deleted.

test/emqx_omp_SUITE.erl

@@ -26,6 +26,7 @@ api_get(Path) ->
     handle_result(Result).
 
 api_get_raw(Path) ->
+    ct:pal("GET ~s~n~n", [api_url(Path)]),
     Result = hackney:request(get, api_url(Path), []),
     handle_result_raw(Result).
 
@@ -34,6 +35,8 @@ api_post(Path, Body) ->
     handle_result(Result).
 
 api_post_raw(Path, Headers0, Body) ->
+    HeadersS = [io_lib:format("~s: ~s~n", [Key, Value]) || {Key, Value} <- Headers0],
+    ct:pal("POST ~s~n~s~n~n...", [api_url(Path), iolist_to_binary(HeadersS)]),
     Headers = [auth_header() | Headers0],
     Result = hackney:request(post, api_url(Path), Headers, Body),
     handle_result_raw(Result).
@@ -51,16 +54,16 @@ api_delete(Path) ->
     handle_result(Result).
 
 make_request({get, Path}) ->
-    ct:print("GET ~s~n~n", [api_url(Path)]),
+    ct:pal("GET ~s~n~n", [api_url(Path)]),
     hackney:request(get, api_url(Path), headers());
 make_request({post, Path, Body}) ->
-    ct:print("POST ~s~n~n~s~n~n", [api_url(Path), encode_json(Body)]),
+    ct:pal("POST ~s~n~n~s~n~n", [api_url(Path), encode_json(Body)]),
     hackney:request(post, api_url(Path), headers(), encode_json(Body));
 make_request({put, Path, Body}) ->
-    ct:print("PUT ~s~n~n~s~n~n", [api_url(Path), encode_json(Body)]),
+    ct:pal("PUT ~s~n~n~s~n~n", [api_url(Path), encode_json(Body)]),
     hackney:request(put, api_url(Path), headers(), encode_json(Body));
 make_request({delete, Path}) ->
-    ct:print("DELETE ~s~n~n", [api_url(Path)]),
+    ct:pal("DELETE ~s~n~n", [api_url(Path)]),
     hackney:request(delete, api_url(Path), headers()).
 
 handle_result({ok, Code, _Headers, ClientRef}) when Code >= 200 andalso Code < 300 ->