Do not change state for local member listener in v1

In stream coordinator. A v2 node, even if running v1, can
call the function to register a local member when an AMQP listener
is registered, so this commit makes such a call a no-op.

References #4133

(cherry picked from commit 6133416)
(cherry picked from commit aed576f)

Author: acogoluegnes

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -442,18 +442,22 @@ apply(#{machine_version := MachineVersion} = Meta, {down, Pid, Reason} = Cmd,
     end;
 apply(#{machine_version := MachineVersion} = Meta,
       {register_listener, #{pid := Pid,
-                            stream_id := StreamId}},
+                            stream_id := StreamId} = Args},
       #?MODULE{streams = Streams,
                monitors = Monitors0} = State0) when MachineVersion =< 1 ->
-    case Streams of
-        #{StreamId := #stream{listeners = Listeners0} = Stream0} ->
+    Type = maps:get(type, Args, leader),
+    case {Streams, Type} of
+        {#{StreamId := #stream{listeners = Listeners0} = Stream0}, leader} ->
             Stream1 = Stream0#stream{listeners = maps:put(Pid, undefined, Listeners0)},
             {Stream, Effects} = eval_listeners(MachineVersion, Stream1, []),
             Monitors = maps:put(Pid, {StreamId, listener}, Monitors0),
             return(Meta,
                    State0#?MODULE{streams = maps:put(StreamId, Stream, Streams),
                                   monitors = Monitors}, ok,
                    [{monitor, process, Pid} | Effects]);
+        {#{StreamId := _Stream}, local_member} ->
+            %% registering a local member listener does not change the state in v1
+            return(Meta, State0, ok, []);
         _ ->
             return(Meta, State0, stream_not_found, [])
     end;