Fix mediabunny stale state on republish, stream close cleanup, and mid-stream subscriber keyframe delivery

Replace computeIfAbsent with compute in MediaBunnyStreamRegistry.subscribe()
to detect when a publisher has changed and detach the orphaned listener.

Add onClosed() default method to IStreamListener and call it from
ClientBroadcastStream.close() before clearing listeners. The mediabunny
listener flushes remaining buffers and notifies the registry to remove the
stale state and send a poison-pill to unblock waiting servlet threads.

Cache the last video keyframe fragment on StreamState so new mid-stream
subscribers receive init segment + keyframe fragment, preventing the
VideoDecoder key chunk error on the player side.

Co-Authored-By: Claude <noreply@anthropic.com>

Author: mondain

common/src/main/java/org/red5/server/api/stream/IStreamListener.java

@@ -23,6 +23,13 @@ public interface IStreamListener {
      * @param packet
      *            the packet received
      */
-    public void packetReceived(IBroadcastStream stream, IStreamPacket packet);
+    void packetReceived(IBroadcastStream stream, IStreamPacket packet);
+
+    /**
+     * Stream is closed, notify the listener and release all resources.
+     */
+    default void onClosed() {
+        // default implementation does nothing
+    }
 
 }

common/src/main/java/org/red5/server/stream/ClientBroadcastStream.java

