Version des `spring-producer`, der JSON-Nachrichten sendet

diff --git a/README.sh b/README.sh
index 1d208d534a05f579cb7a49e78aecf53c9699b79c..3b52a5ddc3ca4e592a23ff5bc37cd059e9fc0478 100755 (executable)
--- a/README.sh
+++ b/README.sh
@@ -1,6 +1,6 @@
 #!/bin/bash
 
-IMAGE=juplo/spring-producer:2.0-SNAPSHOT
+IMAGE=juplo/spring-producer:2.0-json-SNAPSHOT
 
 if [ "$1" = "cleanup" ]
 then

diff --git a/build.gradle b/build.gradle
index dd25d2857f00a9f013d399e0005ec6a81919b630..88937de37171cf0d1732eebfd539e454202a77fd 100644 (file)
--- a/build.gradle
+++ b/build.gradle
@@ -8,7 +8,7 @@ plugins {
 }
 
 group = 'de.juplo.kafka'
-version = '2.0-SNAPSHOT'
+version = '2.0-json-SNAPSHOT'
 
 java {
        toolchain {

diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml
index d151dab5dd111e95b65cfc4d4c92b380e45ea14b..5be08a7b33bd72aff50bdd8161440f566d2b5e22 100644 (file)
--- a/docker/docker-compose.yml
+++ b/docker/docker-compose.yml
@@ -173,7 +173,7 @@ services:
       - kafka-3
 
   producer:
-    image: juplo/spring-producer:2.0-SNAPSHOT
+    image: juplo/spring-producer:2.0-json-SNAPSHOT
     environment:
       spring.kafka.bootstrap-servers: kafka:9092
       spring.kafka.client-id: producer

diff --git a/pom.xml b/pom.xml
index 77011f97d373b55bfa4643fc25a3d30988f4b0e6..c7bca0dc02adee6a6b311e8483924d1b2e5ea5b9 100644 (file)
--- a/pom.xml
+++ b/pom.xml
@@ -15,7 +15,7 @@
   <artifactId>spring-producer</artifactId>
   <name>Spring Producer</name>
   <description>A Simple Producer, based on Spring Boot, that sends messages via Kafka</description>
-  <version>2.0-SNAPSHOT</version>
+  <version>2.0-json-SNAPSHOT</version>
 
   <properties>
     <java.version>21</java.version>

diff --git a/src/main/java/de/juplo/kafka/AddNumberMessage.java b/src/main/java/de/juplo/kafka/AddNumberMessage.java
new file mode 100644 (file)
index 0000000..deb6350
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/AddNumberMessage.java
@@ -0,0 +1,11 @@
+package de.juplo.kafka;
+
+import lombok.Value;
+
+
+@Value
+public class AddNumberMessage implements SumupMessage
+{
+  private final int number;
+  private final int next;
+}

diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
index efdfafa179d6e0c27c16cdeb15ea8d0e7666f506..e212a2533e9254405963d98da2d473d567bd65e8 100644 (file)
--- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
@@ -19,7 +19,7 @@ public class ApplicationConfiguration
   public ExampleProducer exampleProducer(
     @Value("${spring.kafka.client-id}") String clientId,
     ApplicationProperties properties,
-    Producer<String, String> kafkaProducer,
+    Producer<String, SumupMessage> kafkaProducer,
     ConfigurableApplicationContext applicationContext)
   {
     return

diff --git a/src/main/java/de/juplo/kafka/CalculateSumMessage.java b/src/main/java/de/juplo/kafka/CalculateSumMessage.java
new file mode 100644 (file)
index 0000000..6aa0121
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/CalculateSumMessage.java
@@ -0,0 +1,11 @@
+package de.juplo.kafka;
+
+
+import lombok.Value;
+
+
+@Value
+public class CalculateSumMessage implements SumupMessage
+{
+  private final int number;
+}

diff --git a/src/main/java/de/juplo/kafka/ExampleProducer.java b/src/main/java/de/juplo/kafka/ExampleProducer.java
index 93d0d1738a3ce0eeabb83cd040c01b41dac4bda9..04a0a99235314a830ad9bb062403ee32564c35b7 100644 (file)
--- a/src/main/java/de/juplo/kafka/ExampleProducer.java
+++ b/src/main/java/de/juplo/kafka/ExampleProducer.java
@@ -13,7 +13,7 @@ public class ExampleProducer implements Runnable
   private final String id;
   private final String topic;
   private final Duration throttle;
-  private final Producer<String, String> producer;
+  private final Producer<String, SumupMessage> producer;
   private final Thread workerThread;
   private final Runnable closeCallback;
 
@@ -25,7 +25,7 @@ public class ExampleProducer implements Runnable
     String id,
     String topic,
     Duration throttle,
-    Producer<String, String> producer,
+    Producer<String, SumupMessage> producer,
     Runnable closeCallback)
   {
     this.id = id;
@@ -43,13 +43,18 @@ public class ExampleProducer implements Runnable
   @Override
   public void run()
   {
-    long i = 0;
+    int i = 0;
 
     try
     {
       for (; running; i++)
       {
-        send(Long.toString(i%10), Long.toString(i));
+        int number = i % 10;
+        SumupMessage message = (i % 7 == 0)
+          ? new CalculateSumMessage(number)
+          : new AddNumberMessage(number, i);
+
+        send(Long.toString(number), message);
 
         if (throttle.isPositive())
         {
@@ -78,11 +83,11 @@ public class ExampleProducer implements Runnable
     }
   }
 
-  void send(String key, String value)
+  void send(String key, SumupMessage value)
   {
     final long sendRequested = System.currentTimeMillis();
 
-    final ProducerRecord<String, String> record = new ProducerRecord<>(
+    final ProducerRecord<String, SumupMessage> record = new ProducerRecord<>(
       topic,  // Topic
       key,    // Key
       value   // Value

diff --git a/src/main/java/de/juplo/kafka/SumupMessage.java b/src/main/java/de/juplo/kafka/SumupMessage.java
new file mode 100644 (file)
index 0000000..739efd1
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/SumupMessage.java
@@ -0,0 +1,5 @@
+package de.juplo.kafka;
+
+public interface SumupMessage
+{
+}

diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
index 3f6c233a1751059d20f011c9b365a87000206669..12dc5e8a999f3d18542c03a301c24395ada7e395 100644 (file)
--- a/src/main/resources/application.yml
+++ b/src/main/resources/application.yml
@@ -11,7 +11,11 @@ spring:
       buffer-memory: 33554432
       batch-size: 16384
       compression-type: gzip
+      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
       properties:
+        spring.json.type.mapping: >-
+          ADD:de.juplo.kafka.AddNumberMessage,
+          CALC:de.juplo.kafka.CalculateSumMessage
         metadata.max.age.ms: 5000
         request.timeout.ms: 5000
         delivery.timeout.ms: 10000