tomasz
/
code-examples
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

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

master
Tomasz Polgrabia 2024-12-27 21:19:58 +01:00
parent 9e8a692eb4
commit 4ad8473aaa
4 changed files with 38 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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();
}
}

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

@ -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;
}

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

@ -3,6 +3,8 @@ 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;
@ -15,11 +17,16 @@ public class WebsocketServerHandler implements Handler<ServerWebSocket> {
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<>();
@ -50,8 +57,10 @@ public class WebsocketServerHandler implements Handler<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);
}
@ -62,6 +71,7 @@ public class WebsocketServerHandler implements Handler<ServerWebSocket> {
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
}
}

9
2024/08/chat_demo1/tools/docker-compose.yaml
View File

@ -5,13 +5,16 @@ services:
environment:
KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_LISTENERS: PLAINTEXT://localhost:9092,CONTROLLER://localhost:9093
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
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@localhost:9093
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"