Merge pull request #1989 from jponge/fix/1988

fix: memory leak in UniMemoizeOp

Author: jponge

implementation/src/main/java/io/smallrye/mutiny/operators/uni/UniMemoizeOp.java

@@ -3,8 +3,8 @@
 import static io.smallrye.mutiny.helpers.ParameterValidation.nonNull;
 
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
-import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.BooleanSupplier;
 
@@ -33,7 +33,7 @@ private enum State {
 
     private final ReentrantLock internalLock = new ReentrantLock();
 
-    private final ConcurrentLinkedQueue<UniSubscriber<? super I>> awaiters = new ConcurrentLinkedQueue<>();
+    private final List<UniSubscriber<? super I>> awaiters = new ArrayList<>();
 
     private Object cachedResult = null;
 
@@ -118,12 +118,17 @@ public void onItem(I item) {
     }
 
     private List<UniSubscriber<? super I>> gatherAwaiters() {
-        return new ArrayList<>(awaiters);
+        ArrayList<UniSubscriber<? super I>> copy = new ArrayList<>(awaiters);
+        awaiters.clear();
+        return copy;
     }
 
     private void notifyAwaiters(List<UniSubscriber<? super I>> toNotify, Object result) {
-        for (UniSubscriber<? super I> awaiter : toNotify) {
+        Iterator<UniSubscriber<? super I>> iterator = toNotify.iterator();
+        while (iterator.hasNext()) {
+            UniSubscriber<? super I> awaiter = iterator.next();
             forwardTo(awaiter, result);
+            iterator.remove();
         }
     }
 
@@ -166,7 +171,12 @@ private class MemoizedSubscription implements UniSubscription {
 
         @Override
         public void cancel() {
-            awaiters.remove(subscriber);
+            internalLock.lock();
+            try {
+                awaiters.remove(subscriber);
+            } finally {
+                internalLock.unlock();
+            }
         }
     }
 }