Hacking some tests, InMemory broker isn't really required and doesnt

seem to work out the box

quarkus-kafaka-homework/kafka-producer/pom.xml

@@ -63,11 +63,20 @@
       <groupId>io.quarkus</groupId>
       <artifactId>quarkus-smallrye-openapi</artifactId>
     </dependency>
+    <!-- Test Dependencies -->
     <dependency>
-      <groupId>io.quarkus</groupId>
+      <groupId>io.smallrye.reactive</groupId>
-      <artifactId>quarkus-container-image-jib</artifactId>
+      <artifactId>smallrye-reactive-messaging-in-memory</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.testcontainers</groupId>
+      <artifactId>kafka</artifactId>
+      <version>${testcontainers.version}</version>
+      <scope>test</scope>
     </dependency>
-</dependencies>
+
+  </dependencies>
   <build>
     <plugins>
       <plugin>

quarkus-kafaka-homework/kafka-producer/src/main/java/me/jochrist/kafka/MessageProducer.java

@@ -1,15 +1,10 @@
 package me.jochrist.kafka;
-import io.reactivex.Flowable;
-import org.eclipse.microprofile.reactive.messaging.Outgoing;
 import org.eclipse.microprofile.reactive.messaging.Channel;
 import org.eclipse.microprofile.reactive.messaging.Emitter;
 import javax.enterprise.context.ApplicationScoped;
 import javax.inject.Inject;
-import java.sql.Timestamp;
-import java.util.Date;
-import java.util.Random;
-import java.util.concurrent.TimeUnit;
 import org.jboss.logging.Logger;
