Merge pull request #1539 from jponge/backport/1536

fix(BroadcastProcessor): prevent memory leaks by clearing active subscriptions
smallrye · Mar 6, 2024 · 8a1c8b3 · 8a1c8b3
2 parents 44d5999 + 7bd37e4
commit 8a1c8b3
Showing 1 changed file with 13 additions and 8 deletions.
diff --git a/...ation/src/main/java/io/smallrye/mutiny/operators/multi/processors/BroadcastProcessor.java b/...ation/src/main/java/io/smallrye/mutiny/operators/multi/processors/BroadcastProcessor.java
@@ -133,35 +133,40 @@ public void onSubscribe(Subscription subscription) {
     @Override
     public void onNext(T item) {
         ParameterValidation.nonNullNpe(item, "item");
-        for (BroadcastSubscription<T> s : subscribers.get()) {
-            s.onNext(item);
+        List<BroadcastSubscription<T>> subscriptions = subscribers.get();
+        if (subscriptions != TERMINATED) {
+            for (BroadcastSubscription<T> s : subscriptions) {
+                s.onNext(item);
+            }
         }
     }
 
     @SuppressWarnings({ "unchecked", "rawtypes" })
     @Override
     public void onError(Throwable failure) {
         ParameterValidation.nonNullNpe(failure, "failure");
-        if (subscribers.get() == TERMINATED) {
+        List<BroadcastSubscription<?>> subscriptions = subscribers.getAndSet((List) TERMINATED);
+        if (subscriptions == TERMINATED) {
             return;
         }
         this.failure = failure;
-        List<BroadcastSubscription<?>> andSet = subscribers.getAndSet((List) TERMINATED);
-        for (BroadcastSubscription<?> s : andSet) {
+        for (BroadcastSubscription<?> s : subscriptions) {
             s.onError(failure);
         }
+        subscriptions.clear();
     }
 
     @SuppressWarnings({ "unchecked", "rawtypes" })
     @Override
     public void onComplete() {
-        if (subscribers.get() == TERMINATED) {
+        List<BroadcastSubscription<?>> subscriptions = subscribers.getAndSet((List) TERMINATED);
+        if (subscriptions == TERMINATED) {
             return;
         }
-        List<BroadcastSubscription<?>> andSet = subscribers.getAndSet((List) TERMINATED);
-        for (BroadcastSubscription<?> s : andSet) {
+        for (BroadcastSubscription<?> s : subscriptions) {
             s.onComplete();
         }
+        subscriptions.clear();
     }
 
     /**