MP Reactive Messaging impl

bom/pom.xml

@@ -626,11 +626,18 @@
 
             <!-- Reactive Streams Operators -->
             <dependency>
-                <groupId>io.helidon.microprofile</groupId>
+                <groupId>io.helidon.microprofile.reactive-streams</groupId>
                 <artifactId>helidon-microprofile-reactive-streams</artifactId>
                 <version>${helidon.version}</version>
             </dependency>
 
+            <!-- Reactive Messaging -->
+            <dependency>
+                <groupId>io.helidon.microprofile.messaging</groupId>
+                <artifactId>helidon-microprofile-messaging</artifactId>
+                <version>${helidon.version}</version>
+            </dependency>
+
             <!-- integrations -->
             <dependency>
                 <groupId>io.helidon.serviceconfiguration</groupId>

common/common/src/main/java/io/helidon/common/Errors.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2017, 2019 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2017, 2020 Oracle and/or its affiliates. All rights reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,6 +21,7 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.logging.Logger;
+import java.util.stream.Collectors;
 
 /**
  * Errors utility used to file processing messages (e.g. validation, provider, resource building errors, hint).
@@ -156,6 +157,13 @@ public boolean log(Logger logger) {
         return true;
     }
 
+    @Override
+    public String toString() {
+        return this.stream()
+                .map(ErrorMessage::toString)
+                .collect(Collectors.joining("\n"));
+    }
+
     /**
      * Check if these messages are a valid result.
      *

dependencies/pom.xml

@@ -81,6 +81,9 @@
         <version.lib.microprofile-fault-tolerance-api>2.0.2</version.lib.microprofile-fault-tolerance-api>
         <version.lib.microprofile-tracing>1.3.1</version.lib.microprofile-tracing>
         <version.lib.microprofile-rest-client>1.3.3</version.lib.microprofile-rest-client>
+        <version.lib.microprofile-reactive-messaging-api>1.0</version.lib.microprofile-reactive-messaging-api>
+        <version.lib.microprofile-reactive-streams-operators-api>1.0.1</version.lib.microprofile-reactive-streams-operators-api>
+        <version.lib.microprofile-reactive-streams-operators-core>1.0.1</version.lib.microprofile-reactive-streams-operators-core>
         <version.lib.mockito>2.23.4</version.lib.mockito>
         <version.lib.mongodb.reactivestreams>1.11.0</version.lib.mongodb.reactivestreams>
         <version.lib.mysql-connector-java>8.0.11</version.lib.mysql-connector-java>
@@ -485,6 +488,21 @@
                     </exclusion>
                 </exclusions>
             </dependency>
+            <dependency>
+                <groupId>org.eclipse.microprofile.reactive-streams-operators</groupId>
+                <artifactId>microprofile-reactive-streams-operators-api</artifactId>
+                <version>${version.lib.microprofile-reactive-streams-operators-api}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.eclipse.microprofile.reactive-streams-operators</groupId>
+                <artifactId>microprofile-reactive-streams-operators-core</artifactId>
+                <version>${version.lib.microprofile-reactive-streams-operators-core}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.eclipse.microprofile.reactive.messaging</groupId>
+                <artifactId>microprofile-reactive-messaging-api</artifactId>
+                <version>${version.lib.microprofile-reactive-messaging-api}</version>
+            </dependency>
             <dependency>
                 <groupId>com.netflix.hystrix</groupId>
                 <artifactId>hystrix-core</artifactId>

microprofile/messaging/pom.xml

@@ -0,0 +1,73 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>io.helidon.microprofile</groupId>
+        <artifactId>helidon-microprofile-project</artifactId>
+        <version>2.0.0-SNAPSHOT</version>
+    </parent>
+
+    <groupId>io.helidon.microprofile.messaging</groupId>
+    <artifactId>helidon-microprofile-messaging</artifactId>
+    <name>Helidon MicroProfile Reactive Messaging</name>
+    <description>
+        Helidon MicroProfile Reactive Messaging
+    </description>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.eclipse.microprofile.reactive.messaging</groupId>
+            <artifactId>microprofile-reactive-messaging-api</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.eclipse.microprofile.reactive-streams-operators</groupId>
+                    <artifactId>microprofile-reactive-streams-operators-api</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.eclipse.microprofile.reactive-streams-operators</groupId>
+                    <artifactId>microprofile-reactive-streams-operators-core</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>io.helidon.microprofile.config</groupId>
+            <artifactId>helidon-microprofile-config</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.helidon.microprofile.server</groupId>
+            <artifactId>helidon-microprofile-server</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.helidon.microprofile.reactive-streams</groupId>
+            <artifactId>helidon-microprofile-reactive-streams</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>javax.interceptor</groupId>
+            <artifactId>javax.interceptor-api</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.helidon.microprofile.bundles</groupId>
+            <artifactId>internal-test-libs</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

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

@@ -0,0 +1,126 @@
+/*
+ * 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.lang.reflect.Method;
+import java.util.Optional;
+
+import javax.enterprise.inject.spi.Bean;
+import javax.enterprise.inject.spi.BeanManager;
+
+import io.helidon.common.Errors;
+import io.helidon.config.Config;
+
+import org.eclipse.microprofile.reactive.messaging.Acknowledgment;
+
+abstract class AbstractMessagingMethod {
+
+    private String incomingChannelName;
+    private String outgoingChannelName;
+
+    private Bean<?> bean;
+    private Object beanInstance;
+    private MethodSignatureType type;
+    private final Method method;
+    private final Errors.Collector errors;
+    private Acknowledgment.Strategy ackStrategy;
+
+
+    AbstractMessagingMethod(Method method, Errors.Collector errors) {
+        this.method = method;
+        this.errors = errors;
+        Optional<MethodSignatureType> signatureType = MethodSignatureResolver
+                .create(method)
+                .resolve();
+        if (signatureType.isPresent()) {
+            this.type = signatureType.get();
+            resolveAckStrategy();
+        } else {
+            errors.fatal("Unsupported method signature " + method);
+        }
+    }
+
+    void validate() {
+        Optional.ofNullable(method.getAnnotation(Acknowledgment.class))
+                .map(Acknowledgment::value)
+                .filter(s -> !type.getSupportedAckStrategies().contains(s))
+                .ifPresent(strategy -> {
+                    errors.fatal(String.format("Acknowledgment strategy %s is not supported for method signature: %s",
+                            strategy, type));
+                });
+    }
+
+    void init(BeanManager beanManager, Config config) {
+        this.beanInstance = ChannelRouter.lookup(bean, beanManager);
+    }
+
+    Method getMethod() {
+        return method;
+    }
+
+    Errors.Collector errors() {
+        return errors;
+    }
+
+    Object getBeanInstance() {
+        return beanInstance;
+    }
+
+    void setDeclaringBean(Bean<?> bean) {
+        this.bean = bean;
+    }
+
+    Class<?> getDeclaringType() {
+        return method.getDeclaringClass();
+    }
+
+    String getIncomingChannelName() {
+        return incomingChannelName;
+    }
+
+    String getOutgoingChannelName() {
+        return outgoingChannelName;
+    }
+
+    void setIncomingChannelName(String incomingChannelName) {
+        this.incomingChannelName = incomingChannelName;
+    }
+
+    void setOutgoingChannelName(String outgoingChannelName) {
+        this.outgoingChannelName = outgoingChannelName;
+    }
+
+    MethodSignatureType getType() {
+        return type;
+    }
+
+    void setType(MethodSignatureType type) {
+        this.type = type;
+    }
+
+    Acknowledgment.Strategy getAckStrategy() {
+        return ackStrategy;
+    }
+
+    private void resolveAckStrategy() {
+        ackStrategy =
+                Optional.ofNullable(method.getAnnotation(Acknowledgment.class))
+                        .map(Acknowledgment::value)
+                        .orElse(type.getDefaultAckType());
+    }
+}

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

@@ -0,0 +1,61 @@
+/*
+ * 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.HashMap;
+import java.util.Map;
+
+import io.helidon.config.Config;
+import io.helidon.config.ConfigSources;
+
+/**
+ * Detached configuration of a single connector.
+ */
+class AdHocConfigBuilder {
+    private final Map<String, String> configuration = new HashMap<>();
+
+    private AdHocConfigBuilder() {
+    }
+
+    static AdHocConfigBuilder from(Config config) {
+        AdHocConfigBuilder result = new AdHocConfigBuilder();
+        result.putAll(config);
+        return result;
+    }
+
+    AdHocConfigBuilder put(String key, String value) {
+        configuration.put(key, value);
+        return this;
+    }
+
+    AdHocConfigBuilder putAll(Config configToPut) {
+        configuration.putAll(configToPut.detach().asMap().orElse(Map.of()));
+        return this;
+    }
+
+    org.eclipse.microprofile.config.Config build() {
+        Config newConfig = Config.builder(ConfigSources.create(configuration))
+                .disableEnvironmentVariablesSource()
+                .disableSystemPropertiesSource()
+                .disableFilterServices()
+                .disableSourceServices()
+                .disableParserServices()
+                .build();
+        return (org.eclipse.microprofile.config.Config) newConfig;
+    }
+}