Changeset 7:6d4d4d6b7cbd

Added more producer/consumer pairs
author unexist
date Thu, 11 Feb 2021 17:07:40 +0100
parents 2e6cbf140ff8
children 036e0f1b9b10
files pom.xml src/main/java/dev/unexist/showcase/todo/adapters/MessageResource.java src/main/java/dev/unexist/showcase/todo/adapters/TodoConsumer.java src/main/java/dev/unexist/showcase/todo/adapters/TodoProducer.java src/main/java/dev/unexist/showcase/todo/adapters/TodoResource.java src/main/java/dev/unexist/showcase/todo/adapters/cdi/TodoCdiConsumer.java src/main/java/dev/unexist/showcase/todo/adapters/cloudevents/TodoCloudEventConsumer.java src/main/java/dev/unexist/showcase/todo/adapters/cloudevents/TodoCloudEventProducer.java src/main/java/dev/unexist/showcase/todo/adapters/smallrye/TodoSmallryeConsumer.java src/main/java/dev/unexist/showcase/todo/adapters/smallrye/TodoSmallryeProducer.java src/main/java/dev/unexist/showcase/todo/domain/todo/TodoRepository.java src/main/java/dev/unexist/showcase/todo/domain/todo/TodoService.java
diffstat 12 files changed, 313 insertions(+), 169 deletions(-) [+]
line wrap: on
line diff
--- a/pom.xml	Wed Feb 10 18:29:31 2021 +0100
+++ b/pom.xml	Thu Feb 11 17:07:40 2021 +0100
@@ -14,6 +14,8 @@
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 
         <cloudevents.version>2.0.0.RC2</cloudevents.version>
+        <mutiny.version>0.13.0</mutiny.version>
+        <kafka-cdi-extension.version>0.1.0</kafka-cdi-extension.version>
 
         <apache-commons-lang.version>3.11</apache-commons-lang.version>
         <hg-revision-plugin.version>0.10</hg-revision-plugin.version>
@@ -26,6 +28,10 @@
         <quarkus.platform.group-id>io.quarkus</quarkus.platform.group-id>
         <quarkus.platform.version>1.11.1.Final</quarkus.platform.version>
 
+        <!-- Smallrye -->
+        <smallrye-reactive-messaging.version>3.0.0</smallrye-reactive-messaging.version>
+        <smallrye-reactive-messaging-cloud-events.version>2.8.0</smallrye-reactive-messaging-cloud-events.version>
+
         <!-- Audit -->
         <checkstyle-plugin.version>3.1.1</checkstyle-plugin.version>
         <spotbugs-maven-plugin.version>4.1.3</spotbugs-maven-plugin.version>
@@ -62,6 +68,10 @@
         </dependency>
         <dependency>
             <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-smallrye-reactive-messaging-kafka</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
             <artifactId>quarkus-hibernate-validator</artifactId>
         </dependency>
 
@@ -84,6 +94,30 @@
             <version>${cloudevents.version}</version>
         </dependency>
 
