fix(mongo): update the health check method

apps/emqx_connector/rebar.config

@@ -8,7 +8,7 @@
   {mysql, {git, "https://github.com/emqx/mysql-otp", {tag, "1.7.1"}}},
   {epgsql, {git, "https://github.com/epgsql/epgsql", {tag, "4.4.0"}}},
   %% NOTE: mind poolboy version when updating mongodb-erlang version
-  {mongodb, {git,"https://github.com/emqx/mongodb-erlang", {tag, "v3.0.8"}}},
+  {mongodb, {git,"https://github.com/emqx/mongodb-erlang", {tag, "v3.0.9"}}},
   %% NOTE: mind poolboy version when updating eredis_cluster version
   {eredis_cluster, {git, "https://github.com/emqx/eredis_cluster", {tag, "0.6.7"}}},
   %% mongodb-erlang uses a special fork https://github.com/comtihon/poolboy.git

apps/emqx_connector/src/emqx_connector_mongo.erl

@@ -148,15 +148,24 @@ on_query(InstId, {Action, Collection, Selector, Docs}, AfterQuery, #{poolname :=
     end.
 
 -dialyzer({nowarn_function, [on_health_check/2]}).
-on_health_check(_InstId, #{test_opts := TestOpts} = State) ->
-    case mc_worker_api:connect(TestOpts) of
-        {ok, TestConn} ->
-            mc_worker_api:disconnect(TestConn),
-            {ok, State};
-        {error, _} ->
-            {error, health_check_failed, State}
+on_health_check(_InstId, #{poolname := PoolName} = State) ->
+    case health_check(PoolName) of
+        true -> {ok, State};
+        false -> {error, health_check_failed, State}
     end.
 
+health_check(PoolName) ->
+    Status = [begin
+        case ecpool_worker:client(Worker) of
+            {ok, Conn} ->
+                %% we don't care if this returns something or not, we just to test the connection
+                Res = mongo_api:find_one(Conn, <<"foo">>, {}, #{}),
+                Res == undefined orelse is_map(Res);
+            _ -> false
+        end
+    end || {_WorkerName, Worker} <- ecpool:workers(PoolName)],
+    length(Status) > 0 andalso lists:all(fun(St) -> St =:= true end, Status).
+
 %% ===================================================================
 connect(Opts) ->
     Type = proplists:get_value(mongo_type, Opts, single),