Topic/ggivo/cae 2680 redis server stub for push notifications

pom.xml

@@ -579,6 +579,8 @@
                         <include>**/*Handler.java</include>
                         <include>**/ConnectionTestHelper.java</include>
 						<include>**/*MockTest.java</include>
+						<include>**/server/**/*.java</include>
+						<include>**/pubsub/**/*.java</include>
 					</includes>
                 </configuration>
 				<executions>

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

@@ -11,7 +11,6 @@
 import java.nio.CharBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Supplier;
@@ -22,7 +21,6 @@
 import redis.clients.jedis.Protocol.Command;
 import redis.clients.jedis.Protocol.Keyword;
 import redis.clients.jedis.annots.Experimental;
-import redis.clients.jedis.annots.VisibleForTesting;
 import redis.clients.jedis.args.ClientAttributeOption;
 import redis.clients.jedis.args.Rawable;
 import redis.clients.jedis.authentication.AuthXManager;

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

@@ -3,8 +3,6 @@
 import org.apache.commons.pool2.PooledObjectFactory;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import redis.clients.authentication.core.Token;
 import redis.clients.jedis.annots.Experimental;
 import redis.clients.jedis.authentication.AuthXManager;

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

@@ -10637,4 +10637,5 @@ public boolean vsetattr(byte[] key, byte[] element, byte[] attributes) {
     checkIsInMultiOrPipeline();
     return connection.executeCommand(commandObjects.vsetattr(key, element, attributes));
   }
+
 }

src/test/java/redis/clients/jedis/ConnectionMockTest.java

@@ -7,24 +7,24 @@
 
 import java.io.IOException;
 import java.time.Duration;
