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

pom.xml

@@ -570,8 +570,19 @@
 						<include>**/*TestUtil.java</include>
 						<include>**/executors/aggregators/*.java</include>
 						<include>**/*MapMatcher.java</include>
+                        <include>**/Maintenance*.java</include>
+                        <include>**/Push*.java</include>
+                        <include>**/Rebind*.java</include>
+                        <include>src/test/java/redis/clients/jedis/upgrade/*.java</include>
+                        <include>src/test/java/redis/clients/jedis/util/server/*.java</include>
+                        <include>**/TimeoutOptions.java</include>
+                        <include>**/*Handler.java</include>
+                        <include>**/ConnectionTestHelper.java</include>
+						<include>**/*MockTest.java</include>
+						<include>**/server/**/*.java</include>
+						<include>**/pubsub/**/*.java</include>
 					</includes>
-				</configuration>
+                </configuration>
 				<executions>
 					<execution>
 						<goals>

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

@@ -16,6 +16,8 @@
 import java.util.function.Supplier;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import redis.clients.jedis.Protocol.Command;
 import redis.clients.jedis.Protocol.Keyword;
 import redis.clients.jedis.annots.Experimental;
@@ -28,10 +30,13 @@
 import redis.clients.jedis.exceptions.JedisException;
 import redis.clients.jedis.exceptions.JedisValidationException;
 import redis.clients.jedis.util.IOUtils;
+import redis.clients.jedis.util.NumberUtils;
 import redis.clients.jedis.util.RedisInputStream;
 import redis.clients.jedis.util.RedisOutputStream;
