feat: implement setUpdateCallback(Consumer func) for Watcher interface (

#2)
jcasbin · Nov 5, 2023 · 660ae42 · 660ae42
1 parent 75a6a50
commit 660ae42
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 6 deletions.
diff --git a/pom.xml b/pom.xml
@@ -21,15 +21,15 @@
             <artifactId>slf4j-api</artifactId>
             <version>${slf4j.version}</version>
         </dependency>
-<!--        <dependency>-->
-<!--            <groupId>org.slf4j</groupId>-->
-<!--            <artifactId>slf4j-simple</artifactId>-->
-<!--            <version>${slf4j.version}</version>-->
-<!--        </dependency>-->
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-simple</artifactId>
+            <version>${slf4j.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.casbin</groupId>
             <artifactId>jcasbin</artifactId>
-            <version>1.6.3</version>
+            <version>1.13.3</version>
         </dependency>
         <dependency>
             <groupId>junit</groupId>

diff --git a/src/main/java/org/casbin/watcher/KafkaWatcher.java b/src/main/java/org/casbin/watcher/KafkaWatcher.java
@@ -16,9 +16,11 @@
 import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
+import java.util.function.Consumer;
 
 public class KafkaWatcher implements Watcher{
     private Runnable updateCallback;
+    private Consumer<String> updateConsumerCallback;
     private final String localId;
     private final Map<String,Object> producerProps;
     private final Map<String,Object> consumerProps;
@@ -38,6 +40,11 @@ public void setUpdateCallback(Runnable runnable) {
         this.updateCallback=runnable;
     }
 
+    @Override
+    public void setUpdateCallback(Consumer<String> consumer) {
+        this.updateConsumerCallback = consumer;
+    }
+
     @Override
     public void update() {
         KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
@@ -72,6 +79,9 @@ public void startSub(){
                         log.info(String.format("topic: %s, partition: %s, offset: %s, key: %s, value: %s",
                                 record.topic(), record.partition(), record.offset(), record.key(), record.value()));
                         updateCallback.run();
+                        if (updateConsumerCallback != null)
+                            updateConsumerCallback.accept(String.format("topic: %s, partition: %s, offset: %s, key: %s, value: %s",
+                                record.topic(), record.partition(), record.offset(), record.key(), record.value()));
                     }
                 }
             }

diff --git a/src/test/java/KafkaWatcherTest.java b/src/test/java/KafkaWatcherTest.java
@@ -47,4 +47,15 @@ public void testUpdate() throws InterruptedException {
         kafkaWatcher.update();
         countDownLatch.await(1, TimeUnit.MINUTES);
     }
+
+    @Test
+    public void testConsumerCallback() throws InterruptedException {
+        CountDownLatch countDownLatch = new CountDownLatch(1);
+        kafkaWatcher.setUpdateCallback((s)-> {
+            countDownLatch.countDown();
+            System.out.println("[callback]" + s);
+        });
+        kafkaWatcher.update();
+        countDownLatch.await(1, TimeUnit.MINUTES);
+    }
 }