fix: Override newStatefulRedisClusterConnection method

redis · Jan 25, 2025 · 03edb5d · 03edb5d
1 parent 56f9740
commit 03edb5d
Show file tree

Hide file tree

Showing 4 changed files with 270 additions and 217 deletions.
diff --git a/core/lettucemod/src/main/java/com/redis/lettucemod/StatefulRedisModulesConnectionImpl.java b/core/lettucemod/src/main/java/com/redis/lettucemod/StatefulRedisModulesConnectionImpl.java
@@ -12,77 +12,94 @@
 import io.lettuce.core.StatefulRedisConnectionImpl;
 import io.lettuce.core.cluster.api.sync.RedisClusterCommands;
 import io.lettuce.core.codec.RedisCodec;
+import io.lettuce.core.json.JsonParser;
 import io.lettuce.core.protocol.ConnectionWatchdog;
 import io.lettuce.core.protocol.PushHandler;
+import reactor.core.publisher.Mono;
 
 /**
- * A thread-safe connection to a Redis server. Multiple threads may share
- * one {@link StatefulRedisModulesConnectionImpl}
- * A {@link ConnectionWatchdog} monitors each connection and reconnects
- * automatically until {@link #close} is called. All pending commands will be
- * (re)sent after successful reconnection.
+ * A thread-safe connection to a Redis server. Multiple threads may share one
+ * {@link StatefulRedisModulesConnectionImpl} A {@link ConnectionWatchdog}
+ * monitors each connection and reconnects automatically until {@link #close} is
+ * called. All pending commands will be (re)sent after successful reconnection.
  *
  * @param <K> Key type.
  * @param <V> Value type.
  * @author Mark Paluch
  * @author Julien Ruaux
  */
-public class StatefulRedisModulesConnectionImpl<K, V> extends StatefulRedisConnectionImpl<K, V> implements StatefulRedisModulesConnection<K, V> {
+public class StatefulRedisModulesConnectionImpl<K, V> extends StatefulRedisConnectionImpl<K, V>
+		implements StatefulRedisModulesConnection<K, V> {
 
-    /**
-     * Initialize a new connection.
-     *
-     * @param writer      the channel writer.
-     * @param pushHandler the handler for push notifications.
-     * @param codec       Codec used to encode/decode keys and values.
-     * @param timeout     Maximum time to wait for a response.
-     */
-    public StatefulRedisModulesConnectionImpl(RedisChannelWriter writer, PushHandler pushHandler, RedisCodec<K, V> codec, Duration timeout) {
-        super(writer, pushHandler, codec, timeout);
-    }
+	/**
+	 * Initialize a new connection.
+	 *
+	 * @param writer      the channel writer.
+	 * @param pushHandler the handler for push notifications.
+	 * @param codec       Codec used to encode/decode keys and values.
+	 * @param timeout     Maximum time to wait for a response.
+	 */
+	public StatefulRedisModulesConnectionImpl(RedisChannelWriter writer, PushHandler pushHandler,
+			RedisCodec<K, V> codec, Duration timeout) {
+		super(writer, pushHandler, codec, timeout);
+	}
 
-    /**
-     * Create a new instance of {@link RedisModulesAsyncCommandsImpl}. Can be
-     * overriden to extend.
-     */
-    @Override
-    protected RedisModulesAsyncCommandsImpl<K, V> newRedisAsyncCommandsImpl() {
-        return new RedisModulesAsyncCommandsImpl<>(this, codec);
-    }
+	/**
+	 * Initialize a new connection.
+	 *
+	 * @param writer      the channel writer.
+	 * @param pushHandler the handler for push notifications.
+	 * @param codec       Codec used to encode/decode keys and values.
+	 * @param timeout     Maximum time to wait for a response.
+	 * @param parser      the parser to use for JSON commands.
+	 */
+	public StatefulRedisModulesConnectionImpl(RedisChannelWriter writer, PushHandler pushHandler,
+			RedisCodec<K, V> codec, Duration timeout, Mono<JsonParser> parser) {
+		super(writer, pushHandler, codec, timeout, parser);
+	}
 
-    /**
-     * Create a new instance of {@link RedisModulesReactiveCommandsImpl}. Can be
-     * overriden to extend.
-     */
-    @Override
-    protected RedisReactiveCommandsImpl<K, V> newRedisReactiveCommandsImpl() {
-        return new RedisModulesReactiveCommandsImpl<>(this, codec);
-    }
+	/**
+	 * Create a new instance of {@link RedisModulesAsyncCommandsImpl}. Can be
+	 * overriden to extend.
+	 */
+	@Override
+	protected RedisModulesAsyncCommandsImpl<K, V> newRedisAsyncCommandsImpl() {
+		return new RedisModulesAsyncCommandsImpl<>(this, codec);
+	}
 
-    /**
-     * Create a new instance of {@link RedisModulesCommands}. Can be overriden to
-     * extend.
-     *
-     * @return a new instance
-     */
-    @Override
-    protected RedisModulesCommands<K, V> newRedisSyncCommandsImpl() {
-        return syncHandler(async(), RedisModulesCommands.class, RedisClusterCommands.class);
-    }
+	/**
+	 * Create a new instance of {@link RedisModulesReactiveCommandsImpl}. Can be
+	 * overriden to extend.
+	 */
+	@Override
+	protected RedisReactiveCommandsImpl<K, V> newRedisReactiveCommandsImpl() {
+		return new RedisModulesReactiveCommandsImpl<>(this, codec);
+	}
 
-    @Override
-    public RedisModulesAsyncCommands<K, V> async() {
-        return (RedisModulesAsyncCommands<K, V>) super.async();
-    }
+	/**
+	 * Create a new instance of {@link RedisModulesCommands}. Can be overriden to
+	 * extend.
+	 *
+	 * @return a new instance
+	 */
+	@Override
+	protected RedisModulesCommands<K, V> newRedisSyncCommandsImpl() {
+		return syncHandler(async(), RedisModulesCommands.class, RedisClusterCommands.class);
+	}
 
-    @Override
-    public RedisModulesCommands<K, V> sync() {
-        return (RedisModulesCommands<K, V>) super.sync();
-    }
+	@Override
+	public RedisModulesAsyncCommands<K, V> async() {
+		return (RedisModulesAsyncCommands<K, V>) super.async();
+	}
 
-    @Override
-    public RedisModulesReactiveCommands<K, V> reactive() {
-        return (RedisModulesReactiveCommands<K, V>) super.reactive();
-    }
+	@Override
+	public RedisModulesCommands<K, V> sync() {
+		return (RedisModulesCommands<K, V>) super.sync();
+	}
+
+	@Override
+	public RedisModulesReactiveCommands<K, V> reactive() {
+		return (RedisModulesReactiveCommands<K, V>) super.reactive();
+	}
 
 }