+import redis.clients.jedis.util.SafeEncoder;
 
 public class Connection implements Closeable {
+  public static Logger logger = LoggerFactory.getLogger(Connection.class);
 
   public static class Builder {
     private JedisSocketFactory socketFactory;
@@ -72,6 +77,9 @@ public static Builder builder() {
   private Socket socket;
   private RedisOutputStream outputStream;
   private RedisInputStream inputStream;
+  private boolean relaxedTimeoutEnabled = false;
+  private int relaxedTimeout = NumberUtils.safeToInt(TimeoutOptions.DISABLED_TIMEOUT.toMillis());
+  private int relaxedBlockingTimeout = NumberUtils.safeToInt(TimeoutOptions.DISABLED_TIMEOUT.toMillis());
   private int soTimeout = 0;
   private int infiniteSoTimeout = 0;
   private boolean broken = false;
@@ -81,8 +89,12 @@ public static Builder builder() {
   protected String version;
   private AtomicReference<RedisCredentials> currentCredentials = new AtomicReference<>(null);
   private AuthXManager authXManager;
+  private boolean isBlocking = false;
+  private boolean isRelaxed = false;
   private JedisClientConfig clientConfig;
+  private boolean rebindRequested = false;
 
+  private PushConsumerChain pushConsumers;
   public Connection() {
     this(Protocol.DEFAULT_HOST, Protocol.DEFAULT_PORT);
   }
@@ -101,6 +113,9 @@ public Connection(final HostAndPort hostAndPort, final JedisClientConfig clientC
 
   public Connection(final JedisSocketFactory socketFactory) {
     this.socketFactory = socketFactory;
+    this.authXManager = null;
+
+    initPushConsumers(null);
   }
 
   public Connection(final JedisSocketFactory socketFactory, JedisClientConfig clientConfig) {
@@ -114,6 +129,29 @@ protected Connection(Builder builder) {
     this.clientConfig = builder.getClientConfig();
   }
 
+  protected void initPushConsumers(JedisClientConfig config) {
+      /*
+       * Default consumers to process push messages.
+       * Marks all @{link PushMessages as processed, except for pub/sub.
+       * Pub/sub messages are propagated to the client.
+       */
+      this.pushConsumers = PushConsumerChain.of(
+              PushConsumerChain.PUBSUB_ONLY_CONSUMER
+      );
+
+      if (config != null) {
+
+          /*
+           * Add consumer to handle server maintenance events.
+           * Per-connection concerns (timeout relaxation, rebind flag) are handled inline.
+           * Pool-level concerns (factory rebind, pool clear) are notified via memberOf.
+           */
+          if (config.isProactiveRebindEnabled() || relaxedTimeoutEnabled) {
+              addPushConsumer(new MaintenanceEventConsumer(config.isProactiveRebindEnabled()));
+          }
+      }
+  }
+
   @Override
   public String toString() {
     return getClass().getSimpleName() + "{" + socketFactory + "}";
@@ -194,7 +232,8 @@ public void setTimeoutInfinite() {
 
   public void rollbackTimeout() {
     try {
-      socket.setSoTimeout(this.soTimeout);
+      int timeout = getDesiredTimeout();
+      socket.setSoTimeout(timeout);
     } catch (SocketException ex) {
       setBroken();
       throw new JedisConnectionException(ex);
@@ -217,9 +256,11 @@ public <T> T executeCommand(final CommandObject<T> commandObject) {
       return commandObject.getBuilder().build(getOne());
     } else {
       try {
+        isBlocking = true;
         setTimeoutInfinite();
         return commandObject.getBuilder().build(getOne());
       } finally {
+        isBlocking = false;
         rollbackTimeout();
       }
     }
@@ -303,7 +344,7 @@ public void close() {
     if (this.memberOf != null) {
       ConnectionPool pool = this.memberOf;
       this.memberOf = null;
-      if (isBroken()) {
+      if (isBroken() || isRebindRequested()) {
         pool.returnBrokenResource(this);
       } else {
         pool.returnResource(this);
@@ -313,6 +354,10 @@ public void close() {
     }
   }
 
+  private boolean isRebindRequested() {
+    return rebindRequested;
+  }
+
   /**
    * Close the socket and disconnect the server.
    */
@@ -431,8 +476,8 @@ protected void flush() {
   }
 
   @Experimental
-  protected Object protocolRead(RedisInputStream is) {
-    return Protocol.read(is);
+  protected Object protocolRead(RedisInputStream is, PushConsumer handler) {
+    return Protocol.read(is, handler);
   }
 
   @Experimental
@@ -445,7 +490,7 @@ protected Object readProtocolWithCheckingBroken() {
     }
 
     try {
-      return protocolRead(inputStream);
+      return protocolRead(inputStream, pushConsumers);
     } catch (JedisConnectionException exc) {
       broken = true;
       throw exc;
@@ -508,6 +553,13 @@ protected void initializeFromClientConfig(final JedisClientConfig config) {
     try {
       this.soTimeout = config.getSocketTimeoutMillis();
       this.infiniteSoTimeout = config.getBlockingSocketTimeoutMillis();
+      // TODO : ggivo Align configuration properties to other clients
+      this.relaxedTimeout = NumberUtils.safeToInt(config.getTimeoutOptions().getRelaxedTimeout().toMillis());
+      this.relaxedBlockingTimeout = NumberUtils.safeToInt(config.getTimeoutOptions().getRelaxedBlockingTimeout().toMillis());
+      this.relaxedTimeoutEnabled =  TimeoutOptions.isRelaxedTimeoutEnabled(relaxedTimeout) ||
+              TimeoutOptions.isRelaxedTimeoutEnabled(relaxedBlockingTimeout);
+
+      initPushConsumers(config);
 
       connect();
 
@@ -666,4 +718,156 @@ protected boolean isTokenBasedAuthenticationEnabled() {
   protected AuthXManager getAuthXManager() {
     return authXManager;
   }
+
+  /**
+   * Returns an unmodifiable view of the registered push consumers.
+   *
+   * @return
+   */
+  List<PushConsumer> getPushConsumers() {
+    return pushConsumers.getConsumers();
+  }
+
+  @Experimental
+  protected void addPushConsumer(PushConsumer consumer) {
+    this.pushConsumers.add(consumer);
+  }
+
+  @Experimental
+  public boolean isRelaxedTimeoutActive() {
+    return isRelaxed;
+  }
+
+  /**
+   * Calculate the desired timeout based on current state (blocking/non-blocking and relaxed/normal).
+   * When relaxed timeouts are enabled, use configured relaxed timeout if available, otherwise fallback to default timeout.
+   */
+  private int getDesiredTimeout() {
+    if (!isRelaxed) {
+      if (!isBlocking) {
+        return soTimeout;
+      } else {
+        return infiniteSoTimeout;
+      }
+    } else {
+      if (!isBlocking) {
+        return TimeoutOptions.isRelaxedTimeoutEnabled(relaxedTimeout) ? relaxedTimeout : soTimeout;
+      } else {
+        return TimeoutOptions.isRelaxedTimeoutEnabled(relaxedBlockingTimeout) ? relaxedBlockingTimeout : infiniteSoTimeout;
+      }
+    }
+  }
+
+  @Experimental
+  public void relaxTimeouts() {
+    if (!relaxedTimeoutEnabled) {
+      return;
+    }
+
+    if (!isRelaxed) {
+      isRelaxed = true;
+      try {
+        if (isConnected()) {
+          socket.setSoTimeout(getDesiredTimeout());
+        }
+      } catch (SocketException ex) {
+        setBroken();
+        throw new JedisConnectionException(ex);
+      }
+    }
+  }
+
+  @Experimental
+  public void disableRelaxedTimeout() {
+    if (isRelaxed) {
+      isRelaxed = false;
+      try {
+        if (isConnected()) {
+          socket.setSoTimeout(getDesiredTimeout());
+        }
+      } catch (SocketException ex) {
+        setBroken();
+        throw new JedisConnectionException(ex);
+      }
+    }
+  }
+
+  /**
+   * Push consumer that handles server maintenance events.
+   * <p>
+   * Handles per-connection concerns (timeout relaxation, rebind flag) inline.
+   * Notifies the owning {@link ConnectionPool} via {@code memberOf} for pool-level concerns.
+   * </p>
+   */
+  class MaintenanceEventConsumer implements PushConsumer {
+    private final boolean proactiveRebindEnabled;
+
+    public MaintenanceEventConsumer(boolean proactiveRebindEnabled) {
+      this.proactiveRebindEnabled = proactiveRebindEnabled;
+    }
+
+    @Override
+    public void accept(PushConsumerContext context) {
+      PushMessage message = context.getMessage();
+
+      switch (message.getType()) {
+      case "MOVING":
+        onMoving(message);
+        break;
+      case "MIGRATING":
+        relaxTimeouts();
+        break;
+      case "MIGRATED":
+        disableRelaxedTimeout();
+        break;
+      case "FAILING_OVER":
+        relaxTimeouts();
+        break;
+      case "FAILED_OVER":
+        disableRelaxedTimeout();
+        break;
+      }
+    }
+
+    private void onMoving(PushMessage message) {
+      HostAndPort rebindTarget = getRebindTarget(message);
+
+      // per-connection: mark rebind requested and relax timeouts
+      if (proactiveRebindEnabled) {
+        rebindRequested = true;
+      }
+      relaxTimeouts();
+
+      // notify owning pool to rebind factory and clear idle connections
+      if (memberOf != null) {
+        memberOf.onMoving(rebindTarget);
+      }
+    }
+
+    private HostAndPort getRebindTarget(PushMessage message) {
+      // Extract domain/ip and port from the message
+      // MOVING push message format: ["MOVING", slot, "host:port"]
+      List<Object> content = message.getContent();
+
+      if (content.size() < 3) {
+        logger.warn("MOVING push message is malformed: {}", message);
+        return null;
+      }
+
+      Object addressObject = content.get(2); // Get the 3rd element (index 2)
+      if (!(addressObject instanceof byte[])) {
+        logger.warn("Invalid re-bind message format, expected 3rd element to be a byte[], got {}", addressObject);
+        return null;
+      }
+
+
+      try {
+        String addressAndPort = SafeEncoder.encode((byte[]) addressObject);
+        return HostAndPort.from(addressAndPort);
+      } catch (Exception e) {
+        logger.warn("Error parsing re-bind target from message: {}", message, e);
+        return null;
+      }
+    }
+  }
 }

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

@@ -19,7 +19,7 @@
 /**
  * PoolableObjectFactory custom impl.
  */
-public class ConnectionFactory implements PooledObjectFactory<Connection> {
+public class ConnectionFactory implements PooledObjectFactory<Connection> , RebindAware {
 
   public static class Builder {
     private JedisClientConfig clientConfig;
@@ -221,4 +221,22 @@ private void reAuthenticate(Connection jedis) throws Exception {
       throw e;
     }
   }
+
+
+  @Override
+  public void rebind(HostAndPort newHostAndPort) {
+    JedisSocketFactory jedisSocketFactory = connectionBuilder.getSocketFactory();
+    if ((connectionBuilder.getSocketFactory() instanceof RebindAware)) {
+      RebindAware factory = (RebindAware) jedisSocketFactory;
+      logger.debug("Rebinding to {}", newHostAndPort);
+      factory.rebind(newHostAndPort);
+    } else {
+      if ( logger.isDebugEnabled()) {
+        logger.debug("Rebind not supported: {}", jedisSocketFactory.getClass().getName());
+      }
+    }
+
+
+  }
+
 }