feat: introduce EMQX 6.0 compatibility

Author: savonarola

Makefile

@@ -77,6 +77,10 @@ up:
 down:
 	docker compose down --volumes
 
+.PHONY: emqx-logs
+emqx-logs:
+	docker compose logs -f emqx
+
 # bump-version-patch/minor/major
 .PHONY: bump-version-%
 bump-version-%:

docker-compose.yml

@@ -1,6 +1,6 @@
 services:
   emqx:
-    image: emqx/emqx-enterprise:5.9.0
+    image: emqx/emqx-enterprise:6.0.0
     container_name: emqx
     environment:
       EMQX_LOG__CONSOLE__LEVEL: debug
@@ -128,6 +128,7 @@ services:
 
   redis-cluster-node-1:
     image: redis:7.0
+    container_name: redis-cluster-node-1
     command: redis-server --requirepass public --masterauth public --port 7001 --cluster-enabled yes --cluster-config-file nodes.conf --cluster-node-timeout 5000
     ports:
       - "7001:7001"
@@ -141,6 +142,7 @@ services:
 
   redis-cluster-node-2:
     image: redis:7.0
+    container_name: redis-cluster-node-2
     command: redis-server --requirepass public --masterauth public --port 7002 --cluster-enabled yes --cluster-config-file nodes.conf --cluster-node-timeout 5000
     ports:
       - "7002:7002"
@@ -154,6 +156,7 @@ services:
 
   redis-cluster-node-3:
     image: redis:7.0
+    container_name: redis-cluster-node-3
     command: redis-server --requirepass public --masterauth public --port 7003 --cluster-enabled yes --cluster-config-file nodes.conf --cluster-node-timeout 5000
     ports:
       - "7003:7003"
@@ -167,6 +170,7 @@ services:
 
   redis-cluster-setup:
     image: redis:7.0
+    container_name: redis-cluster-setup
     depends_on:
       redis-cluster-node-1:
         condition: service_healthy
@@ -180,6 +184,7 @@ services:
 
   redis-master:
     image: redis:7.0
+    container_name: redis-master
     command: redis-server --requirepass public --masterauth public --port 8001
     ports:
       - "8001:8001"
@@ -193,6 +198,7 @@ services:
 
   redis-slave:
     image: redis:7.0
+    container_name: redis-slave
     command: redis-server --requirepass public --masterauth public --port 8002 --replicaof redis-master 8001
     ports:
       - "8002:8002"
@@ -209,6 +215,7 @@ services:
 
   redis-sentinel:
     image: redis:7.0
+    container_name: redis-sentinel
     command: >
       bash -c "
         echo 'sentinel monitor mymaster redis-master 8001 1' > sentinel.conf

src/emqx_omp_mysql.erl

@@ -423,7 +423,8 @@ make_mysql_resource_config(#{<<"insert_message_sql">> := InsertMessageStatement}
         }
     },
 
-    ResourceOpts = emqx_omp_utils:make_resource_opts(RawConfig0),
+    ResourceOpts0 = emqx_omp_utils:make_resource_opts(RawConfig0),
+    ResourceOpts = ResourceOpts0#{spawn_buffer_workers => true},
 
     {MysqlConfig, ResourceOpts}.
 
@@ -437,7 +438,7 @@ init_default_schema(#{<<"init_default_schema">> := true} = ConfigRaw) ->
     {ok, _} = emqx_resource:create_local(
         ?RESOURCE_ID_INIT,
         ?RESOURCE_GROUP,
-        emqx_bridge_mysql_connector,
+        emqx_mysql,
         MysqlConfig,
         ResourceOpts
     ),

src/emqx_omp_utils.erl

@@ -45,7 +45,8 @@ make_resource_opts(RawConfig) ->
         %% nothing to remove yet.
         %% After some time, the message will be persisted in the database, and the client
         %% will receive it once more on reconnect.
-        query_mode => sync
+        query_mode => sync,
+        owner_id => <<"omp">>
     }.
 
 check_config(Schema, ConfigRaw) ->

test/emqx_omp_SUITE.erl

@@ -264,7 +264,11 @@ t_message_order(_Config) ->
     ok = emqtt:stop(ClientSub),
 
     %% Check messages order
-    ?assertEqual(lists:seq(1, 200), Messages).
+    Expected = lists:seq(1, 200),
+    ?assertEqual([], Expected -- Messages, "Not all messages were received"),
+    ?assertEqual([], Messages -- Expected, "Duplicate messages were received"),
+    InvalidOrder = lists:filter(fun({A, B}) -> A =/= B end, lists:zip(Expected, Messages)),
+    ?assertEqual([], InvalidOrder, "Messages were received in the wrong order").
 
 t_mysql_table_cleanup(Config) ->
     %% setup and cleanup

test/emqx_omp_test_helpers.erl

@@ -78,16 +78,16 @@ api_delete(Path) ->
 
 make_request({get, Path}) ->
     ct:pal("GET ~s~n~n", [api_url(Path)]),
-    hackney:request(get, api_url(Path), headers());
+    hackney:request(get, api_url(Path), headers(), <<>>, hackney_options());
 make_request({post, Path, 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));
+    hackney:request(post, api_url(Path), headers(), encode_json(Body), hackney_options());
 make_request({put, Path, 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));
+    hackney:request(put, api_url(Path), headers(), encode_json(Body), hackney_options());
 make_request({delete, Path}) ->
     ct:pal("DELETE ~s~n~n", [api_url(Path)]),
-    hackney:request(delete, api_url(Path), headers()).
+    hackney:request(delete, api_url(Path), headers(), <<>>, hackney_options()).
 
 handle_result({ok, Code, _Headers, ClientRef}) when Code >= 200 andalso Code < 300 ->
     {ok, Json} = hackney:body(ClientRef),
@@ -152,3 +152,6 @@ decode_json(Body) ->
 
 encode_json(Data) ->
     jiffy:encode(Data).
+
+hackney_options() ->
+    [{recv_timeout, 10000}, {connect_timeout, 5000}, {checkout_timeout, 5000}].