diff --git a/apps/emqx_connector/src/emqx_connector_http.erl b/apps/emqx_connector/src/emqx_connector_http.erl index 33407f17c..bd129f1a8 100644 --- a/apps/emqx_connector/src/emqx_connector_http.erl +++ b/apps/emqx_connector/src/emqx_connector_http.erl @@ -234,6 +234,7 @@ on_start( PoolName = emqx_plugin_libs_pool:pool_name(InstId), State = #{ pool_name => PoolName, + pool_type => PoolType, host => Host, port => Port, connect_timeout => ConnectTimeout, @@ -283,7 +284,7 @@ on_query(InstId, {Method, Request, Timeout}, State) -> on_query( InstId, {KeyOrNum, Method, Request, Timeout, Retry}, - #{pool_name := PoolName, base_path := BasePath} = State + #{base_path := BasePath} = State ) -> ?TRACE( "QUERY", @@ -291,12 +292,10 @@ on_query( #{request => Request, connector => InstId, state => State} ), NRequest = formalize_request(Method, BasePath, Request), + Worker = resolve_pool_worker(State, KeyOrNum), case ehttpc:request( - case KeyOrNum of - undefined -> PoolName; - _ -> {PoolName, KeyOrNum} - end, + Worker, Method, NRequest, Timeout, @@ -364,19 +363,15 @@ on_query_async( InstId, {KeyOrNum, Method, Request, Timeout}, ReplyFunAndArgs, - #{pool_name := PoolName, base_path := BasePath} = State + #{base_path := BasePath} = State ) -> + Worker = resolve_pool_worker(State, KeyOrNum), ?TRACE( "QUERY_ASYNC", "http_connector_received", #{request => Request, connector => InstId, state => State} ), NRequest = formalize_request(Method, BasePath, Request), - Worker = - case KeyOrNum of - undefined -> ehttpc_pool:pick_worker(PoolName); - _ -> ehttpc_pool:pick_worker(PoolName, KeyOrNum) - end, ok = ehttpc:request_async( Worker, Method, @@ -386,6 +381,16 @@ on_query_async( ), {ok, Worker}. +resolve_pool_worker(State, undefined) -> + resolve_pool_worker(State, self()); +resolve_pool_worker(#{pool_name := PoolName} = State, Key) -> + case maps:get(pool_type, State, random) of + random -> + ehttpc_pool:pick_worker(PoolName); + hash -> + ehttpc_pool:pick_worker(PoolName, Key) + end. + on_get_status(_InstId, #{pool_name := PoolName, connect_timeout := Timeout} = State) -> case do_get_status(PoolName, Timeout) of ok ->