wip

Author: kjnilsson

deps/rabbit/src/rabbit_fifo.erl

@@ -2066,6 +2066,7 @@ delivery_effect(ConsumerKey, Msgs,
     {RaftIdxs, _Num} = lists:foldr(fun ({_, ?MSG(I, _)}, {Acc, N}) ->
                                            {[I | Acc], N+1}
                                    end, {[], 0}, Msgs),
+    rabbit_log:debug("DELIVERY EFFECT created for ~s ~p!", [CTag, RaftIdxs]),
     {log_ext, RaftIdxs,
      fun
          % (Commands)
@@ -2079,15 +2080,9 @@ delivery_effect(ConsumerKey, Msgs,
          %                      length(Commands)]),
          %     [];
          (ReadPlan) ->
-             % Fun = fun (Flru0) ->
-             %               {Entries, Flru} = ra_log:execute_read(ReadPlan, Flru0),
-             %               %% pretend entries is a map
-             %               {lists:map(fun ({MsgId, ?MSG(Idx,  Header)}) ->
-             %                                 {_, _, Cmd} = maps:get(Idx, Entries),
-             %                                 %% hacky
-             %                                 {MsgId, {Header, get_msg(element(3, Cmd))}}
-             %                         end, Msgs), Flru}
-             %       end,
+             rabbit_log:debug("READPLAN created for ~s ~p!", [CTag, RaftIdxs]),
+             %% TODO: check if CPid is local or not
+             %% TODO: could consider introducing a leader local proxy process
              [{send_msg, CPid, {delivery, CTag, ReadPlan, Msgs},
                ?DELIVERY_SEND_MSG_OPTS}]
      end,
@@ -3026,10 +3021,10 @@ incr_msg(Msg0, DelFailed, Anns) ->
     end.
 
 exec_read(Flru0, ReadPlan, Msgs) ->
-    {Entries, Flru} = ra_log:execute_read(ReadPlan, Flru0),
+    {Entries, Flru} = ra:execute_read_plan(ReadPlan, Flru0),
     %% pretend entries is a map
     {lists:map(fun ({MsgId, ?MSG(Idx,  Header)}) ->
-                       {_, _, Cmd} = maps:get(Idx, Entries),
+                       {_, _, Cmd, _} = maps:get(Idx, Entries),
                        %% hacky
-                       {MsgId, {Header, get_msg(element(3, Cmd))}}
+                       {MsgId, {Header, get_msg(Cmd)}}
                end, Msgs), Flru}.

deps/rabbit/src/rabbit_fifo_client.erl

@@ -168,7 +168,7 @@ enqueue(QName, Correlation, Msg,
 %% @param QueueName Name of the queue.
 %% @param Msg an arbitrary erlang term representing the message.
 %% @param State the current {@module} state.
-%% @returns
+%% @return's
 %% `{ok, State, Actions}' if the command was successfully sent.
 %% {@module} assigns a sequence number to every raft command it issues. The
 %% SequenceNumber can be correlated to the applied sequence numbers returned

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -3632,6 +3632,7 @@ receive_and_ack(Ch) ->
             amqp_channel:cast(Ch, #'basic.ack'{delivery_tag = DeliveryTag,
                                                multiple = false})
     after 5000 ->
+              flush(1),
               ct:fail("receive_and_ack timed out", [])
     end.