+
 @ApplicationScoped
 public class MessageProducer {
 

quarkus-kafaka-homework/kafka-producer/src/main/java/me/jochrist/kafka/Producer.java

@@ -14,17 +14,6 @@ import org.jboss.resteasy.annotations.jaxrs.PathParam;
 @Path("/kafka")
 public class Producer {
 
-    @Inject
-    Vertx vertx;
-
-    /* Simple example
-    @GET
-    @Produces(MediaType.TEXT_PLAIN)
-    @Path("/message/{message}")
-    public String hello(@PathParam String message) {
-        return "hello " + message;
-    }
-    */
     @Inject MessageProducer messageProducer;
     @GET
     @Produces(MediaType.TEXT_PLAIN)
@@ -34,13 +23,6 @@ public class Producer {
         messageProducer.sendMessage(message);
     }
 
-    /*
-    public Uni<String> sendKafkaMessage()
-    {
-        return Uni.createFrom().item(() -> )
 
-    }
-     */
     @Inject TimeStampProducer timeStampProducer;
-
 }

quarkus-kafaka-homework/kafka-producer/src/main/java/me/jochrist/kafka/TimeStampProducer.java

@@ -4,13 +4,10 @@ import org.eclipse.microprofile.reactive.messaging.Outgoing;
 
 import javax.enterprise.context.ApplicationScoped;
 import java.sql.Timestamp;
-import java.util.Date;
-import java.util.Random;
 import java.util.concurrent.TimeUnit;
 
 @ApplicationScoped
 public class TimeStampProducer {
-    //private Timestamp timestamp = new Timestamp(System.currentTimeMillis());
 
     @Outgoing("timestamp")
     public Flowable<String> generate()

quarkus-kafaka-homework/kafka-producer/src/main/resources/application.properties

@@ -3,6 +3,7 @@
 quarkus.kubernetes-client.trust-certs=true
 kafka.bootstrap.servers=kafka-cluster-kafka-bootstrap:9092
 %dev.kafka.bootstrap.servers=localhost:9092
+%test.kafka.bootstrap.servers=localhost:9092
 %dev.quarkus.http.port=8080
 quarkus.http.port=8080
 quarkus.swagger-ui.always-include=true

quarkus-kafaka-homework/kafka-producer/src/test/java/me/jochrist/kafka/KafkaResource.java

@@ -0,0 +1,20 @@
+package me.jochrist.kafka;
+import java.util.Collections;
+import java.util.Map;
+import org.testcontainers.containers.KafkaContainer;
+import io.quarkus.test.common.QuarkusTestResourceLifecycleManager;
+
+public class KafkaResource implements QuarkusTestResourceLifecycleManager {
+    private final KafkaContainer kafka = new KafkaContainer();
+
+    @Override
+    public Map<String, String> start() {
+        kafka.start();
+        return Collections.singletonMap("kafka.bootstrap.servers", kafka.getBootstrapServers());
+    }
+
+    @Override
+    public void stop() {
+        kafka.close();
+    }
+}

quarkus-kafaka-homework/kafka-producer/src/test/java/me/jochrist/kafka/ProducerTest.java

@@ -5,18 +5,56 @@ import org.junit.jupiter.api.Test;
 
 import static io.restassured.RestAssured.given;
 import static org.hamcrest.CoreMatchers.is;
+import io.smallrye.reactive.messaging.connectors.InMemoryConnector;
+import io.smallrye.reactive.messaging.connectors.InMemorySource;
+import io.smallrye.reactive.messaging.connectors.InMemorySink;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import javax.enterprise.inject.Any;
+import javax.inject.Inject;
+import io.quarkus.test.common.QuarkusTestResource;
+
 
 @QuarkusTest
+@QuarkusTestResource(KafkaResource.class)
 public class ProducerTest {
 
-/*
+    @BeforeAll
- * @Test
+    public static void switchMyChannels() {
-    public void testHelloEndpoint() {
+        InMemoryConnector.switchIncomingChannelsToInMemory("timestamp");
+        InMemoryConnector.switchIncomingChannelsToInMemory("user-message");
+        InMemoryConnector.switchOutgoingChannelsToInMemory("timestamp");
+        InMemoryConnector.switchOutgoingChannelsToInMemory("user-message");
+    }
+
+    @AfterAll
+    public static void revertMyChannels() {
+        InMemoryConnector.clear();
+    }
+
+
+    //See https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/204
+    //As we dont need to return anything 204 is still a success
+    @Test
+    public void testMessage() {
         given()
-          .when().get("/kafka/message")
+          .when().get("/kafka/message/test")
           .then()
-             .statusCode(200)
+             .statusCode(204);
-             .body(is("hello"));
+        //InMemorySink<Integer> results = connector.sink("user-message");
+
+    }
+
+    @Inject @Any
+    InMemoryConnector connector;
+
+    @Test
+    public void testTimestamp()
+    {
+        InMemorySink<String> timestamps = connector.sink("timestamp");
+        System.out.println(timestamps);
     }
-*/
+
 }

quarkus-kafaka-homework/pom.xml

@@ -9,6 +9,21 @@
 
   <name>jochrist hacky Kafka project :: Parent</name>
 
+  <!--
+  For kafka deployments lets skip 'em as default
+  mvn -P docker to activate
+  -->
+    <profiles>
+        <profile>
+            <id>docker</id>
+            <activation>
+                <activeByDefault>false</activeByDefault>
+            </activation>
+            <modules>
+                <module>docker-resources</module>
+            </modules>
+        </profile>
+    </profiles>
   <modules>
     <module>kafka-consumer</module>
     <module>kafka-producer</module>
@@ -27,23 +42,10 @@
      <quarkus.platform.version>1.8.1.Final</quarkus.platform.version>
      <surefire-plugin.version>3.0.0-M5</surefire-plugin.version>
      <quarkus.container-image.builder>s2i</quarkus.container-image.builder>
+      <testcontainers.version>1.14.1</testcontainers.version>
  </properties>
 
-  <!--
+
-  For kafka deployments lets skip 'em as default
-  mvn -P docker to activate
-  -->
-  <profiles>
-    <profile>
-      <id>docker</id>
-      <activation>
-        <activeByDefault>false</activeByDefault>
-      </activation>
-      <modules>
-        <module>docker-resources</module>
-      </modules>
-    </profile>
-  </profiles>
   <scm>
     <connection>scm:git:ssh://git@gitlab.cee.redhat.com/FOO</connection>
     <tag>1.0-SNAPSHOT</tag>