Fix completable queue and clean-up of messaging tests

microprofile/messaging/src/main/java/io/helidon/microprofile/messaging/CompletableQueue.java

@@ -21,6 +21,7 @@
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.BiConsumer;
 
@@ -32,10 +33,11 @@
  */
 class CompletableQueue<T> {
 
-    private static final long MAX_QUEUE_SIZE = 1024;
+    private static final long MAX_QUEUE_SIZE = 2048;
+    private static final long BACK_PRESSURE_LIMIT = MAX_QUEUE_SIZE - (MAX_QUEUE_SIZE >> 2);
     private final ReentrantLock queueLock = new ReentrantLock();
     private final LinkedList<Item<T>> queue = new LinkedList<>();
-    private long size = 0;
+    private final AtomicLong size = new AtomicLong();
     private volatile BiConsumer<Item<T>, ? super Throwable> onEachComplete;
 
     private CompletableQueue(final BiConsumer<Item<T>, ? super Throwable> onEachComplete) {
@@ -62,6 +64,15 @@ void onEachComplete(BiConsumer<Item<T>, ? super Throwable> onEachComplete) {
         tryFlush();
     }
 
+    /**
+     * If limit for safe prefetch is reached applying back-pressure is advised.
+     *
+     * @return true if limit reached
+     */
+    boolean isBackPressureLimitReached() {
+        return BACK_PRESSURE_LIMIT <= size.get();
+    }
+
     /**
      * Add completable to the queue, if completed {@code onEachComplete} is invoked only after all older completables
      * invoked {@code onEachComplete}.
@@ -74,7 +85,7 @@ void add(CompletableFuture<T> future, Object metadata) {
         try {
             queueLock.lock();
             queue.add(Item.create(future, metadata));
-            if (++size > MAX_QUEUE_SIZE) {
+            if (size.incrementAndGet() > MAX_QUEUE_SIZE) {
                 throw ExceptionUtils.createCompletableQueueOverflow(MAX_QUEUE_SIZE);
             }
             future.whenComplete((t, u) -> tryFlush());
@@ -103,10 +114,11 @@ private void tryFlush() {
                 while (!queue.isEmpty() && queue.getFirst().getCompletableFuture().isDone()) {
                     var item = queue.poll();
                     if (Objects.isNull(item)) return;
-                    size--;
+                    size.decrementAndGet();
                     item.setValue(item.getCompletableFuture().get());
                     onEachComplete.accept(item, null);
                 }
+                return;
             } catch (InterruptedException | ExecutionException e) {
                 onEachComplete.accept(null, e);
             }

microprofile/messaging/src/main/java/io/helidon/microprofile/messaging/InternalProcessor.java

@@ -20,6 +20,7 @@
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 
 import org.eclipse.microprofile.reactive.messaging.Acknowledgment;
@@ -91,13 +92,20 @@ public void onNext(final Object incomingValue) {
                 } else {
                     publisherBuilder = (PublisherBuilder<?>) processedValue;
                 }
-                publisherBuilder.forEach(subVal -> {
-                    if (!completionStageAwait(incomingValue, subVal)) {
-                        subscriber.onNext(postProcess(incomingValue, subVal));
-                    }
-                }).run();
+                publisherBuilder
+                        .flatMapCompletionStage(o -> {
+                            if (o instanceof CompletionStage) {
+                                return (CompletionStage<?>) o;
+                            } else {
+                                return CompletableFuture.completedStage(o);
+                            }
+                        })
+                        .map(o -> postProcess(incomingValue, o))
+                        .to(subscriber)
+                        .run();
             } else {
                 if (!completionStageAwait(incomingValue, processedValue)) {
+                    //FIXME: apply back-pressure instead of buffering
                     subscriber.onNext(postProcess(incomingValue, processedValue));
                 }
             }

microprofile/messaging/src/main/java/io/helidon/microprofile/messaging/InternalPublisher.java

@@ -23,6 +23,8 @@
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import io.helidon.common.reactive.RequestedCounter;
+
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
@@ -37,13 +39,15 @@ class InternalPublisher implements Publisher<Object>, Subscription {
     private final Object beanInstance;
     private final AtomicBoolean closed = new AtomicBoolean(false);
     private final CompletableQueue<Object> completableQueue;
+    private final RequestedCounter requestedCounter = new RequestedCounter();
 
     InternalPublisher(Method method, Object beanInstance) {
         this.method = method;
         this.beanInstance = beanInstance;
         completableQueue = CompletableQueue.create((o, throwable) -> {
             if (Objects.isNull(throwable)) {
                 subscriber.onNext(o.getValue());
+                trySubmit();
             } else {
                 subscriber.onError(throwable);
             }
@@ -57,10 +61,15 @@ public void subscribe(Subscriber<? super Object> s) {
     }
 
     @Override
-    @SuppressWarnings("unchecked")
     public void request(long n) {
+        requestedCounter.increment(n, subscriber::onError);
+        trySubmit();
+    }
+
+    @SuppressWarnings("unchecked")
+    private void trySubmit() {
         try {
-            for (long i = 0; i < n && !closed.get(); i++) {
+            while (!completableQueue.isBackPressureLimitReached() && requestedCounter.tryDecrement() && !closed.get()) {
                 Object result = method.invoke(beanInstance);
                 if (result instanceof CompletionStage) {
                     CompletionStage<Object> completionStage = (CompletionStage<Object>) result;

microprofile/messaging/src/main/java/io/helidon/microprofile/messaging/InternalSubscriber.java

@@ -39,8 +39,7 @@ class InternalSubscriber implements Subscriber<Object> {
     @Override
     public void onSubscribe(Subscription s) {
         subscription = s;
-        // request one by one
-        subscription.request(1);
+        subscription.request(Long.MAX_VALUE);
     }
 
     @Override
@@ -51,7 +50,6 @@ public void onNext(Object message) {
             Object preProcessedMessage = preProcess(message, paramType);
             Object methodResult = method.invoke(incomingMethod.getBeanInstance(), preProcessedMessage);
             postProcess(message, methodResult);
-            subscription.request(1);
         } catch (Exception e) {
             // Notify publisher to stop sending
             subscription.cancel();

microprofile/messaging/src/test/java/io/helidon/microprofile/messaging/AbstractCDITest.java

@@ -23,14 +23,13 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.logging.LogManager;
-import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import javax.enterprise.inject.se.SeContainer;
 import javax.enterprise.inject.se.SeContainerInitializer;
@@ -177,19 +176,24 @@ public Optional<? extends Class<? extends Throwable>> getExpectedThrowable() {
         }
 
         @SuppressWarnings("unchecked")
-        public List<Class<? extends CountableTestBean>> getCountableBeanClasses() {
+        public Stream<Class<? extends AsyncTestBean>> getAsyncBeanClasses() {
+            return Arrays.stream(clazzes)
+                    .filter(AsyncTestBean.class::isAssignableFrom)
+                    .map(c -> (Class<? extends AsyncTestBean>) c);
+        }
+
+        @SuppressWarnings("unchecked")
+        public Stream<Class<? extends CountableTestBean>> getCountableBeanClasses() {
             return Arrays.stream(clazzes)
                     .filter(CountableTestBean.class::isAssignableFrom)
-                    .map(c -> (Class<? extends CountableTestBean>) c)
-                    .collect(Collectors.toList());
+                    .map(c -> (Class<? extends CountableTestBean>) c);
         }
 
         @SuppressWarnings("unchecked")
-        public List<Class<? extends AssertableTestBean>> getCompletableBeanClasses() {
+        public Stream<Class<? extends AssertableTestBean>> getCompletableBeanClasses() {
             return Arrays.stream(clazzes)
                     .filter(AssertableTestBean.class::isAssignableFrom)
-                    .map(c -> (Class<? extends AssertableTestBean>) c)
-                    .collect(Collectors.toList());
+                    .map(c -> (Class<? extends AssertableTestBean>) c);
         }
     }
 }

microprofile/messaging/src/test/java/io/helidon/microprofile/messaging/AssertableTestBean.java

@@ -20,10 +20,57 @@
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import org.hamcrest.Matcher;
 
 public interface AssertableTestBean {
 
     Set<String> TEST_DATA = new HashSet<>(Arrays.asList("teST1", "TEst2", "tESt3"));
 
     void assertValid();
+
+    default void await(String msg, CountDownLatch countDownLatch) {
+        try {
+            assertThat(msg + composeOrigin(), countDownLatch.await(500, TimeUnit.MILLISECONDS));
+        } catch (InterruptedException e) {
+            fail(msg + composeOrigin(), e);
+        }
+    }
+
+    default <T> T await(String msg, CompletableFuture<T> cs) {
+        try {
+            return cs.get(500, TimeUnit.MILLISECONDS);
+        } catch (InterruptedException | ExecutionException | TimeoutException e) {
+            return fail(msg + composeOrigin(), e);
+        }
+    }
+
+    default <T> void assertWithOrigin(String msg, boolean assertion) {
+        assertThat(msg + composeOrigin(), assertion);
+    }
+
+    default <T> void assertWithOrigin(String msg, T actual, Matcher<? super T> matcher) {
+        assertThat(msg + composeOrigin(), actual, matcher);
+    }
+
+    default void failWithOrigin(String msg) {
+        fail(msg + composeOrigin());
+    }
+
+    default void failWithOrigin(String msg, Throwable e) {
+        fail(msg + composeOrigin(), e);
+    }
+
+    default String composeOrigin() {
+        StackTraceElement e = Thread.currentThread().getStackTrace()[3];
+        return " at " + e.toString();
+    }
 }

microprofile/messaging/src/test/java/io/helidon/microprofile/messaging/AsyncTestBean.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright (c)  2020 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.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package io.helidon.microprofile.messaging;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.fail;
+
+public interface AsyncTestBean {
+
+    default ExecutorService createExecutor() {
+        return Executors.newSingleThreadExecutor();
+    }
+
+    default void awaitShutdown(ExecutorService executor) {
+        executor.shutdown();
+        var msg = "Executor of test bean " + this.getClass().getSimpleName() + " didn't shutdown in time.";
+        try {
+            assertThat(msg, executor.awaitTermination(500, TimeUnit.MILLISECONDS));
+        } catch (InterruptedException e) {
+            fail(msg, e);
+        }
+    }
+
+    void tearDown();
+}

microprofile/messaging/src/test/java/io/helidon/microprofile/messaging/connector/ConnectedBean.java

@@ -26,6 +26,10 @@
 
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+/**
+ * This test is modified version of official tck test in version 1.0
+ * https://github.com/eclipse/microprofile-reactive-messaging
+ */
 @ApplicationScoped
 public class ConnectedBean {
 

microprofile/messaging/src/test/java/io/helidon/microprofile/messaging/connector/ConnectedOnlyProcessorBean.java

@@ -22,6 +22,10 @@
 
 import javax.enterprise.context.ApplicationScoped;
 
+/**
+ * This test is modified version of official tck test in version 1.0
+ * https://github.com/eclipse/microprofile-reactive-messaging
+ */
 @ApplicationScoped
 public class ConnectedOnlyProcessorBean {
 

microprofile/messaging/src/test/java/io/helidon/microprofile/messaging/connector/ConnectedProcessorBean.java

@@ -29,6 +29,10 @@
 
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+/**
+ * This test is modified version of official tck test in version 1.0
+ * https://github.com/eclipse/microprofile-reactive-messaging
+ */
 @ApplicationScoped
 public class ConnectedProcessorBean {
 

microprofile/messaging/src/test/java/io/helidon/microprofile/messaging/connector/IterableConnector.java

@@ -16,6 +16,13 @@
 
 package io.helidon.microprofile.messaging.connector;
 
+import java.util.Arrays;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.stream.Collectors;
+
+import javax.enterprise.context.ApplicationScoped;
+
 import org.eclipse.microprofile.config.Config;
 import org.eclipse.microprofile.reactive.messaging.Message;
 import org.eclipse.microprofile.reactive.messaging.spi.Connector;
@@ -25,15 +32,10 @@
 import org.eclipse.microprofile.reactive.streams.operators.ReactiveStreams;
 import org.eclipse.microprofile.reactive.streams.operators.SubscriberBuilder;
 
-import javax.enterprise.context.ApplicationScoped;
-
-import java.util.Arrays;
-import java.util.Set;
-import java.util.concurrent.CountDownLatch;
-import java.util.stream.Collectors;
-
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
+/**
+ * This test is modified version of official tck test in version 1.0
+ * https://github.com/eclipse/microprofile-reactive-messaging
+ */
 @ApplicationScoped
 @Connector("iterable-connector")
 public class IterableConnector implements IncomingConnectorFactory, OutgoingConnectorFactory {
@@ -45,15 +47,17 @@ public class IterableConnector implements IncomingConnectorFactory, OutgoingConn
 
     @Override
     public PublisherBuilder<? extends Message<?>> getPublisherBuilder(Config config) {
-        //TODO: use ReactiveStreams.of().map when engine is ready(supports more than one stage)
-        return ReactiveStreams.fromIterable(Arrays.stream(TEST_DATA).map(Message::of).collect(Collectors.toSet()));
+        return ReactiveStreams.of(TEST_DATA).map(Message::of);
     }
 
     @Override
     public SubscriberBuilder<? extends Message<?>, Void> getSubscriberBuilder(Config config) {
-        return ReactiveStreams.<Message<?>>builder().forEach(m -> {
-            assertTrue(PROCESSED_DATA.contains(m.getPayload()));
-            LATCH.countDown();
-        });
+        return ReactiveStreams.<Message<?>>builder()
+                .map(Message::getPayload)
+                .forEach(p -> {
+                    if (PROCESSED_DATA.contains(p)) {
+                        LATCH.countDown();
+                    }
+                });
     }
 }