From: Kai Moritz <kai@juplo.de>
Date: Tue, 31 May 2022 04:17:07 +0000 (+0200)
Subject: Für PUT anstelle von POST wird eine `FooMessage` als `foo` verschickt
X-Git-Tag: bytestream-producer~8
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=459f64072f3b657990a3bda0d39306208702d537;p=demos%2Fkafka%2Ftraining

Für PUT anstelle von POST wird eine `FooMessage` als `foo` verschickt
---

diff --git a/README.sh b/README.sh
index 56ddbb5..832a42f 100755
--- a/README.sh
+++ b/README.sh
@@ -31,6 +31,7 @@ sleep 15
 
 echo 'Hallo Welt!' | http -v :8080/peter
 echo peter | http -v :8080/
+http -v PUT :8080/peter
 dd if=/dev/zero bs=1024 count=1024  | http -v :8080/fehler
 http -v :8081/seen
 
diff --git a/src/main/java/de/juplo/kafka/FooMessage.java b/src/main/java/de/juplo/kafka/FooMessage.java
new file mode 100644
index 0000000..2e9e8ba
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/FooMessage.java
@@ -0,0 +1,11 @@
+package de.juplo.kafka;
+
+import lombok.Value;
+
+
+@Value
+public class FooMessage
+{
+  private final String client;
+  private final Long timestamp;
+}
diff --git a/src/main/java/de/juplo/kafka/RestProducer.java b/src/main/java/de/juplo/kafka/RestProducer.java
index 0ecd667..70b327a 100644
--- a/src/main/java/de/juplo/kafka/RestProducer.java
+++ b/src/main/java/de/juplo/kafka/RestProducer.java
@@ -42,6 +42,7 @@ public class RestProducer
     props.put("value.serializer", JsonSerializer.class.getName());
     props.put(JsonSerializer.TYPE_MAPPINGS,
         "message:" + ClientMessage.class.getName() + "," +
+        "foo:" + FooMessage.class.getName() + "," +
         "greeting:" + Greeting.class.getName());
 
     this.producer = new KafkaProducer<>(props);
@@ -62,6 +63,19 @@ public class RestProducer
     return send(record);
   }
 
+  @PutMapping(path = "{key}")
+  public DeferredResult<ProduceResult> message(@PathVariable String key)
+  {
+    key = key.trim();
+    final ProducerRecord<String, Object> record = new ProducerRecord<>(
+        topic,  // Topic
+        key,    // Key
+        new FooMessage(key, System.currentTimeMillis()) // Value
+    );
+
+    return send(record);
+  }
+
   @PostMapping(path = "/")
   public DeferredResult<ProduceResult> greeting(
       @RequestBody String name)
diff --git a/src/test/java/de/juplo/kafka/ApplicationTests.java b/src/test/java/de/juplo/kafka/ApplicationTests.java
index 71625fd..cd7d928 100644
--- a/src/test/java/de/juplo/kafka/ApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/ApplicationTests.java
@@ -61,6 +61,17 @@ public class ApplicationTests
 				.until(() -> consumer.received.size() == 1);
 	}
 
+	@Test
+	void testSendFooMessage() throws Exception
+	{
+		mockMvc
+				.perform(put("/peter"))
+				.andExpect(status().isOk());
+		await("Message was send")
+				.atMost(Duration.ofSeconds(5))
+				.until(() -> consumer.received.size() == 1);
+	}
+
 	@Test
 	void testSendGreeting() throws Exception
 	{