+        <!-- Smallrye -->
+        <dependency>
+            <groupId>io.smallrye.reactive</groupId>
+            <artifactId>smallrye-reactive-messaging-cloud-events</artifactId>
+            <version>${smallrye-reactive-messaging-cloud-events.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.smallrye.reactive</groupId>
+            <artifactId>smallrye-reactive-messaging-provider</artifactId>
+            <version>${smallrye-reactive-messaging.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.smallrye.reactive</groupId>
+            <artifactId>mutiny-reactive-streams-operators</artifactId>
+            <version>${mutiny.version}</version>
+        </dependency>
+
+        <!-- Kafka CDI -->
+        <dependency>
+            <groupId>org.aerogear.kafka</groupId>
+            <artifactId>kafka-cdi-extension</artifactId>
+            <version>${kafka-cdi-extension.version}</version>
+        </dependency>
+
         <!-- Apache Commons -->
         <dependency>
             <groupId>org.apache.commons</groupId>
--- a/src/main/java/dev/unexist/showcase/todo/adapters/MessageResource.java	Wed Feb 10 18:29:31 2021 +0100
+++ b/src/main/java/dev/unexist/showcase/todo/adapters/MessageResource.java	Thu Feb 11 17:07:40 2021 +0100
@@ -2,7 +2,7 @@
  * @package Quarkus-Messaging-Showcase
  *
  * @file Todo resource
- * @copyright 2020-2021 Christoph Kappel <christoph@unexist.dev>
+ * @copyright 2020 Christoph Kappel <christoph@unexist.dev>
  * @version $Id$
  *
  * This program can be distributed under the terms of the GNU GPLv2.
@@ -11,29 +11,59 @@
 
 package dev.unexist.showcase.todo.adapters;
 
+import dev.unexist.showcase.todo.adapters.cdi.TodoCdiConsumer;
+import dev.unexist.showcase.todo.adapters.cloudevents.TodoCloudEventConsumer;
+import dev.unexist.showcase.todo.adapters.cloudevents.TodoCloudEventProducer;
+import dev.unexist.showcase.todo.adapters.smallrye.TodoSmallryeConsumer;
+import dev.unexist.showcase.todo.adapters.smallrye.TodoSmallryeProducer;
+import org.aerogear.kafka.SimpleKafkaProducer;
+import org.aerogear.kafka.cdi.annotation.Producer;
 import org.eclipse.microprofile.openapi.annotations.Operation;
 import org.eclipse.microprofile.openapi.annotations.responses.APIResponse;
 import org.eclipse.microprofile.openapi.annotations.responses.APIResponses;
 import org.eclipse.microprofile.openapi.annotations.tags.Tag;
 
 import javax.inject.Inject;
-import javax.ws.rs.GET;
 import javax.ws.rs.POST;
 import javax.ws.rs.Path;
-import javax.ws.rs.Produces;
-import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
-import java.util.List;
-import java.util.Map;
 
 @Path("/message")
 public class MessageResource {
 
+    /* Cdi */
+    @Producer
+    SimpleKafkaProducer<String, String> todoCdiProducer;
+
     @Inject
-    TodoProducer todoProducer;
+    TodoCdiConsumer todoCdiConsumer;
+
+    /* Cloudevents */
+    @Inject
+    TodoCloudEventProducer todoCloudEventProducer;
 
     @Inject
-    TodoConsumer todoConsumer;
+    TodoCloudEventConsumer todoCloudEventConsumer;
+
+    /* Smallrye */
+    @Inject
+    TodoSmallryeProducer todoSmallryeProducer;
+
+    @Inject
+    TodoSmallryeConsumer todoSmallryeConsumer;
+
+    @POST
+    @Operation(summary = "Send cdi event via kafka")
+    @Tag(name = "Message")
+    @APIResponses({
+            @APIResponse(responseCode = "204", description = "Nothing found"),
+            @APIResponse(responseCode = "500", description = "Server error")
+    })
+    public Response sendCdiEvent() {
+        this.todoCdiProducer.send("todos-cdi", "test", "test");
+
+        return Response.noContent().build();
+    }
 
     @POST
     @Operation(summary = "Send cloudevent via kafka")
@@ -43,28 +73,21 @@
             @APIResponse(responseCode = "500", description = "Server error")
     })
     public Response sendCloudEvent() {
-        this.todoProducer.send();
+        this.todoCloudEventProducer.send();
 
         return Response.noContent().build();
     }
 
-    @GET
-    @Produces(MediaType.APPLICATION_JSON)
-    @Operation(summary = "Receive cloudevent via kafka")
+    @POST
+    @Operation(summary = "Send smallrye via kafka")
     @Tag(name = "Message")
     @APIResponses({
             @APIResponse(responseCode = "204", description = "Nothing found"),
             @APIResponse(responseCode = "500", description = "Server error")
     })
-    public Response receiveCloudEvent() {
-        Response.ResponseBuilder builder = Response.noContent();
-
-        List<Map<String, String>> list = this.todoConsumer.receive();
+    public Response sendSmallryeEvent() {
+        this.todoSmallryeProducer.send();
 
-        if (!list.isEmpty()) {
-            builder = Response.ok(list);
-        }
-
-        return builder.build();
+        return Response.noContent().build();
     }
 }
--- a/src/main/java/dev/unexist/showcase/todo/adapters/TodoConsumer.java	Wed Feb 10 18:29:31 2021 +0100
+++ /dev/null	Thu Jan 01 00:00:00 1970 +0000
@@ -1,69 +0,0 @@
-/**
- * @package Quarkus-Messaging-Showcase
- *
- * @file Todo consumer
- * @copyright 2020-2021 Christoph Kappel <christoph@unexist.dev>
- * @version $Id$
- *
- * This program can be distributed under the terms of the GNU GPLv2.
- * See the file LICENSE for details.
- **/
-
-package dev.unexist.showcase.todo.adapters;
-
-import io.cloudevents.CloudEvent;
-import io.cloudevents.kafka.CloudEventDeserializer;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.common.serialization.StringDeserializer;
-
-import javax.enterprise.context.ApplicationScoped;
-import java.time.Duration;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-
-@ApplicationScoped
-public class TodoConsumer {
-    private KafkaConsumer<String, CloudEvent> consumer;
-
-    TodoConsumer() {
-        Properties props = new Properties();
-
-        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
-        props.put(ConsumerConfig.GROUP_ID_CONFIG, "todo-cloudevents-consumer");
-        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
-        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, CloudEventDeserializer.class);
-        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
-        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
-
-        this.consumer = new KafkaConsumer<>(props);
-
-        consumer.subscribe(Collections.singletonList("todos"));
-
-        System.out.println("Consumer started");
-    }
-
-    public List<Map<String, String>> receive() {
-        java.util.List<Map<String, String>> list = Collections.emptyList();
-
-        ConsumerRecords<String, CloudEvent> consumerRecords = consumer.poll(Duration.ofMillis(100));
-        consumerRecords.forEach(record -> {
-            Map<String, String> recordEntry = new HashMap<>();
-
-            recordEntry.put("Record Key", record.key());
-            recordEntry.put("Record value", record.value().toString());
-            recordEntry.put("Record partition",
-                    String.valueOf(record.partition()));
-            recordEntry.put("Record offset",
-                    String.valueOf(record.offset()));
-
-            list.add(recordEntry);
-        });
-
-        return list;
-    }
-}
--- a/src/main/java/dev/unexist/showcase/todo/adapters/TodoProducer.java	Wed Feb 10 18:29:31 2021 +0100
+++ /dev/null	Thu Jan 01 00:00:00 1970 +0000
@@ -1,76 +0,0 @@
-/**
- * @package Quarkus-Messaging-Showcase
- *
- * @file Todo producer
- * @copyright 2020-2021 Christoph Kappel <christoph@unexist.dev>
- * @version $Id$
- *
- * This program can be distributed under the terms of the GNU GPLv2.
- * See the file LICENSE for details.
- **/
-
-package dev.unexist.showcase.todo.adapters;
-
-import io.cloudevents.CloudEvent;
-import io.cloudevents.core.message.Encoding;
-import io.cloudevents.core.v1.CloudEventBuilder;
-import io.cloudevents.jackson.JsonFormat;
-import io.cloudevents.kafka.CloudEventSerializer;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.clients.producer.RecordMetadata;
-import org.apache.kafka.common.serialization.StringSerializer;
-
-import javax.enterprise.context.ApplicationScoped;
-import java.net.URI;
-import java.util.Properties;
-import java.util.UUID;
-import java.util.concurrent.ExecutionException;
-
-@ApplicationScoped
-public class TodoProducer {
-    private KafkaProducer<String, CloudEvent> producer;
-    private CloudEventBuilder eventBuilder;
-
-    TodoProducer() {
-        Properties props = new Properties();
-
-        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
-        props.put(ProducerConfig.CLIENT_ID_CONFIG, "todo-cloudevents-producer");
-        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, CloudEventSerializer.class);
-        props.put(CloudEventSerializer.ENCODING_CONFIG, Encoding.STRUCTURED);
-        props.put(CloudEventSerializer.EVENT_FORMAT_CONFIG, JsonFormat.CONTENT_TYPE);
-
-        this.producer = new KafkaProducer<>(props);
-
-        this.eventBuilder = io.cloudevents.core.builder.CloudEventBuilder.v1()
-                .withSource(URI.create("https://unexist.dev"))
-                .withType("todo");
-    }
-
-    public void send() {
-        try {
-            String id = UUID.randomUUID().toString();
-            String data = "Todo event";
-
-            CloudEvent event = this.eventBuilder.newBuilder()
-                    .withId(id)
-                    .withData("text/plain", data.getBytes())
-                    .build();
-
-            RecordMetadata metadata = this.producer
-                    .send(new ProducerRecord<>("todos", id, event))
-                    .get();
-
-            System.out.println("Record sent to partition " + metadata.partition() +
-                    " with offset " + metadata.offset());
-        } catch (InterruptedException|ExecutionException e) {
-            System.out.println("Error while trying to send the record");
-            e.printStackTrace();
-        }
-
-        this.producer.flush();
-    }
-}
--- a/src/main/java/dev/unexist/showcase/todo/adapters/TodoResource.java	Wed Feb 10 18:29:31 2021 +0100
+++ b/src/main/java/dev/unexist/showcase/todo/adapters/TodoResource.java	Thu Feb 11 17:07:40 2021 +0100
@@ -2,7 +2,7 @@
  * @package Quarkus-Messaging-Showcase
  *
  * @file Todo resource
