Merge pull request #146 from zmstone/251027-support-process-alias-as-request-ref

feat: support process alias as request reference

Author: zmstone

changelog.md

@@ -1,3 +1,9 @@
+* 4.3.0
+  - Allow process alias as request reference.
+    Previously `kpro_req_lib:produce/5` creates a reference for the caller,
+    now new API `kpro_req_lib:produce/6` can be used with pre-made reference.
+    If the reference is passed as `{alias, Ref}`, the Kafka response is sent to the alias, but not the `kpro:send` caller.
+
 * 4.2.9
   - Improve message encoding performance.
   - Allow `{magic_v2, Size, IoList}` as batch input for `produce` request.

include/kpro.hrl

@@ -21,15 +21,15 @@
 -include("kpro_error_codes.hrl").
 
 -record(kpro_req,
-        { ref :: reference()
+        { ref :: kpro:req_ref()
         , api :: kpro:api()
         , vsn :: kpro:vsn()
         , no_ack = false :: boolean() %% set to true for fire-n-forget requests
         , msg :: iodata() | kpro:struct()
         }).
 
 -record(kpro_rsp,
-        { ref :: false | reference()
+        { ref :: false | kpro:req_ref()
         , api :: kpro:api()
         , vsn :: kpro:vsn()
         , msg :: binary() | kpro:struct()

src/kpro.erl

@@ -118,6 +118,7 @@
              , producer_id/0
              , protocol/0
              , req/0
+             , req_ref/0
              , required_acks/0
              , rsp/0
              , schema/0
@@ -194,6 +195,7 @@
                 | [{field_name(), field_value()}].
 -type api() :: atom().
 -type req() :: #kpro_req{}.
+-type req_ref() :: reference() | {alias, reference()}.
 -type rsp() :: #kpro_rsp{}.
 -type compress_option() :: ?no_compression
                          | ?gzip

src/kpro_connection.erl

@@ -405,7 +405,7 @@ handle_msg({_, Sock, Bin}, #state{ sock     = Sock
   {CorrId, Body} = kpro_lib:decode_corr_id(Bin),
   {Caller, Ref, API, Vsn} = kpro_sent_reqs:get_req(Requests, CorrId),
   Rsp = kpro_rsp_lib:decode(API, Vsn, Body, Ref),
-  ok = cast(Caller, {msg, self(), Rsp}),
+  ok = kafka_reply(Caller, {msg, self(), Rsp}),
   NewRequests = kpro_sent_reqs:del(Requests, CorrId),
   State1 = maybe_flush_backlog(State#state{requests = NewRequests}),
   ?MODULE:loop(State1, Debug);
@@ -534,13 +534,12 @@ sasl_authenticate(#state{client_id = ClientId, mod = Mod, sock = Sock, remote =
   ok = setopts(Sock, Mod, [{active, once}]),
   State.
 
-cast(Pid, Msg) ->
-  try
-    Pid ! Msg,
-    ok
-  catch _ : _ ->
-    ok
-  end.
+kafka_reply(_CallerPid, {msg, _SelfPid, #kpro_rsp{ref = {alias, Ref}}} = Msg) ->
+  erlang:send(Ref, Msg),
+  ok;
+kafka_reply(Pid, Msg) when is_pid(Pid) ->
+  _ = erlang:send(Pid, Msg),
+  ok.
 
 system_continue(_Parent, Debug, State) ->
   ?MODULE:loop(State, Debug).

src/kpro_req_lib.erl

@@ -28,6 +28,7 @@
 
 -export([ produce/4
         , produce/5
+        , produce/6
         ]).
 
 -export([ metadata/2
@@ -51,6 +52,7 @@
 
 -export([ encode/3
         , make/3
+        , make/4
         ]).
 
 -export_type([ fetch_opts/0
@@ -250,17 +252,20 @@ fetch(Vsn, Topic, Partition, Offset, Opts) ->
   make(fetch, Vsn, Fields).
 
 %% @doc Help function to construct a non-transactional produce request.
-%% `Batch' arg can be be a `[map()]' like `[#{key => Key, value => Value, ts => Ts}]'.
-%%  Current system time will be taken if `ts' is missing in batch input.
-%%  It may also be `binary()' or `{magic_v2, Bytes, iolist()}' if user choose to encode
-%%  a batch beforehand which could be helpful when a large batch can be encoded
-%%  in other processes.
+%% @see produce/6.
 -spec produce(vsn(), topic(), partition(),
               binary() | batch_input() | {magic_v2, non_neg_integer(), iolist()}) -> req().
 produce(Vsn, Topic, Partition, Batch) ->
   produce(Vsn, Topic, Partition, Batch, #{}).
 
 %% @doc Help function to construct a produce request.
+%% @see produce/6.
+-spec produce(vsn(), topic(), partition(),
+              binary() | batch_input() | {magic_v2, non_neg_integer(), iolist()}, produce_opts()) -> req().
+produce(Vsn, Topic, Partition, Batch, Opts) when is_map(Opts) ->
+    produce(Vsn, Topic, Partition, Batch, Opts, make_ref()).
+
+%% @doc Help function to construct a produce request with a specified request reference.
 %% By default, it constructs a non-transactional produce request.
 %% `Batch' arg can be be a `[map()]' like `[#{key => Key, value => Value, ts => Ts}]'.
 %%  Current system time will be taken if `ts' is missing in batch input.
@@ -274,8 +279,8 @@ produce(Vsn, Topic, Partition, Batch) ->
 %%   monotonically increasing, with one sequence number per topic-partition.
 %% - `txn_ctx' (which is of spec `kpro:txn_ctx()') must exist in `Opts'
 -spec produce(vsn(), topic(), partition(),
-              binary() | batch_input() | {magic_v2, non_neg_integer(), iolist()}, produce_opts()) -> req().
-produce(Vsn, Topic, Partition, Batch, Opts) ->
+              binary() | batch_input() | {magic_v2, non_neg_integer(), iolist()}, produce_opts(), kpro:req_ref()) -> req().
+produce(Vsn, Topic, Partition, Batch, Opts, Ref) ->
   ok = assert_known_api_and_vsn(produce, Vsn),
   RequiredAcks = required_acks(maps:get(required_acks, Opts, all_isr)),
   Compression = maps:get(compression, Opts, ?no_compression),
@@ -313,7 +318,7 @@ produce(Vsn, Topic, Partition, Batch, Opts) ->
   #kpro_req{ api = produce
            , vsn = Vsn
            , msg = Msg
-           , ref = make_ref()
+           , ref = Ref
            , no_ack = RequiredAcks =:= 0
            }.
 
@@ -434,11 +439,17 @@ alter_configs(Vsn, Resources, Opts) ->
 %% @doc Help function to make a request body.
 -spec make(api(), vsn(), struct()) -> req().
 make(API, Vsn, Fields) ->
+  make(API, Vsn, Fields, make_ref()).
+
+%% @doc Help function to make a request body with provided reference or process alias.
+%% The refreence can be an alias to receive response this request.
+-spec make(api(), vsn(), struct(), reference()| {alias, reference()}) -> req().
+make(API, Vsn, Fields, Ref) ->
   ok = assert_known_api_and_vsn(API, Vsn),
   #kpro_req{ api = API
            , vsn = Vsn
            , msg = encode_struct(API, Vsn, Fields)
-           , ref = make_ref()
+           , ref = Ref
            }.
 
 %% @doc Encode a request to bytes that can be sent on wire.

src/kpro_sent_reqs.erl

@@ -39,7 +39,7 @@
 
 -type corr_id() :: kpro:corr_id().
 -define(REQ(Caller, Ref, API, Vsn, Ts), {Caller, Ref, API, Vsn, Ts}).
--type req() :: ?REQ(pid(), reference(), kpro:api(), kpro:vsn(), integer()).
+-type req() :: ?REQ(pid(), kpro:req_ref(), kpro:api(), kpro:vsn(), integer()).
 
 -record(requests,
         { corr_id = 0
@@ -61,7 +61,7 @@ is_empty(#requests{sent = Sent}) -> maps:size(Sent) == 0.
 
 %% @doc Add a new request to sent collection.
 %% Return the last corrlation ID and the new collection.
--spec add(requests(), pid(), reference(), kpro:api(), kpro:vsn()) ->
+-spec add(requests(), pid(), kpro:req_ref(), kpro:api(), kpro:vsn()) ->
         {corr_id(), requests()}.
 add(#requests{ corr_id = CorrId
              , sent    = Sent
@@ -82,7 +82,7 @@ del(#requests{sent = Sent} = Requests, CorrId) ->
 %% @doc Get caller of a request having the given correlation ID.
 %% Crash if the request is not found.
 -spec get_req(requests(), corr_id()) ->
-        {pid(), reference(), kpro:api(), kpro:vsn()}.
+        {pid(), kpro:req_ref(), kpro:api(), kpro:vsn()}.
 get_req(#requests{sent = Sent}, CorrId) ->
   ?REQ(Caller, Ref, API, Vsn, _Ts) = maps:get(CorrId, Sent),
   {Caller, Ref, API, Vsn}.

test/kpro_produce_tests.erl

@@ -171,6 +171,29 @@ async_send_test() ->
         receive Msg -> erlang:throw({unexpected, Msg}) after 10 -> ok end
     end).
 
+%% async send with alias test
+async_send_alias_test() ->
+  {_, Vsn} = get_api_vsn_range(),
+  Batch1 = [#{ts => kpro_lib:now_ts(), value => make_value(?LINE)}],
+  Batch2 = [#{ts => kpro_lib:now_ts(), value => make_value(?LINE)}],
+  Ref1 = {alias, erlang:alias([reply])},
+  Req1 = kpro_req_lib:produce(Vsn, topic(), ?PARTI, Batch1, #{}, Ref1),
+  Ref2 = {alias, erlang:alias([reply])},
+  Req2 = kpro_req_lib:produce(Vsn, topic(), ?PARTI, Batch2, #{}, Ref2),
+  with_connection(
+    fun(Pid) ->
+        {_, MRef} = spawn_monitor(fun() -> kpro:send(Pid, Req1) end),
+        receive {'DOWN', MRef, _, _, _} -> ok end,
+        _ = spawn(fun() -> kpro:request_async(Pid, Req2) end),
+        Assert = fun(Req, Rsp) ->
+                     ?ASSERT_RESPONSE_NO_ERROR(Vsn, Rsp),
+                     ?assertEqual(Req#kpro_req.ref, Rsp#kpro_rsp.ref)
+                 end,
+        receive {msg, Pid, Rsp1} -> Assert(Req1, Rsp1) end,
+        receive {msg, Pid, Rsp2} -> Assert(Req2, Rsp2) end,
+        receive Msg -> erlang:throw({unexpected, Msg}) after 10 -> ok end
+    end).
+
 make_req(Vsn) ->
   Batch = make_batch(Vsn),
   kpro_req_lib:produce(Vsn, topic(), ?PARTI, Batch).