rabbit_stream_queue: avoid double local pid query

When starting a new consumer. Also move some metrics
registration a bit later when we know we are likely to succeed.

Author: kjnilsson

deps/rabbit/src/rabbit_stream_queue.erl

@@ -240,7 +240,7 @@ consume(Q, Spec,
         #stream_client{filtering_supported = FilteringSupported} = QState0) when ?amqqueue_is_stream(Q) ->
     %% Messages should include the offset as a custom header.
     case check_queue_exists_in_local_node(Q) of
-        ok ->
+        {ok, LocalPid} ->
             #{no_ack := NoAck,
               channel_pid := ChPid,
               prefetch_count := ConsumerPrefetchCount,
@@ -253,21 +253,20 @@ consume(Q, Spec,
                 {error, _} = Err ->
                     Err;
                 {ok, OffsetSpec} ->
-                    _ = rabbit_stream_coordinator:register_local_member_listener(Q),
-                    rabbit_core_metrics:consumer_created(ChPid, ConsumerTag, ExclusiveConsume,
-                                                         not NoAck, QName,
-                                                         ConsumerPrefetchCount, false,
-                                                         up, Args),
-                    %% FIXME: reply needs to be sent before the stream begins sending
-                    %% really it should be sent by the stream queue process like classic queues
-                    %% do
-                    maybe_send_reply(ChPid, OkMsg),
                     FilterSpec = filter_spec(Args),
                     case {FilterSpec, FilteringSupported} of
                         {#{filter_spec := _}, false} ->
                             {protocol_error, precondition_failed, "Filtering is not supported", []};
                         _ ->
-                            begin_stream(QState0, ConsumerTag, OffsetSpec,
+                            QState = QState0#stream_client{local_pid = LocalPid},
+                            rabbit_core_metrics:consumer_created(ChPid, ConsumerTag, ExclusiveConsume,
+                                                                 not NoAck, QName,
+                                                                 ConsumerPrefetchCount, false,
+                                                                 up, Args),
+                            %% reply needs to be sent before the stream begins sending
+                            maybe_send_reply(ChPid, OkMsg),
+                            _ = rabbit_stream_coordinator:register_local_member_listener(Q),
+                            begin_stream(QState, ConsumerTag, OffsetSpec,
                                          ConsumerPrefetchCount, FilterSpec)
                     end
             end;
@@ -410,7 +409,8 @@ credit(QName, CTag, Credit, Drain, #stream_client{readers = Readers0,
             true ->
                 case Readers1 of
                     #{CTag := #stream{credit = Credit1} = Str2} ->
-                        {Readers0#{CTag => Str2#stream{credit = 0}}, [{send_drained, {CTag, Credit1}}]};
+                        {Readers0#{CTag => Str2#stream{credit = 0}},
+                         [{send_drained, {CTag, Credit1}}]};
                     _ ->
                         {Readers1, []}
                 end;
@@ -997,7 +997,7 @@ check_queue_exists_in_local_node(Q) ->
     #{name := StreamId} = amqqueue:get_type_state(Q),
     case rabbit_stream_coordinator:local_pid(StreamId) of
         {ok, Pid} when is_pid(Pid) ->
-            ok;
+            {ok, Pid};
         _ ->
             {protocol_error, precondition_failed,
              "queue '~ts' does not have a replica on the local node",