feat: listeners and post processors

swisspost · Aug 23, 2021 · b6659c8 · b6659c8
1 parent 66de03e
commit b6659c8
Show file tree

Hide file tree

Showing 6 changed files with 34 additions and 8 deletions.
diff --git a/kobuka-client/pom.xml b/kobuka-client/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>kobuka</artifactId>
         <groupId>org.swisspush</groupId>
-        <version>1.0.0</version>
+        <version>1.0.1</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

diff --git a/kobuka-gen/pom.xml b/kobuka-gen/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>kobuka</artifactId>
         <groupId>org.swisspush</groupId>
-        <version>1.0.0</version>
+        <version>1.0.1</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

diff --git a/kobuka-spring/pom.xml b/kobuka-spring/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>kobuka</artifactId>
         <groupId>org.swisspush</groupId>
-        <version>1.0.0</version>
+        <version>1.0.1</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

diff --git a/...c/main/java/org/swisspush/kobuka/spring/internal/DefaultKafkaConsumerFactoryProvider.java b/...c/main/java/org/swisspush/kobuka/spring/internal/DefaultKafkaConsumerFactoryProvider.java
@@ -1,8 +1,11 @@
 package org.swisspush.kobuka.spring.internal;
 
 import net.karneim.pojobuilder.GeneratePojoBuilder;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.ConsumerPostProcessor;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 
+import java.util.List;
 import java.util.Map;
 
 public class DefaultKafkaConsumerFactoryProvider {
@@ -13,7 +16,17 @@ public class DefaultKafkaConsumerFactoryProvider {
             withGenerationGap = true,
             intoPackage = "org.swisspush.kobuka.spring"
     )
-    public static <K, V> DefaultKafkaConsumerFactory<K, V> createDefaultKafkaConsumerFactory(Map<String, Object> configs) {
-        return new DefaultKafkaConsumerFactory<K, V>(configs);
+    public static <K, V> DefaultKafkaConsumerFactory<K, V>
+    createDefaultKafkaConsumerFactory(Map<String, Object> configs,
+                                      List<ConsumerFactory.Listener<K, V>> withListeners,
+                                      List<ConsumerPostProcessor<K, V>> withPostProcessors) {
+        DefaultKafkaConsumerFactory<K, V> result = new DefaultKafkaConsumerFactory<>(configs);
+        if(withListeners != null) {
+            withListeners.forEach(result::addListener);
+        }
+        if(withPostProcessors != null) {
+            withPostProcessors.forEach(result::addPostProcessor);
+        }
+        return result;
     }
 }
diff --git a/...c/main/java/org/swisspush/kobuka/spring/internal/DefaultKafkaProducerFactoryProvider.java b/...c/main/java/org/swisspush/kobuka/spring/internal/DefaultKafkaProducerFactoryProvider.java
@@ -2,7 +2,10 @@
 
 import net.karneim.pojobuilder.GeneratePojoBuilder;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.core.ProducerPostProcessor;
 
+import java.util.List;
 import java.util.Map;
 
 public class DefaultKafkaProducerFactoryProvider {
@@ -13,7 +16,17 @@ public class DefaultKafkaProducerFactoryProvider {
             withGenerationGap = true,
             intoPackage = "org.swisspush.kobuka.spring"
     )
-    public static <K, V> DefaultKafkaProducerFactory<K, V> create(Map<String, Object> configs) {
-        return new DefaultKafkaProducerFactory<K, V>(configs);
+    public static <K, V> DefaultKafkaProducerFactory<K, V> create(Map<String, Object> configs,
+                                                                  List<ProducerFactory.Listener<K, V>> withListeners,
+                                                                  List<ProducerPostProcessor<K, V>> withPostProcessors)
+    {
+        DefaultKafkaProducerFactory<K, V> result = new DefaultKafkaProducerFactory<>(configs);
+        if(withListeners != null) {
+            withListeners.forEach(result::addListener);
+        }
+        if(withPostProcessors != null) {
+            withPostProcessors.forEach(result::addPostProcessor);
+        }
+        return result;
     }
 }
diff --git a/pom.xml b/pom.xml
@@ -7,7 +7,7 @@
     <groupId>org.swisspush</groupId>
     <artifactId>kobuka</artifactId>
     <packaging>pom</packaging>
-    <version>1.0.0</version>
+    <version>1.0.1</version>
 
     <build>
         <plugins>