kafka changes

goafabric · Jul 5, 2024 · a368fab · a368fab
1 parent 1a7d253
commit a368fab
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 8 deletions.
diff --git a/src/main/java/org/goafabric/eventdispatcher/consumer/kafka/CalendarConsumer.java b/src/main/java/org/goafabric/eventdispatcher/consumer/kafka/CalendarConsumer.java
@@ -14,9 +14,9 @@ public class CalendarConsumer {
 
     static final String CONSUMER_NAME = "Calendar";
 
-    @KafkaListener(groupId = CONSUMER_NAME, topics = {"patient.create", "patient.update"})
-    @KafkaListener(groupId = CONSUMER_NAME, topics = {"practitioner.create", "practitioner.update"})
-    public void processKafka(@Header(KafkaHeaders.RECEIVED_TOPIC) String key, EventData eventData) {
+    @KafkaListener(groupId = CONSUMER_NAME, topics = "patient")
+    @KafkaListener(groupId = CONSUMER_NAME, topics = "practitioner")
+    public void processKafka(@Header(KafkaHeaders.RECEIVED_KEY) String key, EventData eventData) {
         process(key, eventData);
     }
 

diff --git a/src/main/java/org/goafabric/eventdispatcher/consumer/kafka/LoggerConsumer.java b/src/main/java/org/goafabric/eventdispatcher/consumer/kafka/LoggerConsumer.java
@@ -14,8 +14,8 @@ public class LoggerConsumer {
 
     static final String CONSUMER_NAME = "Logger";
 
-    @KafkaListener(groupId = CONSUMER_NAME, topicPattern = ".*\\..*")
-    public void processKafka(@Header(KafkaHeaders.RECEIVED_TOPIC) String key, EventData eventData) {
+    @KafkaListener(groupId = CONSUMER_NAME, topicPattern = ".*")
+    public void processKafka(@Header(KafkaHeaders.RECEIVED_KEY) String key, EventData eventData) {
         process(key, eventData);
     }
 

diff --git a/src/main/java/org/goafabric/eventdispatcher/producer/EventProducerKafka.java b/src/main/java/org/goafabric/eventdispatcher/producer/EventProducerKafka.java
@@ -23,12 +23,12 @@ public EventProducerKafka(KafkaTemplate kafkaTemplate, HashMap<String, String> e
     }
 
     public void produce(ChangeEvent changeEvent) {
-        send(changeEvent.type().toLowerCase() + "."  + changeEvent.operation().toString().toLowerCase(),
+        send(changeEvent.type().toLowerCase(), changeEvent.type().toLowerCase() + "."  + changeEvent.operation().toString().toLowerCase(),
                 changeEvent.referenceId());
     }
 
-    private void send(String topic, String referenceId) {
-        kafkaTemplate.send(topic, referenceId, new EventData(referenceId));
+    private void send(String topic, String key, String referenceId) {
+        kafkaTemplate.send(topic, key, new EventData(referenceId));
     }
 
 }