Messaging 3.0

common/reactive/src/main/java/io/helidon/common/reactive/BufferedEmittingPublisher.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.
@@ -16,6 +16,7 @@
 
 package io.helidon.common.reactive;
 
+import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Flow;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -34,7 +35,7 @@
  */
 public class BufferedEmittingPublisher<T> implements Flow.Publisher<T> {
 
-    private final ConcurrentLinkedQueue<T> buffer = new ConcurrentLinkedQueue<>();
+    private final Queue<T> buffer;
     private volatile Throwable error;
     private BiConsumer<Long, Long> requestCallback = null;
     private Consumer<? super T> onEmitCallback = null;
@@ -60,17 +61,12 @@ public class BufferedEmittingPublisher<T> implements Flow.Publisher<T> {
     //         against a completion (isCancelled() and isComplete() are both true)
     private boolean cancelled;
 
-    protected BufferedEmittingPublisher() {
+    protected BufferedEmittingPublisher(Queue<T> queue) {
+        buffer = queue;
     }
 
-    /**
-     * Create new {@link BufferedEmittingPublisher}.
-     *
-     * @param <T> type of emitted item
-     * @return new instance of BufferedEmittingPublisher
-     */
-    public static <T> BufferedEmittingPublisher<T> create() {
-        return new BufferedEmittingPublisher<T>();
+    protected BufferedEmittingPublisher() {
+        buffer = new ConcurrentLinkedQueue<>();
     }
 
     @Override
@@ -488,4 +484,60 @@ private void drain() {
             throw ise;
         }
     }
+
+    /**
+     * Create new {@link BufferedEmittingPublisher}.
+     *
+     * @param <T> type of emitted item
+     * @return new instance of BufferedEmittingPublisher
+     */
+    public static <T> BufferedEmittingPublisher<T> create() {
+        BufferedEmittingPublisher<T> bep = BufferedEmittingPublisher.<T>builder().build();
+        return bep;
+    }
+
+    /**
+     * Create new builder for BufferedEmittingPublisher.
+     *
+     * @param <T> type of the expected payload
+     * @return new builder
+     */
+    public static <T> BufferedEmittingPublisher.Builder<T> builder() {
+        return new BufferedEmittingPublisher.Builder<>();
+    }
+
+    /**
+     * Fluent API builder to create {@link io.helidon.common.reactive.BufferedEmittingPublisher}.
+     *
+     * @param <T> type of the expected payload
+     */
+    public static class Builder<T> implements io.helidon.common.Builder<BufferedEmittingPublisher.Builder<T>, BufferedEmittingPublisher<T>> {
+
+        private Queue<T> queue;
+
+        private Builder() {
+        }
+
+        /**
+         * Set up custom buffer queue implementation for the emitter to use.
+         *
+         * @param queue to be used as a buffer
+         * @return this builder
+         */
+        public BufferedEmittingPublisher.Builder<T> buffer(Queue<T> queue) {
+            this.queue = queue;
+            return this;
+        }
+
+        @Override
+        public BufferedEmittingPublisher<T> build() {
+            BufferedEmittingPublisher<T> bep;
+            if (queue != null) {
+                bep = new BufferedEmittingPublisher<>(queue);
+            } else {
+                bep = new BufferedEmittingPublisher<>();
+            }
+            return bep;
+        }
+    }
 }

dependencies/pom.xml

@@ -114,9 +114,9 @@
         <version.lib.microprofile-jwt>2.0</version.lib.microprofile-jwt>
         <version.lib.microprofile-metrics-api>4.0</version.lib.microprofile-metrics-api>
         <version.lib.microprofile-openapi-api>3.0</version.lib.microprofile-openapi-api>
-        <version.lib.microprofile-reactive-messaging-api>2.0.1</version.lib.microprofile-reactive-messaging-api>
-        <version.lib.microprofile-reactive-streams-operators-api>2.0</version.lib.microprofile-reactive-streams-operators-api>
-        <version.lib.microprofile-reactive-streams-operators-core>2.0</version.lib.microprofile-reactive-streams-operators-core>
+        <version.lib.microprofile-reactive-messaging-api>3.0-RC2</version.lib.microprofile-reactive-messaging-api>
+        <version.lib.microprofile-reactive-streams-operators-api>3.0-RC1</version.lib.microprofile-reactive-streams-operators-api>
+        <version.lib.microprofile-reactive-streams-operators-core>3.0-RC1</version.lib.microprofile-reactive-streams-operators-core>
         <version.lib.microprofile-rest-client>3.0</version.lib.microprofile-rest-client>
         <version.lib.microprofile-tracing>3.0</version.lib.microprofile-tracing>
         <version.lib.microprofile-lra-api>2.0-RC1</version.lib.microprofile-lra-api>

microprofile/messaging/core/src/main/java/io/helidon/microprofile/messaging/AckCtx.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright (c) 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.
+ * 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.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.eclipse.microprofile.reactive.messaging.Acknowledgment;
+import org.eclipse.microprofile.reactive.messaging.Message;
+
+class AckCtx {
+    private final AbstractMessagingMethod m;
+    private final Message<?> msg;
+    private final AtomicBoolean acked = new AtomicBoolean();
+
+    private AckCtx(AbstractMessagingMethod m, Message<?> msg) {
+        this.m = m;
+        this.msg = msg;
+    }
+
+    static AckCtx create(AbstractMessagingMethod m, Message<?> msg) {
+        return new AckCtx(m, msg);
+    }
+
+    CompletionStage<Void> preAck() {
+        if (m.getAckStrategy().equals(Acknowledgment.Strategy.PRE_PROCESSING) && !acked.getAndSet(true)) {
+            return msg.ack();
+        }
+        return CompletableFuture.completedStage(null);
+    }
+
+    CompletionStage<Void> postNack(Throwable t) {
+        if (m.getAckStrategy().equals(Acknowledgment.Strategy.POST_PROCESSING) && !acked.getAndSet(true)) {
+            return msg.nack(t);
+        }
+        return CompletableFuture.completedStage(null);
+    }
+
+    CompletionStage<Void> postAck() {
+        if (m.getAckStrategy().equals(Acknowledgment.Strategy.POST_PROCESSING) && !acked.getAndSet(true)) {
+            return msg.ack();
+        }
+        return CompletableFuture.completedStage(null);
+    }
+}

microprofile/messaging/core/src/main/java/io/helidon/microprofile/messaging/BufferedEmitter.java

@@ -0,0 +1,119 @@
+/*
+ * Copyright (c) 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.
+ * 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.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+
+import io.helidon.common.reactive.BufferedEmittingPublisher;
+
+import org.eclipse.microprofile.reactive.messaging.Message;
+import org.eclipse.microprofile.reactive.messaging.OnOverflow;
+import org.reactivestreams.FlowAdapters;
+import org.reactivestreams.Publisher;
+
+/**
+ * Emitter used for {@link org.eclipse.microprofile.reactive.messaging.OnOverflow.Strategy#BUFFER}.
+ */
+class BufferedEmitter extends OutgoingEmitter {
+
+    private final BufferedEmittingPublisher<Object> bep = BufferedEmittingPublisher.create();
+
+    BufferedEmitter(String channelName, String fieldName, OnOverflow onOverflow) {
+        super(channelName, fieldName, onOverflow);
+    }
+
+    @Override
+    @SuppressWarnings("rawtypes")
+    public CompletionStage<Void> send(Object p) {
+        try {
+            lock().lock();
+            validate(p);
+            CompletableFuture<Void> acked = new CompletableFuture<>();
+            this.send(MessageUtils.create(p, acked));
+            return acked;
+        } finally {
+            lock().unlock();
+        }
+    }
+
+    @Override
+    public <M extends Message<? extends Object>> void send(M m) {
+        try {
+            lock().lock();
+            validate(m);
+            bep.emit(m);
+        } finally {
+            lock().unlock();
+        }
+    }
+
+    @Override
+    public void complete() {
+        try {
+            lock().lock();
+            super.complete();
+            bep.complete();
+        } finally {
+            lock().unlock();
+        }
+    }
+
+    @Override
+    public void error(Exception e) {
+        try {
+            lock().lock();
+            super.error(e);
+            bep.fail(e);
+        } finally {
+            lock().unlock();
+        }
+    }
+
+    @Override
+    public boolean isCancelled() {
+        try {
+            lock().lock();
+            return bep.isCancelled() || bep.isCompleted();
+        } finally {
+            lock().unlock();
+        }
+    }
+
+    @Override
+    public boolean hasRequests() {
+        return bep.hasRequests();
+    }
+
+    @Override
+    public Publisher<?> getPublisher() {
+        return FlowAdapters.toPublisher(bep);
+    }
+
+    @Override
+    public void validate(Object payload) {
+        super.validate(payload);
+        if (getOverflowStrategy().equals(OnOverflow.Strategy.BUFFER)) {
+            int bufferSize = bep.bufferSize();
+            if (getBufferLimit() > 0 && bufferSize > getBufferLimit()) {
+                RuntimeException ex = new IllegalStateException("Emitter buffer overflow");
+                throw ex;
+            }
+        }
+    }
+}