@@ -236,8 +236,15 @@ public void close() {
             if (recordingListener != null) {
                 recordingListener.clear();
             }
-            // clear listeners
+            // notify and clear listeners
             if (!listeners.isEmpty()) {
+                for (IStreamListener listener : listeners) {
+                    try {
+                        listener.onClosed();
+                    } catch (Exception e) {
+                        log.warn("Error notifying listener onClosed: {}", listener, e);
+                    }
+                }
                 listeners.clear();
             }
             // deregister with jmx

server/src/main/java/org/red5/server/net/mediabunny/MediaBunnyServlet.java

@@ -117,6 +117,10 @@ private void streamQueue(AsyncContext asyncContext, MediaBunnyStreamRegistry.Str
             BlockingQueue<byte[]> queue = subscription.getQueue();
             while (true) {
                 byte[] chunk = queue.take();
+                if (chunk.length == 0) {
+                    log.debug("MediaBunny received end-of-stream signal");
+                    break;
+                }
                 chunkCount++;
                 if (!loggedFirstChunk) {
                     loggedFirstChunk = true;

server/src/main/java/org/red5/server/net/mediabunny/MediaBunnyStreamListener.java

@@ -155,6 +155,14 @@ public void packetReceived(IBroadcastStream stream, IStreamPacket packet) {
         }
     }
 
+    @Override
+    public void onClosed() {
+        log.info("Closing MediaBunnyStreamListener for stream {}", streamKey);
+        flushVideoBuffer();
+        flushAudioBuffer();
+        registry.onStreamClosed(streamKey);
+    }
+
     private void handleVideo(VideoData video) {
         if (video.getCodecId() != VideoCodec.AVC.getId()) {
             if (log.isDebugEnabled()) {
@@ -220,6 +228,7 @@ private void handleVideo(VideoData video) {
             } else if (!videoBuffer.samples.isEmpty()) {
                 flushVideoBuffer();
             }
+            videoBuffer.startsWithKeyframe = true;
         } else if (!videoKeyframeSeen) {
             return;
         }
@@ -403,16 +412,21 @@ private void flushBuffer(TrackBuffer buffer, long sequence, long trackId, long g
         if (buffer.samples.isEmpty()) {
             return;
         }
+        boolean keyframe = buffer.startsWithKeyframe;
         Fmp4FragmentBuilder.FragmentConfig config = new Fmp4FragmentBuilder.FragmentConfig().setSequenceNumber(sequence).setTrackId(trackId).setBaseDecodeTime(buffer.fragmentStartDecodeTime).setGroupId(groupId).setMediaType(mediaType).setMediaData(buffer.media.toByteArray()).setSamples(new ArrayList<>(buffer.samples));
         byte[] fragment;
         try {
             fragment = fragmentBuilder.buildFragment(config).serialize();
             patchBrandBox(fragment, "styp", "iso6");
             fragment = stripLeadingBox(fragment, "styp");
             if (log.isDebugEnabled()) {
-                log.debug("Built {} fragment for stream {} ({} bytes)", mediaType, streamKey, fragment.length);
+                log.debug("Built {} fragment for stream {} ({} bytes, keyframe={})", mediaType, streamKey, fragment.length, keyframe);
+            }
+            if (keyframe && mediaType == CmafFragment.MediaType.VIDEO) {
+                registry.onKeyframeFragment(streamKey, fragment);
+            } else {
+                registry.onFragment(streamKey, fragment);
             }
-            registry.onFragment(streamKey, fragment);
         } catch (IOException e) {
             log.warn("Failed to build fragment for stream: {}", streamKey, e);
         }
@@ -888,11 +902,14 @@ private static class TrackBuffer {
 
         private long fragmentStartDecodeTime;
 
+        private boolean startsWithKeyframe;
+
         private void reset() {
             media.reset();
             samples.clear();
             bufferedDuration = 0;
             fragmentStartDecodeTime = 0;
+            startsWithKeyframe = false;
         }
     }
 }

server/src/main/java/org/red5/server/net/mediabunny/MediaBunnyStreamRegistry.java

@@ -35,7 +35,18 @@ public StreamSubscription subscribe(IScope scope, String streamName) {
         log.debug("Subscribing to stream: {}", streamName);
         String key = buildKey(scope, streamName);
         log.debug("Subscriber key: {}", key);
-        StreamState state = streams.computeIfAbsent(key, id -> createState(scope, streamName));
+        StreamState state = streams.compute(key, (k, existing) -> {
+            IBroadcastScope bs = scope.getBroadcastScope(streamName);
+            IClientBroadcastStream currentStream = (bs != null) ? bs.getClientBroadcastStream() : null;
+            if (existing != null) {
+                if (currentStream != null && currentStream == existing.stream) {
+                    return existing; // same stream, reuse
+                }
+                log.info("Stream changed for {}, detaching old listener", k);
+                existing.detach();
+            }
+            return createState(scope, streamName);
+        });
         if (state == null) {
             throw new IllegalStateException("Stream not found: " + streamName);
         }
@@ -50,6 +61,10 @@ public StreamSubscription subscribe(IScope scope, String streamName) {
         if (initSegment != null) {
             queue.offer(initSegment);
         }
+        byte[] keyframe = state.keyframeFragment;
+        if (keyframe != null) {
+            queue.offer(keyframe);
+        }
         return new StreamSubscription(key, queue, this);
     }
 
@@ -66,6 +81,18 @@ public void unsubscribe(String key, BlockingQueue<byte[]> queue) {
         }
     }
 
+    void onStreamClosed(String key) {
+        StreamState state = streams.remove(key);
+        if (state != null) {
+            log.info("Stream closed for {}, removed state and notifying {} subscribers", key, state.subscribers.size());
+            // poison-pill empty array to unblock waiting subscribers
+            for (BlockingQueue<byte[]> queue : state.subscribers) {
+                queue.offer(new byte[0]);
+            }
+        }
+        pendingInitSegments.remove(key);
+    }
+
     void onInitSegment(String key, byte[] initSegment) {
         StreamState state = streams.get(key);
         if (state == null) {
@@ -79,6 +106,20 @@ void onInitSegment(String key, byte[] initSegment) {
         }
     }
 
+    void onKeyframeFragment(String key, byte[] fragment) {
+        StreamState state = streams.get(key);
+        if (state == null) {
+            return;
+        }
+        state.keyframeFragment = fragment;
+        if (log.isDebugEnabled()) {
+            log.debug("Dispatching keyframe fragment for {} to {} subscribers ({} bytes)", key, state.subscribers.size(), fragment.length);
+        }
+        for (BlockingQueue<byte[]> queue : state.subscribers) {
+            queue.offer(fragment);
+        }
+    }
+
     void onFragment(String key, byte[] fragment) {
         StreamState state = streams.get(key);
         if (state == null) {
@@ -132,6 +173,8 @@ static class StreamState {
 
         private volatile byte[] initSegment;
 
+        private volatile byte[] keyframeFragment;
+
         StreamState(IClientBroadcastStream stream, MediaBunnyStreamListener listener) {
             this.stream = stream;
             this.listener = listener;