feat: Enable custom message for produce

Signed-off-by: Grace Jansen <grace.jansen1@ibm.com>

Author: gljrobinson

src/main/java/kafka/vertx/demo/Main.java

@@ -27,7 +27,6 @@ public static void main(String[] args) {
 
     // Set vertx timeout to deal with slow DNS connections
     System.setProperty(ResolverProvider.DISABLE_DNS_RESOLVER_PROP_NAME, "true");
-
     Vertx vertx = Vertx.vertx(
       new VertxOptions()
         .setWarningExceptionTime(10).setWarningExceptionTimeUnit(TimeUnit.SECONDS)

src/main/java/kafka/vertx/demo/PeriodicProducer.java

@@ -24,6 +24,7 @@
 public class PeriodicProducer extends AbstractVerticle {
 
   private static final Logger logger = LoggerFactory.getLogger(PeriodicProducer.class);
+  private String customMessage;
 
   @Override
   public void start(Promise<Void> startPromise) {
@@ -41,34 +42,33 @@ public void start(Promise<Void> startPromise) {
   private void setup(JsonObject config, Promise<Void> startPromise) {
     HashMap<String, String> props = new HashMap<>();
     config.forEach(entry -> props.put(entry.getKey(), entry.getValue().toString()));
-    KafkaProducer<String, JsonObject> kafkaProducer = KafkaProducer.create(vertx, props);
+    KafkaProducer<String, String> kafkaProducer = KafkaProducer.create(vertx, props);
 
     TimeoutStream timerStream = vertx.periodicStream(2000);
     timerStream.handler(tick -> produceKafkaRecord(kafkaProducer));
     timerStream.pause();
 
-    vertx.eventBus().<String>consumer(Main.PERIODIC_PRODUCER_ADDRESS, message -> handleCommand(timerStream, message));
+    vertx.eventBus().<JsonObject>consumer(Main.PERIODIC_PRODUCER_ADDRESS, message -> handleCommand(timerStream, message));
     logger.info("🚀 PeriodicConsumer started");
     startPromise.complete();
   }
 
-  private void handleCommand(TimeoutStream timerStream, Message<String> message) {
-    String command = message.body();
+  private void handleCommand(TimeoutStream timerStream, Message<JsonObject> message) {
+    String command = message.body().getString("action", "none");
     if ("start".equals(command)) {
       logger.info("Producing Kafka records");
+      customMessage = message.body().getString("custom", "Hello World");
       timerStream.resume();
     } else if ("stop".equals(command)) {
       logger.info("Stopping producing Kafka records");
       timerStream.pause();
     }
   }
 
-  private void produceKafkaRecord(KafkaProducer<String, JsonObject> kafkaProducer) {
-    JsonObject payload = new JsonObject()
-      .put("type", "tick")
-      .put("when", Instant.now().toString());
+  private void produceKafkaRecord(KafkaProducer<String, String> kafkaProducer) {
+    String payload = customMessage;
 
-    KafkaProducerRecord<String, JsonObject> record = KafkaProducerRecord.create(Main.TOPIC, payload);
+    KafkaProducerRecord<String, String> record = KafkaProducerRecord.create(Main.TOPIC, payload);
 
     kafkaProducer
       .send(record)
@@ -81,6 +81,6 @@ private void produceKafkaRecord(KafkaProducer<String, JsonObject> kafkaProducer)
           .put("value", payload);
         vertx.eventBus().send(Main.PERIODIC_PRODUCER_BROADCAST, kafkaMetaData);
       })
-      .onFailure(err -> logger.error("Error sending {}", payload.encode(), err));
+      .onFailure(err -> logger.error("Error sending {}", payload, err));
   }
 }

src/main/java/kafka/vertx/demo/WebSocketServer.java

@@ -85,9 +85,10 @@ private void handleProduceSocket(ServerWebSocket webSocket) {
     EventBus eventBus = vertx.eventBus();
 
     webSocket.handler(buffer -> {
-      String action = buffer.toJsonObject().getString("action", "none");
+      JsonObject message = buffer.toJsonObject();
+      String action = message.getString("action", "none");
       if ("start".equals(action) || "stop".equals(action)) {
-        eventBus.send(Main.PERIODIC_PRODUCER_ADDRESS, action);
+        eventBus.send(Main.PERIODIC_PRODUCER_ADDRESS, message);
       }
     });
 

src/main/resources/kafka.properties

@@ -5,12 +5,12 @@ group.id=my_group
 auto.offset.reset=earliest
 enable.auto.commit=false
 key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
-value.deserializer=io.vertx.kafka.client.serialization.JsonObjectDeserializer
+value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
 
 ## Producer specific properties
 acks=1
 key.serializer=org.apache.kafka.common.serialization.StringSerializer
-value.serializer=io.vertx.kafka.client.serialization.JsonObjectSerializer
+value.serializer=org.apache.kafka.common.serialization.StringSerializer
 
 ## Optional security options
 # security.protocol=SSL