[KOGITO-4457] Removing knative-eventing-addon (#1199)

* [KOGITO-4457] Removing knative-eventing-addon Signed-off-by: Ricardo Zanini <[email protected]> * Adding Kogito Events API dependency Signed-off-by: Ricardo Zanini <[email protected]>
apache · Apr 14, 2021 · fea3d18 · fea3d18
1 parent 2f3cfd8
commit fea3d18
Show file tree

Hide file tree

Showing 29 changed files with 151 additions and 227 deletions.
diff --git a/addons/cloudevents/cloudevents-common-addon/pom.xml b/addons/cloudevents/cloudevents-common-addon/pom.xml
@@ -34,6 +34,27 @@
       <groupId>org.kie.kogito</groupId>
       <artifactId>kogito-events-api</artifactId>
     </dependency>
+
+    <dependency>
+      <groupId>org.kie.kogito</groupId>
+      <artifactId>kogito-services</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-engine</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.cloudevents</groupId>
+      <artifactId>cloudevents-json-jackson</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
 </project>
diff --git a/...s/knative/ce/CloudEventConverterTest.java → .../cloudevents/CloudEventConverterTest.java b/...s/knative/ce/CloudEventConverterTest.java → .../cloudevents/CloudEventConverterTest.java
@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.kie.kogito.events.knative.ce;
+package org.kie.kogito.addon.cloudevents;
 
 import java.io.IOException;
 import java.net.URI;
@@ -23,13 +23,12 @@
 import org.kie.kogito.event.CloudEventExtensionConstants;
 import org.kie.kogito.services.event.AbstractProcessDataEvent;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 import io.cloudevents.CloudEvent;
 import io.cloudevents.core.builder.CloudEventBuilder;
 import io.cloudevents.jackson.JsonFormat;
-import io.vertx.core.buffer.Buffer;
-import io.vertx.core.json.JsonObject;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -38,7 +37,7 @@ class CloudEventConverterTest {
     static final ObjectMapper objectMapper = new ObjectMapper().registerModule(JsonFormat.getCloudEventJacksonModule());
 
     @Test
-    void verifyBasicCloudEventConversion() {
+    void verifyBasicCloudEventConversion() throws IOException {
         // given
         final String eventId = UUID.randomUUID().toString();
         final URI src = URI.create("/trigger");
@@ -54,8 +53,8 @@ void verifyBasicCloudEventConversion() {
                         .withData(payload.getBytes())
                         .build();
 
-        final JsonObject ceJson = new JsonObject(Buffer.buffer(cloudEvent.getData()));
-        assertThat(ceJson.getString("message")).isNotEmpty().isEqualTo("Oi Mundo!");
+        final JsonNode ceJson = objectMapper.readTree(cloudEvent.getData());
+        assertThat(ceJson.get("message").asText()).isNotEmpty().isEqualTo("Oi Mundo!");
     }
 
     @Test

diff --git a/...e/ce/http/CloudEventListenerResource.java → ...tiontests/CloudEventListenerResource.java b/...e/ce/http/CloudEventListenerResource.java → ...tiontests/CloudEventListenerResource.java
@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.kie.kogito.events.knative.ce.http;
+package org.kie.kogito.integrationtests;
 
 import javax.ws.rs.Consumes;
 import javax.ws.rs.POST;
@@ -22,7 +22,8 @@
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 
-import org.kie.kogito.events.knative.ce.Printer;
+import org.kie.kogito.addon.cloudevents.quarkus.http.Responses;
+import org.kie.kogito.cloudevents.Printer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

diff --git a/.../http/CloudEventListenerResourceTest.java → ...tests/CloudEventListenerResourceTest.java b/.../http/CloudEventListenerResourceTest.java → ...tests/CloudEventListenerResourceTest.java
@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.kie.kogito.events.knative.ce.http;
+package org.kie.kogito.integrationtests;
 
 import java.nio.charset.StandardCharsets;
 import java.util.UUID;

diff --git a/addons/cloudevents/cloudevents-quarkus-addon/pom.xml b/addons/cloudevents/cloudevents-quarkus-addon/pom.xml
@@ -67,6 +67,14 @@
       <groupId>org.eclipse.microprofile.config</groupId>
       <artifactId>microprofile-config-api</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.cloudevents</groupId>
+      <artifactId>cloudevents-http-restful-ws</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.cloudevents</groupId>
+      <artifactId>cloudevents-json-jackson</artifactId>
+    </dependency>
 
     <!-- These dependencies are needed in case Smallrye is configured with HTTP connector -->
     <dependency>
@@ -79,6 +87,11 @@
       <artifactId>jboss-jaxrs-api_2.1_spec</artifactId>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>io.quarkus</groupId>
+      <artifactId>quarkus-resteasy</artifactId>
+      <scope>provided</scope>
+    </dependency>
 
     <dependency>
       <groupId>org.slf4j</groupId>

diff --git a/...on/src/main/java/org/kie/kogito/addon/cloudevents/quarkus/QuarkusCloudEventPublisher.java b/...on/src/main/java/org/kie/kogito/addon/cloudevents/quarkus/QuarkusCloudEventPublisher.java
@@ -15,35 +15,78 @@
  */
 package org.kie.kogito.addon.cloudevents.quarkus;
 
+import java.util.concurrent.CompletionStage;
+
 import javax.enterprise.context.ApplicationScoped;
 import javax.enterprise.inject.Produces;
 import javax.inject.Named;
 
-import org.eclipse.microprofile.reactive.messaging.Channel;
+import org.eclipse.microprofile.reactive.messaging.Incoming;
 import org.eclipse.microprofile.reactive.messaging.Message;
 import org.kie.kogito.event.KogitoEventStreams;
 import org.reactivestreams.Publisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import io.quarkus.runtime.Startup;
 import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.operators.multi.processors.BroadcastProcessor;
 
 /**
  * Takes a @Channel event stream and re-exposes it as a Multi
  * (a subclass of {@link Publisher})
+ * 
+ * @see QuarkusCloudEventReceiver
+ * @see QuarkusCloudEventEmitter
  */
 @Startup
 @ApplicationScoped
 public class QuarkusCloudEventPublisher {
-    @Channel(KogitoEventStreams.INCOMING)
-    Multi<Message<String>> events;
+    private static final Logger LOGGER = LoggerFactory.getLogger(QuarkusCloudEventPublisher.class);
+
+    protected BroadcastProcessor<String> processor = BroadcastProcessor.create();
 
+    /**
+     * Broadcasts the received/produced messages to subscribers
+     *
+     * @see <a href="https://smallrye.io/smallrye-mutiny/guides/hot-streams">How to create a hot stream?</a>
+     * @return A {@link Multi} message to subscribers
+     */
     @Produces
     @ApplicationScoped
     @Named(KogitoEventStreams.PUBLISHER)
-    public Multi<String> makeMulti() {
-        return events
-                .invoke(Message::ack)
-                .map(Message::getPayload)
-                .broadcast().toAllSubscribers();
+    public Multi<String> producerFactory() {
+        return processor;
+    }
+
+    /**
+     * Listens to a message published in the {@link KogitoEventStreams#INCOMING} channel
+     *
+     * @param message the given message in JSON format
+     * @return a {@link CompletionStage} after ack-ing the message
+     */
+    @Incoming(KogitoEventStreams.INCOMING)
+    public CompletionStage<Void> onEvent(Message<String> message) {
+        LOGGER.debug("Received message from channel {}: {}", KogitoEventStreams.INCOMING, message);
+        return message
+                .ack()
+                .exceptionally(e -> {
+                    LOGGER.error("Failed to ack message", e);
+                    return null;
+                })
+                .thenApply(r -> {
+                    produce(message.getPayload());
+                    return null;
+                });
+    }
+
+    /**
+     * Produces a message in the internal application bus
+     *
+     * @param message the given CE message in JSON format
+     */
+    public void produce(final String message) {
+        LOGGER.debug("Producing message to internal bus: {}", message);
+        processor.onNext(message);
     }
 }
diff --git a/...main/java/org/kie/kogito/addon/cloudevents/quarkus/QuarkusKogitoExtensionInitializer.java b/...main/java/org/kie/kogito/addon/cloudevents/quarkus/QuarkusKogitoExtensionInitializer.java
@@ -19,6 +19,7 @@
 import javax.inject.Inject;
 
 import org.kie.kogito.cloudevents.extension.KogitoExtension;
+import org.kie.kogito.cloudevents.extension.KogitoProcessExtension;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -43,6 +44,7 @@ public class QuarkusKogitoExtensionInitializer {
     private void onPostConstruct() {
         mapper.registerModule(JsonFormat.getCloudEventJacksonModule());
         KogitoExtension.register();
+        KogitoProcessExtension.register();
         LOG.info("Registered Kogito CloudEvent extension");
     }
 }
diff --git a/.../org/kie/kogito/addon/cloudevents/quarkus/decorators/CloudEventHttpOutgoingDecorator.java b/.../org/kie/kogito/addon/cloudevents/quarkus/decorators/CloudEventHttpOutgoingDecorator.java
@@ -41,7 +41,7 @@ public final class CloudEventHttpOutgoingDecorator implements MessageDecorator {
                     .withQueryParameter(Collections.emptyMap())
                     .withHeader(HttpHeaders.CONTENT_TYPE, CLOUD_EVENTS_CONTENT_TYPE).build());
 
-    CloudEventHttpOutgoingDecorator() {
+    public CloudEventHttpOutgoingDecorator() {
 
     }
 

diff --git a/...events/knative/ce/http/ResponseError.java → ...oudevents/quarkus/http/ResponseError.java b/...events/knative/ce/http/ResponseError.java → ...oudevents/quarkus/http/ResponseError.java
@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.kie.kogito.events.knative.ce.http;
+package org.kie.kogito.addon.cloudevents.quarkus.http;
 
 /**
  * Simple structure to hold CloudEvent processing errors

diff --git a/...ito/events/knative/ce/http/Responses.java → ...n/cloudevents/quarkus/http/Responses.java b/...ito/events/knative/ce/http/Responses.java → ...n/cloudevents/quarkus/http/Responses.java
@@ -13,11 +13,11 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.kie.kogito.events.knative.ce.http;
+package org.kie.kogito.addon.cloudevents.quarkus.http;
 
 import javax.ws.rs.core.Response;
 
-import org.kie.kogito.events.knative.ce.Printer;
+import org.kie.kogito.cloudevents.Printer;
 
 import io.cloudevents.CloudEvent;
 

diff --git a/...events/knative/ce/http/ResponsesTest.java → ...oudevents/quarkus/http/ResponsesTest.java b/...events/knative/ce/http/ResponsesTest.java → ...oudevents/quarkus/http/ResponsesTest.java
@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.kie.kogito.events.knative.ce.http;
+package org.kie.kogito.addon.cloudevents.quarkus.http;
 
 import javax.ws.rs.core.Response;
 

diff --git a/...in/java/org/kie/kogito/addon/cloudevents/spring/SpringBootKogitoExtensionInitializer.java b/...in/java/org/kie/kogito/addon/cloudevents/spring/SpringBootKogitoExtensionInitializer.java
@@ -18,6 +18,7 @@
 import javax.annotation.PostConstruct;
 
 import org.kie.kogito.cloudevents.extension.KogitoExtension;
+import org.kie.kogito.cloudevents.extension.KogitoProcessExtension;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -43,6 +44,7 @@ public class SpringBootKogitoExtensionInitializer {
     private void onPostConstruct() {
         mapper.registerModule(JsonFormat.getCloudEventJacksonModule());
         KogitoExtension.register();
+        KogitoProcessExtension.register();
         LOG.info("Registered Kogito CloudEvent extension");
     }
 }
diff --git a/addons/cloudevents/cloudevents-utils/pom.xml b/addons/cloudevents/cloudevents-utils/pom.xml
@@ -13,6 +13,10 @@
   <name>Kogito :: Add-Ons :: CloudEvents :: Utils</name>
 
   <dependencies>
+    <dependency>
+      <groupId>org.kie.kogito</groupId>
+      <artifactId>kogito-events-api</artifactId>
+    </dependency>
     <dependency>
       <groupId>io.cloudevents</groupId>
       <artifactId>cloudevents-api</artifactId>
@@ -41,6 +45,11 @@
       <artifactId>slf4j-simple</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
 </project>
diff --git a/...kie/kogito/events/knative/ce/Printer.java → ...a/org/kie/kogito/cloudevents/Printer.java b/...kie/kogito/events/knative/ce/Printer.java → ...a/org/kie/kogito/cloudevents/Printer.java
@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.kie.kogito.events.knative.ce;
+package org.kie.kogito.cloudevents;
 
 import io.cloudevents.CloudEvent;
 

diff --git a/...ce/extensions/KogitoProcessExtension.java → ...nts/extension/KogitoProcessExtension.java b/...ce/extensions/KogitoProcessExtension.java → ...nts/extension/KogitoProcessExtension.java
@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.kie.kogito.events.knative.ce.extensions;
+package org.kie.kogito.cloudevents.extension;
 
 import java.util.Arrays;
 import java.util.HashMap;
@@ -26,6 +26,7 @@
 
 import io.cloudevents.CloudEventExtensions;
 import io.cloudevents.Extension;
+import io.cloudevents.core.provider.ExtensionProvider;
 
 // The size of this extension could be reevaluated since we could make use of `type`, `source` and `subject` for processId, referenceId and instanceState
 
@@ -61,6 +62,10 @@ public KogitoProcessExtension() {
         this.innerValues = new HashMap<>();
     }
 
+    public static void register() {
+        ExtensionProvider.getInstance().registerExtension(KogitoProcessExtension.class, KogitoProcessExtension::new);
+    }
+
     public String getKogitoProcessInstanceId() {
         return kogitoProcessInstanceId;
     }
@@ -156,10 +161,7 @@ public void readFrom(CloudEventExtensions extensions) {
     }
 
     private String getExtension(CloudEventExtensions extensions, String key) {
-        if (extensions.getExtension(key) == null) {
-            return "";
-        }
-        return extensions.getExtension(key).toString();
+        return Objects.toString(extensions.getExtension(key), "");
     }
 
     @Override

diff --git a/...xtensions/KogitoProcessExtensionTest.java → ...extension/KogitoProcessExtensionTest.java b/...xtensions/KogitoProcessExtensionTest.java → ...extension/KogitoProcessExtensionTest.java
@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.kie.kogito.events.knative.ce.extensions;
+package org.kie.kogito.cloudevents.extension;
 
 import java.net.URI;
 import java.util.UUID;
@@ -52,6 +52,8 @@ void verifyKogitoExtensionCanBeRead() {
     @Test
     void verifyKeysAreSet() {
         final KogitoProcessExtension kpe = ExtensionProvider.getInstance().parseExtension(KogitoProcessExtension.class, getExampleCloudEvent());
+        assertThat(kpe).isNotNull();
+        assertThat(kpe.getKeys()).isNotNull();
         assertThat(kpe.getKeys()).isNotEmpty();
     }