feat: update to support the new Watcher interface

Author: seriouszyx

pom.xml

@@ -21,10 +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.casbin</groupId>
             <artifactId>jcasbin</artifactId>
-            <version>1.6.3</version>
+            <version>1.13.3</version>
         </dependency>
 
         <dependency>

src/main/java/org/casbin/watcher/RedisWatcher.java

@@ -6,6 +6,7 @@
 import redis.clients.jedis.JedisPoolConfig;
 
 import java.util.UUID;
+import java.util.function.Consumer;
 
 public class RedisWatcher implements Watcher {
     private Runnable updateCallback;
@@ -27,6 +28,11 @@ public void setUpdateCallback(Runnable runnable) {
         subThread.setUpdateCallback(runnable);
     }
 
+    @Override
+    public void setUpdateCallback(Consumer<String> consumer) {
+        subThread.setUpdateCallback(consumer);
+    }
+
     @Override
     public void update() {
         try (Jedis jedis = jedisPool.getResource()) {

src/main/java/org/casbin/watcher/SubThread.java

@@ -3,6 +3,8 @@
 import redis.clients.jedis.Jedis;
 import redis.clients.jedis.JedisPool;
 
+import java.util.function.Consumer;
+
 public class SubThread extends Thread{
     private final JedisPool jedisPool;
     private final Subscriber subscriber;
@@ -18,6 +20,10 @@ public void setUpdateCallback(Runnable runnable){
         subscriber.setUpdateCallback(runnable);
     }
 
+    public void setUpdateCallback(Consumer<String> consumer) {
+        subscriber.setUpdateCallback(consumer);
+    }
+
     @Override
     public void run() {
         try (Jedis jedis = jedisPool.getResource()) {

src/main/java/org/casbin/watcher/Subscriber.java

@@ -2,8 +2,11 @@
 
 import redis.clients.jedis.JedisPubSub;
 
+import java.util.function.Consumer;
+
 public class Subscriber extends JedisPubSub {
     private Runnable runnable;
+    private Consumer<String> consumer;
 
     public Subscriber(Runnable updateCallback) {
         this.runnable = updateCallback;
@@ -13,7 +16,13 @@ public void setUpdateCallback(Runnable runnable){
         this.runnable = runnable;
     }
 
+    public void setUpdateCallback(Consumer<String> consumer) {
+        this.consumer = consumer;
+    }
+
     public void onMessage(String channel, String message) {
         runnable.run();
+        if (consumer != null)
+            consumer.accept("Channel: " + channel + " Message: " + message);
     }
 }

src/test/java/org/casbin/test/RedisWatcherTest.java

@@ -12,7 +12,6 @@ public class RedisWatcherTest {
     private RedisWatcher redisWatcher;
     private final String expect="update msg";
 
-
     @Before
     public void initWatcher(){
         String redisTopic = "jcasbin-topic";
@@ -30,4 +29,13 @@ public void testUpdate() throws InterruptedException {
         Thread.sleep(100);
         Assert.assertEquals(expect, expect);
     }
+
+    @Test
+    public void testConsumerCallback() throws InterruptedException {
+        redisWatcher.setUpdateCallback((s) -> {
+            System.out.print(s);
+        });
+        redisWatcher.update();
+        Thread.sleep(100);
+    }
 }