- * @copyright 2020-2021 Christoph Kappel <christoph@unexist.dev>
+ * @copyright 2020 Christoph Kappel <christoph@unexist.dev>
  * @version $Id$
  *
  * This program can be distributed under the terms of the GNU GPLv2.
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/main/java/dev/unexist/showcase/todo/adapters/cdi/TodoCdiConsumer.java	Thu Feb 11 17:07:40 2021 +0100
@@ -0,0 +1,31 @@
+/**
+ * @package Quarkus-Messaging-Showcase
+ *
+ * @file Todo cdi consumer
+ * @copyright 2020 Christoph Kappel <christoph@unexist.dev>
+ * @version $Id$
+ *
+ * This program can be distributed under the terms of the GNU GPLv2.
+ * See the file LICENSE for details.
+ **/
+
+package dev.unexist.showcase.todo.adapters.cdi;
+
+import org.aerogear.kafka.cdi.annotation.Consumer;
+import org.aerogear.kafka.cdi.annotation.KafkaConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.enterprise.context.ApplicationScoped;
+
+@ApplicationScoped
+@KafkaConfig(bootstrapServers ="localhost:9092")
+public class TodoCdiConsumer {
+    private static final Logger LOGGER = LoggerFactory.getLogger(TodoCdiConsumer.class);
+
+    @Consumer(topics = "topic-cdi", groupId = "todo-cdi-consumer")
+    public void onMessage(final String key, final String value) {
+        LOGGER.info("Key: {}", key);
+        LOGGER.info("Value: {}", value);
+    }
+}
\ No newline at end of file
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/main/java/dev/unexist/showcase/todo/adapters/cloudevents/TodoCloudEventConsumer.java	Thu Feb 11 17:07:40 2021 +0100
@@ -0,0 +1,60 @@
+/**
+ * @package Quarkus-Messaging-Showcase
+ *
+ * @file Todo cloudevent consumer
+ * @copyright 2020 Christoph Kappel <christoph@unexist.dev>
+ * @version $Id$
+ *
+ * This program can be distributed under the terms of the GNU GPLv2.
+ * See the file LICENSE for details.
+ **/
+
+package dev.unexist.showcase.todo.adapters.cloudevents;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.kafka.CloudEventDeserializer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.enterprise.context.ApplicationScoped;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Properties;
+
+@ApplicationScoped
+public class TodoCloudEventConsumer {
+    private static final Logger LOGGER = LoggerFactory.getLogger(TodoCloudEventConsumer.class);
+
+    KafkaConsumer<String, CloudEvent> consumer;
+
+    TodoCloudEventConsumer() {
+        Properties props = new Properties();
+
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, "todo-ce-consumer");
+        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, CloudEventDeserializer.class);
+        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
+        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
+
+        this.consumer = new KafkaConsumer<>(props);
+
+        consumer.subscribe(Collections.singletonList("todos-ce"));
+
+        LOGGER.info("Cloudevent consumer started");
+    }
+
+    public void receive() {
+        ConsumerRecords<String, CloudEvent> consumerRecords = consumer.poll(Duration.ofMillis(100));
+        consumerRecords.forEach(record -> {
+            LOGGER.info("Record key: {}", record.key());
+            LOGGER.info("Record value: {}", record.value().toString());
+            LOGGER.info("Record partition: {}", record.partition());
+            LOGGER.info("Record offset: {}", record.offset());
+        });
+    }
+}
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/main/java/dev/unexist/showcase/todo/adapters/cloudevents/TodoCloudEventProducer.java	Thu Feb 11 17:07:40 2021 +0100
@@ -0,0 +1,81 @@
+/**
+ * @package Quarkus-Messaging-Showcase
+ *
+ * @file Todo cloudevent producer
+ * @copyright 2020 Christoph Kappel <christoph@unexist.dev>
+ * @version $Id$
+ *
+ * This program can be distributed under the terms of the GNU GPLv2.
+ * See the file LICENSE for details.
+ **/
+
+package dev.unexist.showcase.todo.adapters.cloudevents;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.message.Encoding;
+import io.cloudevents.jackson.JsonFormat;
+import io.cloudevents.kafka.CloudEventSerializer;
+import io.cloudevents.v03.CloudEventBuilder;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.enterprise.context.ApplicationScoped;
+import java.net.URI;
+import java.util.Properties;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+
+@ApplicationScoped
+public class TodoCloudEventProducer {
+    private static final Logger LOGGER = LoggerFactory.getLogger(TodoCloudEventConsumer.class);
+
+    private KafkaProducer<String, CloudEvent> producer;
+    private CloudEventBuilder eventBuilder;
+
+    TodoCloudEventProducer() {
+        Properties props = new Properties();
+
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+        props.put(ProducerConfig.CLIENT_ID_CONFIG, "todo-cloudevents-producer");
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, CloudEventSerializer.class);
+        props.put(CloudEventSerializer.ENCODING_CONFIG, Encoding.STRUCTURED);
+        props.put(CloudEventSerializer.EVENT_FORMAT_CONFIG, JsonFormat.CONTENT_TYPE);
+
+        this.producer = new KafkaProducer<>(props);
+
+        this.eventBuilder = CloudEventBuilder.builder()
+                .withSource(URI.create("https://unexist.dev"))
+                .withType("todo");
+    }
+
+    public void send() {
+        try {
+            String id = UUID.randomUUID().toString();
+            String data = "Todo event";
+
+            CloudEvent event = this.eventBuilder
+                    .withId(id)
+                    .withData(data)
+                    .build();
+
+            RecordMetadata metadata = this.producer
+                    .send(new ProducerRecord<>("todos-ce", id, event))
+                    .get();
+
+            LOGGER.info("Record sent to partition {} with offset {}",
+                    metadata.partition(), metadata.offset());
+        } catch (InterruptedException|ExecutionException e) {
+            LOGGER.error("Error while trying to send the record");
+
+            e.printStackTrace();
+        }
+
+        this.producer.flush();
+    }
+}
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/main/java/dev/unexist/showcase/todo/adapters/smallrye/TodoSmallryeConsumer.java	Thu Feb 11 17:07:40 2021 +0100
@@ -0,0 +1,31 @@
+/**
+ * @package Quarkus-Messaging-Showcase
+ *
+ * @file Todo smallrye consumer
+ * @copyright 2020 Christoph Kappel <christoph@unexist.dev>
+ * @version $Id$
+ *
+ * This program can be distributed under the terms of the GNU GPLv2.
+ * See the file LICENSE for details.
+ **/
+
+package dev.unexist.showcase.todo.adapters.smallrye;
+
+import org.eclipse.microprofile.reactive.messaging.Acknowledgment;
+import org.eclipse.microprofile.reactive.messaging.Incoming;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.enterprise.context.ApplicationScoped;
+
+@ApplicationScoped
+public class TodoSmallryeConsumer {
+    private static final Logger LOGGER = LoggerFactory.getLogger(TodoSmallryeConsumer.class);
+
+    @Incoming("todos-rye-in")
+    @Acknowledgment(Acknowledgment.Strategy.PRE_PROCESSING)
+    public void receive(String value) {
+        LOGGER.info("Value: {}", value);
+
+    }
+}
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/main/java/dev/unexist/showcase/todo/adapters/smallrye/TodoSmallryeProducer.java	Thu Feb 11 17:07:40 2021 +0100
@@ -0,0 +1,29 @@
+/**
+ * @package Quarkus-Messaging-Showcase
+ *
+ * @file Todo producer
+ * @copyright 2020 Christoph Kappel <christoph@unexist.dev>
+ * @version $Id$
+ *
+ * This program can be distributed under the terms of the GNU GPLv2.
+ * See the file LICENSE for details.
+ **/
+
+package dev.unexist.showcase.todo.adapters.smallrye;
+
+import io.smallrye.mutiny.Multi;
+import org.eclipse.microprofile.reactive.messaging.Outgoing;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.enterprise.context.ApplicationScoped;
+
+@ApplicationScoped
+public class TodoSmallryeProducer {
+    private static final Logger LOGGER = LoggerFactory.getLogger(TodoSmallryeConsumer.class);
+
+    @Outgoing("todo-rye-out")
+    public Multi<String> send() {
+        return Multi.createFrom().item("Todo");
+    }
+}
--- a/src/main/java/dev/unexist/showcase/todo/domain/todo/TodoRepository.java	Wed Feb 10 18:29:31 2021 +0100
+++ b/src/main/java/dev/unexist/showcase/todo/domain/todo/TodoRepository.java	Thu Feb 11 17:07:40 2021 +0100
@@ -2,7 +2,7 @@
  * @package Quarkus-Messaging-Showcase
  *
  * @file Todo repository
- * @copyright 2020-2021 Christoph Kappel <christoph@unexist.dev>
+ * @copyright 2020 Christoph Kappel <christoph@unexist.dev>
  * @version $Id$
  *
  * This program can be distributed under the terms of the GNU GPLv2.
--- a/src/main/java/dev/unexist/showcase/todo/domain/todo/TodoService.java	Wed Feb 10 18:29:31 2021 +0100
+++ b/src/main/java/dev/unexist/showcase/todo/domain/todo/TodoService.java	Thu Feb 11 17:07:40 2021 +0100
@@ -2,7 +2,7 @@
  * @package Quarkus-Messaging-Showcase
  *
  * @file Todo service and domain service
- * @copyright 2020-2021 Christoph Kappel <christoph@unexist.dev>
+ * @copyright 2020 Christoph Kappel <christoph@unexist.dev>
  * @version $Id$
  *
  * This program can be distributed under the terms of the GNU GPLv2.