From: Kai Moritz <kai@juplo.de>
Date: Mon, 15 Aug 2022 18:40:16 +0000 (+0200)
Subject: Benennung vereinheitlicht und projektunabhängig gemacht
X-Git-Tag: sumup-requests---lvm-2-tage~3
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=77cbd2b55e26b39be90a722761ebb6e59e8f965a;p=demos%2Fkafka%2Ftraining

Benennung vereinheitlicht und projektunabhängig gemacht
---

diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
index 753422e..60f45a9 100644
--- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
@@ -19,11 +19,11 @@ import java.util.concurrent.Executors;
 public class ApplicationConfiguration
 {
   @Bean
-  public SumUpRecordHandler sumUpRecordHandler(
+  public ApplicationRecordHandler recordHandler(
       KafkaProducer<String, String> kafkaProducer,
       ApplicationProperties properties)
   {
-    return new SumUpRecordHandler(
+    return new ApplicationRecordHandler(
         kafkaProducer,
         properties.getClientId(),
         properties.getTopicOut());
@@ -33,7 +33,7 @@ public class ApplicationConfiguration
   public EndlessConsumer<String, Integer> endlessConsumer(
       KafkaConsumer<String, Integer> kafkaConsumer,
       ExecutorService executor,
-      SumUpRecordHandler sumUpRecordHandler,
+      ApplicationRecordHandler recordHandler,
       ApplicationProperties properties)
   {
     return
@@ -42,7 +42,7 @@ public class ApplicationConfiguration
             properties.getClientId(),
             properties.getTopicIn(),
             kafkaConsumer,
-            sumUpRecordHandler);
+            recordHandler);
   }
 
   @Bean
diff --git a/src/main/java/de/juplo/kafka/ApplicationRecordHandler.java b/src/main/java/de/juplo/kafka/ApplicationRecordHandler.java
new file mode 100644
index 0000000..a4b6a8f
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/ApplicationRecordHandler.java
@@ -0,0 +1,82 @@
+package de.juplo.kafka;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+
+
+@RequiredArgsConstructor
+@Slf4j
+public class ApplicationRecordHandler implements RecordHandler<String, Integer>
+{
+  private final Producer<String, String> producer;
+  private final String id;
+  private final String topic;
+
+
+  @Override
+  public void accept(ConsumerRecord<String, Integer> record)
+  {
+    String key = record.key();
+    int number = record.value();
+
+    send(key, "START");
+    for (int i = 1; i <= number; i++)
+    {
+      send(key, Integer.toString(i));
+    }
+    send(key, "END");
+  }
+
+  private void send(String key, String value)
+  {
+      final long time = System.currentTimeMillis();
+
+      final ProducerRecord<String, String> record = new ProducerRecord<>(
+          topic,  // Topic
+          key,    // Key
+          value   // Value
+      );
+
+      producer.send(record, (metadata, e) ->
+      {
+        long now = System.currentTimeMillis();
+        if (e == null)
+        {
+          // HANDLE SUCCESS
+          log.debug(
+              "{} - Sent key={} message={} partition={}/{} timestamp={} latency={}ms",
+              id,
+              record.key(),
+              record.value(),
+              metadata.partition(),
+              metadata.offset(),
+              metadata.timestamp(),
+              now - time
+          );
+        }
+        else
+        {
+          // HANDLE ERROR
+          log.error(
+              "{} - ERROR key={} timestamp={} latency={}ms: {}",
+              id,
+              record.key(),
+              metadata == null ? -1 : metadata.timestamp(),
+              now - time,
+              e.toString()
+          );
+        }
+      });
+
+      long now = System.currentTimeMillis();
+      log.trace(
+          "{} - Queued message with key={} latency={}ms",
+          id,
+          record.key(),
+          now - time
+      );
+  }
+}
diff --git a/src/main/java/de/juplo/kafka/SumUpRecordHandler.java b/src/main/java/de/juplo/kafka/SumUpRecordHandler.java
deleted file mode 100644
index 5d15b3b..0000000
--- a/src/main/java/de/juplo/kafka/SumUpRecordHandler.java
+++ /dev/null
@@ -1,82 +0,0 @@
-package de.juplo.kafka;
-
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerRecord;
-
-
-@RequiredArgsConstructor
-@Slf4j
-public class SumUpRecordHandler implements RecordHandler<String, Integer>
-{
-  private final Producer<String, String> producer;
-  private final String id;
-  private final String topic;
-
-
-  @Override
-  public void accept(ConsumerRecord<String, Integer> record)
-  {
-    String key = record.key();
-    int number = record.value();
-
-    send(key, "START");
-    for (int i = 1; i <= number; i++)
-    {
-      send(key, Integer.toString(i));
-    }
-    send(key, "END");
-  }
-
-  private void send(String key, String value)
-  {
-      final long time = System.currentTimeMillis();
-
-      final ProducerRecord<String, String> record = new ProducerRecord<>(
-          topic,  // Topic
-          key,    // Key
-          value   // Value
-      );
-
-      producer.send(record, (metadata, e) ->
-      {
-        long now = System.currentTimeMillis();
-        if (e == null)
-        {
-          // HANDLE SUCCESS
-          log.debug(
-              "{} - Sent key={} message={} partition={}/{} timestamp={} latency={}ms",
-              id,
-              record.key(),
-              record.value(),
-              metadata.partition(),
-              metadata.offset(),
-              metadata.timestamp(),
-              now - time
-          );
-        }
-        else
-        {
-          // HANDLE ERROR
-          log.error(
-              "{} - ERROR key={} timestamp={} latency={}ms: {}",
-              id,
-              record.key(),
-              metadata == null ? -1 : metadata.timestamp(),
-              now - time,
-              e.toString()
-          );
-        }
-      });
-
-      long now = System.currentTimeMillis();
-      log.trace(
-          "{} - Queued message with key={} latency={}ms",
-          id,
-          record.key(),
-          now - time
-      );
-  }
-}