-import java.util.List;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
-import redis.clients.jedis.util.server.TcpMockServer;
+import redis.server.stub.RedisServerStub;
+import redis.server.stub.RedisServerStubConfig;
 
 /**
  * Unit tests for Connection that don't require a real Redis server. Uses TcpMockServer to simulate
  * Redis protocol.
  */
 public class ConnectionMockTest {
 
-  private TcpMockServer mockServer;
+  private RedisServerStub mockServer;
 
   @BeforeEach
   public void setUp() throws IOException {
-    mockServer = new TcpMockServer();
+    mockServer = new RedisServerStub(RedisServerStubConfig.builder().build());
     mockServer.start();
   }
 

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

@@ -6,7 +6,6 @@
 import static org.hamcrest.Matchers.is;
 
 import java.io.IOException;
-import java.time.Duration;
 import java.util.List;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -18,7 +17,8 @@
 import redis.clients.jedis.HostAndPort;
 import redis.clients.jedis.PushConsumer;
 import redis.clients.jedis.PushConsumerChain;
-import redis.clients.jedis.util.server.TcpMockServer;
+import redis.server.stub.RedisServerStub;
+import redis.server.stub.RedisServerStubConfig;
 
 /**
  * Unit tests for CacheConnection that don't require a real Redis server. Uses TcpMockServer to
@@ -30,12 +30,12 @@
  */
 public class CacheConnectionMockTest {
 
-  private TcpMockServer mockServer;
+  private RedisServerStub mockServer;
   private Cache cache;
 
   @BeforeEach
   public void setUp() throws IOException {
-    mockServer = new TcpMockServer();
+    mockServer = new RedisServerStub(RedisServerStubConfig.builder().redisVersion("7.4.0").build());
     mockServer.start();
     cache = CacheFactory.getCache(CacheConfig.builder().maxSize(1000).build());
   }

src/test/java/redis/clients/jedis/pubsub/RedisClientPubSubIT.java

@@ -0,0 +1,57 @@
+package redis.clients.jedis.pubsub;
+
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Tag;
+import redis.clients.jedis.EndpointConfig;
+import redis.clients.jedis.Endpoints;
+import redis.clients.jedis.RedisClient;
+import redis.clients.jedis.RedisProtocol;
+
+/**
+ * Integration tests for RedisClient pub/sub functionality.
+ * <p>
+ * These tests run against a real Redis server. Tests are organized in nested classes for RESP2 and
+ * RESP3 protocols.
+ * </p>
+ */
+@Tag("integration")
+public class RedisClientPubSubIT {
+
+  private static final EndpointConfig endpoint = Endpoints.getRedisEndpoint("standalone0");
+
+  /**
+   * RESP2 protocol tests.
+   */
+  @Nested
+  public class Resp2Tests extends RedisClientPubSubTestBase {
+
+    @Override
+    protected RedisClient createClient(RedisProtocol protocol) {
+      return RedisClient.builder().hostAndPort(endpoint.getHostAndPort())
+          .clientConfig(endpoint.getClientConfigBuilder().protocol(protocol).build()).build();
+    }
+
+    @Override
+    protected RedisProtocol getProtocol() {
+      return RedisProtocol.RESP2;
+    }
+  }
+
+  /**
+   * RESP3 protocol tests.
+   */
+  @Nested
+  public class Resp3Tests extends RedisClientPubSubTestBase {
+
+    @Override
+    protected RedisClient createClient(RedisProtocol protocol) {
+      return RedisClient.builder().hostAndPort(endpoint.getHostAndPort())
+          .clientConfig(endpoint.getClientConfigBuilder().protocol(protocol).build()).build();
+    }
+
+    @Override
+    protected RedisProtocol getProtocol() {
+      return RedisProtocol.RESP3;
+    }
+  }
+}

src/test/java/redis/clients/jedis/pubsub/RedisClientPubSubMockTest.java

@@ -0,0 +1,97 @@
+package redis.clients.jedis.pubsub;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import redis.clients.jedis.DefaultJedisClientConfig;
+import redis.clients.jedis.HostAndPort;
+import redis.clients.jedis.RedisClient;
+import redis.clients.jedis.RedisProtocol;
+import redis.clients.jedis.pubsub.util.PubSubTestHelper;
+import redis.server.stub.MaintenanceEvent;
+import redis.server.stub.RedisServerStub;
+import redis.server.stub.RedisServerStubConfig;
+
+/**
+ * Mock tests for RedisClient pub/sub functionality using RedisServerStub.
+ * <p>
+ * These tests run against RedisServerStub (mock Redis) **ONLY with RESP3** since RedisServerStub
+ * only supports RESP3 protocol.
+ * </p>
+ * <p>
+ * RedisServerStub allows testing scenarios that are difficult to simulate with a real Redis server,
+ * such as arbitrary push notifications sent during subscribed connections, connection failures, and
+ * timing-sensitive behaviors.
+ * </p>
+ */
+public class RedisClientPubSubMockTest {
+
+  private static RedisServerStub serverStub;
+
+  @BeforeAll
+  public static void setUpAll() throws Exception {
+    serverStub = new RedisServerStub(RedisServerStubConfig.builder().build());
+    serverStub.start();
+  }
+
+  @AfterAll
+  public static void tearDownAll() throws Exception {
+    if (serverStub != null) {
+      serverStub.stop();
+    }
+  }
+
+  /**
+   * RESP3 tests - RedisServerStub only supports RESP3.
+   */
+  @Nested
+  public class Resp3Tests extends RedisClientPubSubTestBase {
+
+    @Override
+    protected RedisClient createClient(RedisProtocol protocol) {
+      return RedisClient.builder().hostAndPort(new HostAndPort("localhost", serverStub.getPort()))
+          .clientConfig(DefaultJedisClientConfig.builder().protocol(protocol).build()).build();
+    }
+
+    @Override
+    protected RedisProtocol getProtocol() {
+      return RedisProtocol.RESP3;
+    }
+
+    /**
+     * Test that maintenance notifications during subscription don't cause errors.
+     */
+    @Test
+    public void maintenanceNotificationDuringSubscription() throws Exception {
+      PubSubTestHelper.MessageCapture subscriber = new PubSubTestHelper.MessageCapture(1, 1);
+
+      Thread subscriberThread = new Thread(() -> {
+        client.subscribe(subscriber, "test-channel");
+      });
+      subscriberThread.start();
+
+      assertTrue(subscriber.awaitSubscription(), "Should subscribe successfully");
+
+      // Send server maintenance notification (migrating event for shards)
+      serverStub.sendPushMessageToAll(MaintenanceEvent.migrating(1, 30, "shard-001", "shard-002"));
+
+      // Verify subscription still works after maintenance notification
+      long numReceivers = publisherClient.publish("test-channel", "test-message");
+      assertTrue(numReceivers >= 1, "Should have at least 1 receiver");
+
+      assertTrue(subscriber.awaitMessages(), "Should receive pub/sub message");
+
+      List<String> messages = subscriber.getMessages();
+      assertEquals(1, messages.size(), "Should receive 1 message");
+      assertEquals("test-message", messages.get(0), "Should receive correct message");
+
+      subscriber.unsubscribe();
+      subscriberThread.join(THREAD_JOIN_TIMEOUT_MS);
+    }
+  }
+}