Fix pubsub when cache enabled (redis#4086)

* fix pubsub when cache enabled

* polish

Author: atakavci

src/main/java/redis/clients/jedis/Protocol.java

@@ -223,34 +223,39 @@ public static Object read(final RedisInputStream is) {
 
   @Experimental
   public static Object read(final RedisInputStream is, final Cache cache) {
-    readPushes(is, cache, false);
-    return process(is);
+    Object unhandledPush = readPushes(is, cache, false);
+    return unhandledPush == null ? process(is) : unhandledPush;
   }
 
   @Experimental
-  public static void readPushes(final RedisInputStream is, final Cache cache, boolean onlyPendingBuffer) {
+  public static Object readPushes(final RedisInputStream is, final Cache cache,
+      boolean onlyPendingBuffer) {
+    Object unhandledPush = null;
     if (onlyPendingBuffer) {
       try {
-        while (is.available() > 0 && is.peek(GREATER_THAN_BYTE)) {
-          is.readByte();
-          processPush(is, cache);
+        while (unhandledPush == null && is.available() > 0 && is.peek(GREATER_THAN_BYTE)) {
+          unhandledPush = processPush(is, cache);
         }
       } catch (IOException e) {
         throw new JedisConnectionException("Failed to read pending buffer for push messages!", e);
       }
     } else {
-      while (is.peek(GREATER_THAN_BYTE)) {
-        is.readByte();
-        processPush(is, cache);
+      while (unhandledPush == null && is.peek(GREATER_THAN_BYTE)) {
+        unhandledPush = processPush(is, cache);
       }
     }
+    return unhandledPush;
   }
 
-  private static void processPush(final RedisInputStream is, Cache cache) {
+  private static Object processPush(final RedisInputStream is, Cache cache) {
+    is.readByte();
     List<Object> list = processMultiBulkReply(is);
     if (list.size() == 2 && list.get(0) instanceof byte[]
         && Arrays.equals(INVALIDATE_BYTES, (byte[]) list.get(0))) {
       cache.deleteByRedisKeys((List) list.get(1));
+      return null;
+    } else {
+      return list;
     }
   }
 

src/test/java/redis/clients/jedis/csc/UnifiedJedisClientSideCacheTestBase.java

@@ -3,16 +3,21 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
 import static org.awaitility.Awaitility.await;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
+import redis.clients.jedis.JedisPubSub;
 import redis.clients.jedis.UnifiedJedis;
 
 public abstract class UnifiedJedisClientSideCacheTestBase {
@@ -222,4 +227,81 @@ public void invalidationOnCacheHitTest() {
     }
   }
 
+  @Test
+  public void simplePubsubWithClientCache() {
+    String test_channel = "test_channel";
+    String test_message = "test message";
+
+    UnifiedJedis publisher = createCachedJedis(CacheConfig.builder().build());
+    Runnable command = () -> publisher.publish(test_channel, test_message + System.currentTimeMillis());
+    ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+    executor.scheduleAtFixedRate(command, 0, 100, java.util.concurrent.TimeUnit.MILLISECONDS);
+
+    List<String> receivedMessages = new ArrayList<>();
+    try (UnifiedJedis subscriber = createCachedJedis(CacheConfig.builder().build())) {
+      JedisPubSub jedisPubSub = new JedisPubSub() {
+        private int count = 0;
+
+        @Override
+        public void onMessage(String channel, String message) {
+          receivedMessages.add(message);
+          if (message.startsWith(test_message) && count++ > 1) {
+            this.unsubscribe(test_channel);
+          }
+        }
+      };
+      subscriber.subscribe(jedisPubSub, test_channel);
+    }
+
+    executor.shutdown();
+    publisher.close();
+
+    assertTrue(receivedMessages.size() > 1);
+    receivedMessages.forEach(message -> assertTrue(message.startsWith(test_message)));
+  }
+
+  @Test
+  public void advancedPubsubWithClientCache() {
+    String test_channel = "test_channel";
+    String test_message = "test message";
+    String test_key = "test_key";
+    String test_value = "test_value";
+
+    UnifiedJedis publisher = createCachedJedis(CacheConfig.builder().build());
+    Runnable command = () -> publisher.publish(test_channel, test_message + System.currentTimeMillis());
+    ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+    executor.scheduleAtFixedRate(command, 0, 50, java.util.concurrent.TimeUnit.MILLISECONDS);
+
+    int iteration = 0;
+    int totalIteration = 10;
+    while (iteration++ < totalIteration) {
+
+      List<String> receivedMessages = new ArrayList<>();
+      try (UnifiedJedis subscriber = createCachedJedis(CacheConfig.builder().build())) {
+
+        subscriber.set(test_key, test_value);
+        assertEquals(test_value, subscriber.get(test_key));
+        JedisPubSub jedisPubSub = new JedisPubSub() {
+          private int count = 0;
+
+          @Override
+          public void onMessage(String channel, String message) {
+            receivedMessages.add(message);
+            if (message.startsWith(test_message) && count++ > 1) {
+              this.unsubscribe(test_channel);
+            }
+          }
+        };
+        subscriber.subscribe(jedisPubSub, test_channel);
+        subscriber.set(test_key, test_value);
+        assertEquals(test_value, subscriber.get(test_key));
+      }
+
+      assertTrue(receivedMessages.size() > 1);
+      receivedMessages.forEach(message -> assertTrue(message.startsWith(test_message)));
+    }
+
+    executor.shutdown();
+    publisher.close();
+  }
 }