MultiFromBlockingInputStream RC fix (#5061)

Signed-off-by: Daniel Kec <daniel.kec@oracle.com> Signed-off-by: Daniel Kec <daniel.kec@oracle.com>
helidon-io · Oct 3, 2022 · 45e0bcc · 45e0bcc
1 parent 50548e8
commit 45e0bcc
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 24 deletions.
diff --git a/common/reactive/src/main/java/io/helidon/common/reactive/MultiFromBlockingInputStream.java b/common/reactive/src/main/java/io/helidon/common/reactive/MultiFromBlockingInputStream.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020 Oracle and/or its affiliates.
+ * Copyright (c) 2020, 2022 Oracle and/or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -22,8 +22,8 @@
 import java.util.Objects;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Flow;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.IntSupplier;
 
 class MultiFromBlockingInputStream extends MultiFromInputStream {
@@ -68,9 +68,8 @@ public Multi<ByteBuffer> withByteBufferSize(final int bufferSize) {
     static final class InputStreamSubscription extends MultiFromInputStream.InputStreamSubscription {
 
         private final ExecutorService executorService;
-        private final LinkedBlockingQueue<Runnable> submitQueue = new LinkedBlockingQueue<>();
+        private Lock lck = new ReentrantLock();
 
-        private final AtomicBoolean draining = new AtomicBoolean(false);
 
         InputStreamSubscription(Flow.Subscriber<? super ByteBuffer> downstream,
                                 InputStream inputStream,
@@ -81,24 +80,14 @@ static final class InputStreamSubscription extends MultiFromInputStream.InputStr
         }
 
         protected void trySubmit(long n) {
-            submitQueue.add(() -> {
-                submit(n);
-                drainSubmitQueue();
-            });
-            drainSubmitQueue();
-        }
-
-        private void drainSubmitQueue() {
-            if (!draining.getAndSet(true)) {
+            executorService.submit(() -> {
                 try {
-                    Runnable job = submitQueue.poll();
-                    if (job != null) {
-                        executorService.submit(job);
-                    }
+                    lck.lock();
+                    submit(n);
                 } finally {
-                    draining.set(false);
+                    lck.unlock();
                 }
-            }
+            });
         }
     }
 }
diff --git a/...reams/src/test/java/io/helidon/common/reactive/MultiFromNotTrustedInputStreamTckTest.java b/...reams/src/test/java/io/helidon/common/reactive/MultiFromNotTrustedInputStreamTckTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020, 2021 Oracle and/or its affiliates.
+ * Copyright (c) 2020, 2022 Oracle and/or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -61,5 +61,4 @@ protected Multi<ByteBuffer> getPublisher(InputStream is) {
                 .byteBufferSize(BUFFER_SIZE)
                 .build();
     }
-
 }
diff --git a/...-streams/src/test/java/io/helidon/common/reactive/MultiFromTrustedInputStreamTckTest.java b/...-streams/src/test/java/io/helidon/common/reactive/MultiFromTrustedInputStreamTckTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2021 Oracle and/or its affiliates.
+ * Copyright (c) 2021, 2022 Oracle and/or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -31,7 +31,7 @@ public class MultiFromTrustedInputStreamTckTest extends FlowPublisherVerificatio
     static final int BUFFER_SIZE = 4;
 
     public MultiFromTrustedInputStreamTckTest() {
-        super(new TestEnvironment(50));
+        super(new TestEnvironment(500));
     }
 
     @Override