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

Restored working but it needs to be run on the same wsl box.

master
Tomasz Polgrabia 2024-12-27 22:00:03 +01:00
parent 4ad8473aaa
commit a6ea81505c
3 changed files with 10 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,10 +1,13 @@
package ch.polgrabia.demos.server; package ch.polgrabia.demos.server;
import io.vertx.core.Vertx; import io.vertx.core.Vertx;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import java.util.Properties;
public class WebsocketServerApp { public class WebsocketServerApp {
public static void main(String[] args) { public static void main(String[] args) {
/**
Properties props = new Properties(); Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092"); props.put("bootstrap.servers", "localhost:9092");
props.put("linger.ms", 1); props.put("linger.ms", 1);
@ -12,7 +15,7 @@ public class WebsocketServerApp {
props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
Producer<String, byte[]> kafkaProducer = new KafkaProducer<>(props); Producer<String, byte[]> kafkaProducer = new KafkaProducer<>(props);
**/
var vertx = Vertx.vertx(); var vertx = Vertx.vertx();
WebsocketServerChatVerticle websocketServerChatVerticle = new WebsocketServerChatVerticle( WebsocketServerChatVerticle websocketServerChatVerticle = new WebsocketServerChatVerticle(
@ -24,6 +27,6 @@ public class WebsocketServerApp {
.waitUntilFinished() .waitUntilFinished()
.join(); .join();
vertx.close(); vertx.close();
// kafkaProducer.close(); kafkaProducer.close();
} }
} }

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

@ -71,7 +71,7 @@ public class WebsocketServerHandler implements Handler<ServerWebSocket> {
String channel = serverWebSocket.path(); String channel = serverWebSocket.path();
logger.info("[{}] Got message {}", channel, s); logger.info("[{}] Got message {}", channel, s);
serverWebSocket.writeBinaryMessage(Buffer.buffer("Pong: " + s.toString() + "\n")); // TBD send to kafka 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 kafkaProducer.send(new ProducerRecord<>(channel, channel, s.getBytes())); // kafka setup needs to be checked
} }
} }

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

@ -5,11 +5,11 @@ services:
environment: environment:
KAFKA_NODE_ID: 1 KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: broker,controller KAFKA_PROCESS_ROLES: broker,controller
KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://0.0.0.0:9092 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://:9092
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@0.0.0.0:9093 KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1