Merge a6861ad into f86e1ba

SoilChang · web-flow · commit cbd2ab94c6a3 · 2024-02-19T10:03:15.000Z
diff --git a/src/main/java/redis/clients/jedis/JedisPubSubBase.java b/src/main/java/redis/clients/jedis/JedisPubSubBase.java
@@ -172,7 +172,7 @@ private void process() {
       } else {
         throw new JedisException("Unknown message type: " + reply);
       }
-    } while (isSubscribed());
+    } while (!Thread.currentThread().isInterrupted() && isSubscribed());
 
 //    /* Invalidate instance since this thread is no longer listening */
 //    this.client = null;
diff --git a/src/main/java/redis/clients/jedis/JedisShardedPubSubBase.java b/src/main/java/redis/clients/jedis/JedisShardedPubSubBase.java
@@ -99,7 +99,7 @@ private void process() {
       } else {
         throw new JedisException("Unknown message type: " + reply);
       }
-    } while (isSubscribed());
+    } while (!Thread.currentThread().isInterrupted() && isSubscribed());
 
 //    /* Invalidate instance since this thread is no longer listening */
 //    this.client = null;
diff --git a/src/test/java/redis/clients/jedis/JedisPubSubBaseTest.java b/src/test/java/redis/clients/jedis/JedisPubSubBaseTest.java
@@ -0,0 +1,59 @@
+package redis.clients.jedis;
+
+import junit.framework.TestCase;
+import redis.clients.jedis.util.SafeEncoder;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static redis.clients.jedis.Protocol.ResponseKeyword.MESSAGE;
+import static redis.clients.jedis.Protocol.ResponseKeyword.SUBSCRIBE;
+
+public class JedisPubSubBaseTest extends TestCase {
+
+    public void testProceed_givenThreadInterrupt_exitLoop() throws InterruptedException {
+        // setup
+        final JedisPubSubBase<String> pubSub = new JedisPubSubBase<String>() {
+
+            @Override
+            public void onMessage(String channel, String message) {
+                fail("this should not happen when thread is interrupted");
+            }
+
+            @Override
+            protected String encode(byte[] raw) {
+                return SafeEncoder.encode(raw);
+            }
+        };
+
+        final Connection mockConnection = mock(Connection.class);
+        final List<Object> mockSubscribe = Arrays.asList(
+                SUBSCRIBE.getRaw(), "channel".getBytes(), 1L
+        );
+        final List<Object> mockResponse = Arrays.asList(
+                MESSAGE.getRaw(), "channel".getBytes(), "message".getBytes()
+        );
+
+        when(mockConnection.getUnflushedObject()).
+
+                thenReturn(mockSubscribe, mockResponse);
+
+
+        final CountDownLatch countDownLatch = new CountDownLatch(1);
+        // action
+        final Thread thread = new Thread(() -> {
+            Thread.currentThread().interrupt();
+            pubSub.proceed(mockConnection, "channel");
+
+            countDownLatch.countDown();
+        });
+        thread.start();
+
+        assertTrue(countDownLatch.await(10, TimeUnit.MILLISECONDS));
+
+    }
+}
diff --git a/src/test/java/redis/clients/jedis/JedisShardedPubSubBaseTest.java b/src/test/java/redis/clients/jedis/JedisShardedPubSubBaseTest.java
@@ -0,0 +1,56 @@
+package redis.clients.jedis;
+
+import junit.framework.TestCase;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static redis.clients.jedis.Protocol.ResponseKeyword.SMESSAGE;
+import static redis.clients.jedis.Protocol.ResponseKeyword.SSUBSCRIBE;
+
+public class JedisShardedPubSubBaseTest extends TestCase {
+
+    public void testProceed_givenThreadInterrupt_exitLoop() throws InterruptedException {
+        // setup
+        final JedisShardedPubSubBase<String> pubSub = new JedisShardedPubSubBase<String>() {
+
+            @Override
+            public void onSMessage(String channel, String message) {
+                fail("this should not happen when thread is interrupted");
+            }
+
+            @Override
+            protected String encode(byte[] raw) {
+                return new String(raw);
+            }
+
+        };
+
+        final Connection mockConnection = mock(Connection.class);
+        final List<Object> mockSubscribe = Arrays.asList(
+                SSUBSCRIBE.getRaw(), "channel".getBytes(), 1L
+        );
+        final List<Object> mockResponse = Arrays.asList(
+                SMESSAGE.getRaw(), "channel".getBytes(), "message".getBytes()
+        );
+        when(mockConnection.getUnflushedObject()).thenReturn(mockSubscribe, mockResponse);
+
+
+        final CountDownLatch countDownLatch = new CountDownLatch(1);
+        // action
+        final Thread thread = new Thread(() -> {
+            Thread.currentThread().interrupt();
+            pubSub.proceed(mockConnection, "channel");
+
+            countDownLatch.countDown();
+        });
+        thread.start();
+
+        assertTrue(countDownLatch.await(10, TimeUnit.MILLISECONDS));
+
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -172,7 +172,7 @@ private void process() {`
`172`	`172`	`} else {`
`173`	`173`	`throw new JedisException("Unknown message type: " + reply);`
`174`	`174`	`}`
`175`		`- } while (isSubscribed());`
	`175`	`+ } while (!Thread.currentThread().isInterrupted() && isSubscribed());`
`176`	`176`
`177`	`177`	`// /* Invalidate instance since this thread is no longer listening */`
`178`	`178`	`// this.client = null;`
Original file line number	Diff line number	Diff line change
`@@ -99,7 +99,7 @@ private void process() {`
`99`	`99`	`} else {`
`100`	`100`	`throw new JedisException("Unknown message type: " + reply);`
`101`	`101`	`}`
`102`		`- } while (isSubscribed());`
	`102`	`+ } while (!Thread.currentThread().isInterrupted() && isSubscribed());`
`103`	`103`
`104`	`104`	`// /* Invalidate instance since this thread is no longer listening */`
`105`	`105`	`// this.client = null;`