Added stubs for kafka connection but configuration (wsl + windows) needs to be checked.

2024/08/chat_demo1/chat_demo1_web/build.gradle.kts

@@ -12,6 +12,7 @@ repositories {
 dependencies {
     implementation("io.vertx:vertx-core:4.5.9")
     implementation("ch.qos.logback:logback-classic:1.5.6")
+    implementation("org.apache.kafka:kafka-clients:3.9.0")
     testImplementation(platform("org.junit:junit-bom:5.10.0"))
     testImplementation("org.junit.jupiter:junit-jupiter")
 }

2024/08/chat_demo1/chat_demo1_web/src/main/java/ch/polgrabia/demos/server/WebsocketServerApp.java

@@ -1,19 +1,29 @@
 package ch.polgrabia.demos.server;
 
 import io.vertx.core.Vertx;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 public class WebsocketServerApp {
-    private static final Logger logger = LoggerFactory.getLogger(WebsocketServerApp.class);
-
     public static void main(String[] args) {
+        /**
+        Properties props = new Properties();
+        props.put("bootstrap.servers", "localhost:9092");
+        props.put("linger.ms", 1);
+        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+        props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
+
+        Producer<String, byte[]> kafkaProducer = new KafkaProducer<>(props);
+        **/
+
         var vertx = Vertx.vertx();
-        WebsocketServerChatVerticle websocketServerChatVerticle = new WebsocketServerChatVerticle(8080, "localhost");
+        WebsocketServerChatVerticle websocketServerChatVerticle = new WebsocketServerChatVerticle(
+                8080,
+                "localhost",
+                null); // kafka producer
         vertx.deployVerticle(websocketServerChatVerticle);
         websocketServerChatVerticle
                 .waitUntilFinished()
                 .join();
         vertx.close();
+        // kafkaProducer.close();
     }
 }

2024/08/chat_demo1/chat_demo1_web/src/main/java/ch/polgrabia/demos/server/WebsocketServerChatVerticle.java

@@ -2,6 +2,7 @@ package ch.polgrabia.demos.server;
 
 import io.vertx.core.AbstractVerticle;
 import io.vertx.core.http.HttpServer;
+import org.apache.kafka.clients.producer.Producer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -9,6 +10,7 @@ import java.util.concurrent.CompletableFuture;
 
 public class WebsocketServerChatVerticle extends AbstractVerticle {
     private static final Logger logger = LoggerFactory.getLogger(WebsocketServerChatVerticle.class);
+    private final Producer<String, byte[]> kafkaProducer;
     private HttpServer httpServer;
     private final int port;
     private final String hostname;
@@ -16,16 +18,17 @@ public class WebsocketServerChatVerticle extends AbstractVerticle {
     private final Object runningMonitor = new Object();
     private boolean shutdownActivation = false;
 
-    public WebsocketServerChatVerticle(int port, String hostname) {
+    public WebsocketServerChatVerticle(int port, String hostname, Producer<String, byte[]> kafkaProducer) {
         this.port = port;
         this.hostname = hostname;
+        this.kafkaProducer = kafkaProducer;
     }
 
     @Override
     public void start() throws Exception {
         httpServer = getVertx()
                 .createHttpServer();
-        httpServer.webSocketHandler(new WebsocketServerHandler()).listen(port, hostname);
+        httpServer.webSocketHandler(new WebsocketServerHandler(kafkaProducer)).listen(port, hostname);
         shutdownActivation = true;
     }
 

2024/08/chat_demo1/chat_demo1_web/src/main/java/ch/polgrabia/demos/server/WebsocketServerHandler.java

@@ -3,31 +3,75 @@ package ch.polgrabia.demos.server;
 import io.vertx.core.Handler;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.ServerWebSocket;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
 public class WebsocketServerHandler implements Handler<ServerWebSocket> {
     private static final Logger logger = LoggerFactory.getLogger(WebsocketServerHandler.class);
 
+    private static final ConcurrentHashMap<String, Set<ServerWebSocket>>
+            channelSocketMapping = new ConcurrentHashMap<>();
+    private final Producer<String, byte[]> kafkaProducer;
+
+    public WebsocketServerHandler(Producer<String, byte[]> kafkaProducer) {
+        this.kafkaProducer = kafkaProducer;
+    }
+
     @Override
     public void handle(ServerWebSocket serverWebSocket) {
         serverWebSocket.writeBinaryMessage(Buffer.buffer("Hello"));
-        serverWebSocket.handler(new WebsocketMessageHandler(serverWebSocket));
+        serverWebSocket.handler(new WebsocketMessageHandler(kafkaProducer, serverWebSocket));
+        String channel = serverWebSocket.path();
+        Set<ServerWebSocket> sockets = channelSocketMapping.computeIfAbsent(channel, (k) -> {
+            Set<ServerWebSocket> s = new HashSet<>();
+            s.add(serverWebSocket);
+            return s;
+        });
+
+
+        serverWebSocket.closeHandler(aVoid -> {
+            int id = System.identityHashCode(serverWebSocket);
+            logger.info("[{}] Closing web socket with id {}", channel, id);
+            Set<ServerWebSocket> serverWebSockets = channelSocketMapping.get(channel);
+            if (serverWebSockets != null) {
+                boolean removed = serverWebSockets
+                        .remove(serverWebSocket);
+                if (!removed) {
+                    logger.warn("Server websocket channel {} was not found in the channel set of sockets." +
+                            " It should never happen", channel);
+                }
+            } else {
+                logger.warn("Channel {} not found. It should never happen", channel);
+            }
+        });
+
+        sockets.add(serverWebSocket);
     }
 
     static class WebsocketMessageHandler implements Handler<Buffer> {
 
         private final ServerWebSocket serverWebSocket;
+        private final Producer<String, byte[]> kafkaProducer;
 
-        public WebsocketMessageHandler(ServerWebSocket serverWebSocket) {
+        public WebsocketMessageHandler(Producer<String, byte[]> kafkaProducer, ServerWebSocket serverWebSocket) {
+            this.kafkaProducer = kafkaProducer;
             this.serverWebSocket = serverWebSocket;
             this.serverWebSocket.setWriteQueueMaxSize(0);
         }
 
+
         @Override
         public void handle(Buffer s) {
-            logger.info("[{}] Got message {}", serverWebSocket.path(), s);
+            String channel = serverWebSocket.path();
+            logger.info("[{}] Got message {}", channel, s);
             serverWebSocket.writeBinaryMessage(Buffer.buffer("Pong: " + s.toString() + "\n")); // TBD send to kafka
+            // kafkaProducer.send(new ProducerRecord<>(channel, channel, s.getBytes())); // kafka setup needs to be checked
         }
     }
 

2024/08/chat_demo1/tools/docker-compose.yaml

@@ -0,0 +1,20 @@
+services:
+  broker:
+    image: apache/kafka:latest
+    container_name: broker
+    environment:
+      KAFKA_NODE_ID: 1
+      KAFKA_PROCESS_ROLES: broker,controller
+      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://0.0.0.0:9092
+      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@0.0.0.0:9093
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_NUM_PARTITIONS: 3
+    ports:
+      - "9092:9092"
+      - "9093:9093"