refactor: make and send request in one func

Author: zmstone

src/wolff_producer.erl

@@ -423,27 +423,32 @@ use_defaults(Config, [{K, V} | Rest]) ->
     false -> use_defaults(Config#{K => V}, Rest)
   end.
 
-remake_requests(#{q_items := Items} = Sent, St, ?reconnect) ->
-  #kpro_req{ref = Ref} = Req = make_request(Items, St),
-  {[Req], [Sent#{req_ref := Ref}]};
-remake_requests(#{q_items := Items} = Sent, St, ?message_too_large) ->
-  Reqs = lists:map(fun(Item) -> make_request([Item], St) end, Items),
+resend_requests(#{q_items := Items} = Sent, St, ?reconnect) ->
+  {ok, Ref} = send_request(Items, St),
+  [Sent#{req_ref := Ref}];
+resend_requests(#{q_items := Items} = Sent, St, ?message_too_large) ->
+  Refs = lists:map(fun(Item) -> {ok, Ref} = send_request([Item], St), Ref end, Items),
   #{attempts := Attempts, q_ack_ref := Q_AckRef} = Sent,
-  {Reqs, make_sent_items_list(Items, Reqs, Attempts, Q_AckRef)}.
+  make_sent_items_list(Items, Refs, Attempts, Q_AckRef).
 
 %% only ack replayq when the last message is accepted by Kafka
-make_sent_items_list([LastItem], [#kpro_req{ref = Ref}], Attempts, Q_AckRef) ->
+make_sent_items_list([LastItem], [Ref], Attempts, Q_AckRef) ->
   [#{req_ref => Ref,
      q_items => [LastItem],
      q_ack_ref => Q_AckRef,
      attempts => Attempts
     }];
-make_sent_items_list([Item | Items], [#kpro_req{ref = Ref} | Reqs], Attempts, Q_AckRef) ->
+make_sent_items_list([Item | Items], [Ref | Refs], Attempts, Q_AckRef) ->
   [#{req_ref => Ref,
      q_items => [Item],
      q_ack_ref => ?no_queue_ack,
      attempts => Attempts
-    } | make_sent_items_list(Items, Reqs, Attempts, Q_AckRef)].
+    } | make_sent_items_list(Items, Refs, Attempts, Q_AckRef)].
+
+send_request(QueueItems, #{conn := Conn} = St) ->
+  #kpro_req{ref = Ref} = Req = make_request(QueueItems, St),
+  ok = request_async(Conn, Req),
+  {ok, Ref}.
 
 make_request(QueueItems,
              #{config := #{required_acks := RequiredAcks,
@@ -461,13 +466,11 @@ make_request(QueueItems,
                         }).
 
 resend_sent_reqs(#{sent_reqs := SentReqs,
-                   conn := Conn,
                    config := Config
                   } = St, Reason) ->
   F = fun(Sent, Acc) ->
           wolff_metrics:retried_inc(Config, 1),
-          {Reqs, NewSentList} = remake_requests(Sent, St, Reason),
-          lists:foreach(fun(Req) -> ok = request_async(Conn, Req) end, Reqs),
+          NewSentList = resend_requests(Sent, St, Reason),
           lists:reverse(NewSentList, Acc)
       end,
   NewSentReqs = lists:foldl(F, [], queue:to_list(SentReqs)),