Deprecate reactive StreamChannel methods #1434

Reactive methods accepting a StreamChannel argument are now deprecated in favor of using the variant returning Flux<T> to consume large results using streaming. All methods that are likely to return large results (e.g. hgetall) are stream-enabled and there's no need to use a stream bridge that would work like an aggregator or collector. Using Publisher is idiomatic in reactive programming arrangements. Async and sync methods accepting StreamChannel remain untouched and can be used for result streaming if the application follows asynchronous or imperative programming models.
redis · Sep 29, 2020 · 8a4a321 · 8a4a321
1 parent 5fc9d3f
commit 8a4a321
Show file tree

Hide file tree

Showing 9 changed files with 284 additions and 42 deletions.
diff --git a/RELEASE-NOTES.md b/RELEASE-NOTES.md
@@ -206,6 +206,7 @@ With this release, we took the opportunity to introduce a series of changes that
 * Internal: Removal of the internal `LettuceCharsets` utility class.
 * Internal: Reduced visibility of several `protected` fields in `AbstractRedisClient` (`eventLoopGroups`, `genericWorkerPool`, `timer`, `clientResources`, `clientOptions`, `defaultTimeout`).
 * Internal: Consolidation of Future synchronization utilities (`LettuceFutures`, `RefreshFutures`, `Futures`).
+* Deprecate reactive StreamChannel methods #1434
 
 
 Commands

diff --git a/src/main/java/io/lettuce/core/api/reactive/RedisHashReactiveCommands.java b/src/main/java/io/lettuce/core/api/reactive/RedisHashReactiveCommands.java
@@ -102,7 +102,10 @@ public interface RedisHashReactiveCommands<K, V> {
      * @param channel the channel.
      * @param key the key.
      * @return Long count of the keys.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #hgetall}.
      */
+    @Deprecated
     Mono<Long> hgetall(KeyValueStreamingChannel<K, V> channel, K key);
 
     /**
@@ -119,7 +122,10 @@ public interface RedisHashReactiveCommands<K, V> {
      * @param channel the channel.
      * @param key the key.
      * @return Long count of the keys.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #hkeys}.
      */
+    @Deprecated
     Mono<Long> hkeys(KeyStreamingChannel<K> channel, K key);
 
     /**
@@ -146,7 +152,10 @@ public interface RedisHashReactiveCommands<K, V> {
      * @param key the key.
      * @param fields the fields.
      * @return Long count of the keys.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #hmget}.
      */
+    @Deprecated
     Mono<Long> hmget(KeyValueStreamingChannel<K, V> channel, K key, K... fields);
 
     /**
@@ -200,7 +209,10 @@ public interface RedisHashReactiveCommands<K, V> {
      * @param channel streaming channel that receives a call for every key-value pair.
      * @param key the key.
      * @return StreamScanCursor scan cursor.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #hscan}.
      */
+    @Deprecated
     Mono<StreamScanCursor> hscan(KeyValueStreamingChannel<K, V> channel, K key);
 
     /**
@@ -210,7 +222,10 @@ public interface RedisHashReactiveCommands<K, V> {
      * @param key the key.
      * @param scanArgs scan arguments.
      * @return StreamScanCursor scan cursor.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #hscan}.
      */
+    @Deprecated
     Mono<StreamScanCursor> hscan(KeyValueStreamingChannel<K, V> channel, K key, ScanArgs scanArgs);
 
     /**
@@ -221,7 +236,10 @@ public interface RedisHashReactiveCommands<K, V> {
      * @param scanCursor cursor to resume from a previous scan, must not be {@code null}.
      * @param scanArgs scan arguments.
      * @return StreamScanCursor scan cursor.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #hscan}.
      */
+    @Deprecated
     Mono<StreamScanCursor> hscan(KeyValueStreamingChannel<K, V> channel, K key, ScanCursor scanCursor, ScanArgs scanArgs);
 
     /**
@@ -231,7 +249,10 @@ public interface RedisHashReactiveCommands<K, V> {
      * @param key the key.
      * @param scanCursor cursor to resume from a previous scan, must not be {@code null}.
      * @return StreamScanCursor scan cursor.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #hscan}.
      */
+    @Deprecated
     Mono<StreamScanCursor> hscan(KeyValueStreamingChannel<K, V> channel, K key, ScanCursor scanCursor);
 
     /**
@@ -294,6 +315,9 @@ public interface RedisHashReactiveCommands<K, V> {
      * @param channel streaming channel that receives a call for every value.
      * @param key the key.
      * @return Long count of the keys.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #hvals}.
      */
+    @Deprecated
     Mono<Long> hvals(ValueStreamingChannel<V> channel, K key);
 }
diff --git a/src/main/java/io/lettuce/core/api/reactive/RedisKeyReactiveCommands.java b/src/main/java/io/lettuce/core/api/reactive/RedisKeyReactiveCommands.java
@@ -115,7 +115,10 @@ public interface RedisKeyReactiveCommands<K, V> {
      * @param channel the channel.
      * @param pattern the pattern.
      * @return Long array-reply list of keys matching {@code pattern}.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #keys}.
      */
+    @Deprecated
     Mono<Long> keys(KeyStreamingChannel<K> channel, K pattern);
 
     /**
@@ -293,7 +296,10 @@ public interface RedisKeyReactiveCommands<K, V> {
      * @param channel streaming channel that receives a call for every value.
      * @param key the key.
      * @return Long number of values.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #sort}.
      */
+    @Deprecated
     Mono<Long> sort(ValueStreamingChannel<V> channel, K key);
 
     /**
@@ -312,7 +318,10 @@ public interface RedisKeyReactiveCommands<K, V> {
      * @param key the key.
      * @param sortArgs sort arguments.
      * @return Long number of values.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #sort}.
      */
+    @Deprecated
     Mono<Long> sort(ValueStreamingChannel<V> channel, K key, SortArgs sortArgs);
 
     /**
@@ -386,7 +395,10 @@ public interface RedisKeyReactiveCommands<K, V> {
      *
      * @param channel streaming channel that receives a call for every key.
      * @return StreamScanCursor scan cursor.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #scan}.
      */
+    @Deprecated
     Mono<StreamScanCursor> scan(KeyStreamingChannel<K> channel);
 
     /**
@@ -395,7 +407,10 @@ public interface RedisKeyReactiveCommands<K, V> {
      * @param channel streaming channel that receives a call for every key.
      * @param scanArgs scan arguments.
      * @return StreamScanCursor scan cursor.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #scan}.
      */
+    @Deprecated
     Mono<StreamScanCursor> scan(KeyStreamingChannel<K> channel, ScanArgs scanArgs);
 
     /**
@@ -405,7 +420,10 @@ public interface RedisKeyReactiveCommands<K, V> {
      * @param scanCursor cursor to resume from a previous scan, must not be {@code null}.
      * @param scanArgs scan arguments.
      * @return StreamScanCursor scan cursor.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #scan}.
      */
+    @Deprecated
     Mono<StreamScanCursor> scan(KeyStreamingChannel<K> channel, ScanCursor scanCursor, ScanArgs scanArgs);
 
     /**
@@ -414,6 +432,9 @@ public interface RedisKeyReactiveCommands<K, V> {
      * @param channel streaming channel that receives a call for every key.
      * @param scanCursor cursor to resume from a previous scan, must not be {@code null}.
      * @return StreamScanCursor scan cursor.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #scan}.
      */
+    @Deprecated
     Mono<StreamScanCursor> scan(KeyStreamingChannel<K> channel, ScanCursor scanCursor);
 }
diff --git a/src/main/java/io/lettuce/core/api/reactive/RedisListReactiveCommands.java b/src/main/java/io/lettuce/core/api/reactive/RedisListReactiveCommands.java
@@ -200,7 +200,10 @@ public interface RedisListReactiveCommands<K, V> {
      * @param start the start type: long.
      * @param stop the stop type: long.
      * @return Long count of elements in the specified range.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #lrange}.
      */
+    @Deprecated
     Mono<Long> lrange(ValueStreamingChannel<V> channel, K key, long start, long stop);
 
     /**

diff --git a/src/main/java/io/lettuce/core/api/reactive/RedisSetReactiveCommands.java b/src/main/java/io/lettuce/core/api/reactive/RedisSetReactiveCommands.java
@@ -67,7 +67,10 @@ public interface RedisSetReactiveCommands<K, V> {
      * @param channel the channel.
      * @param keys the keys.
      * @return Long count of members of the resulting set.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #sdiff}.
      */
+    @Deprecated
     Mono<Long> sdiff(ValueStreamingChannel<V> channel, K... keys);
 
     /**
@@ -93,7 +96,10 @@ public interface RedisSetReactiveCommands<K, V> {
      * @param channel the channel.
      * @param keys the keys.
      * @return Long count of members of the resulting set.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #sinter}.
      */
+    @Deprecated
     Mono<Long> sinter(ValueStreamingChannel<V> channel, K... keys);
 
     /**
@@ -144,7 +150,10 @@ public interface RedisSetReactiveCommands<K, V> {
      * @param channel the channel.
      * @param key the keys.
      * @return Long count of members of the resulting set.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #smembers}.
      */
+    @Deprecated
     Mono<Long> smembers(ValueStreamingChannel<V> channel, K key);
 
     /**
@@ -190,7 +199,10 @@ public interface RedisSetReactiveCommands<K, V> {
      * @param key the key.
      * @param count the count.
      * @return Long count of members of the resulting set.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #srandmember}.
      */
+    @Deprecated
     Mono<Long> srandmember(ValueStreamingChannel<V> channel, K key, long count);
 
     /**
@@ -216,7 +228,10 @@ public interface RedisSetReactiveCommands<K, V> {
      * @param channel streaming channel that receives a call for every value.
      * @param keys the keys.
      * @return Long count of members of the resulting set.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #sunion}.
      */
+    @Deprecated
     Mono<Long> sunion(ValueStreamingChannel<V> channel, K... keys);
 
     /**
@@ -270,7 +285,10 @@ public interface RedisSetReactiveCommands<K, V> {
      * @param channel streaming channel that receives a call for every value.
      * @param key the key.
      * @return StreamScanCursor scan cursor.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #sscan}.
      */
+    @Deprecated
     Mono<StreamScanCursor> sscan(ValueStreamingChannel<V> channel, K key);
 
     /**
@@ -280,7 +298,10 @@ public interface RedisSetReactiveCommands<K, V> {
      * @param key the key.
      * @param scanArgs scan arguments.
      * @return StreamScanCursor scan cursor.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #sscan}.
      */
+    @Deprecated
     Mono<StreamScanCursor> sscan(ValueStreamingChannel<V> channel, K key, ScanArgs scanArgs);
 
     /**
@@ -291,7 +312,10 @@ public interface RedisSetReactiveCommands<K, V> {
      * @param scanCursor cursor to resume from a previous scan, must not be {@code null}.
      * @param scanArgs scan arguments.
      * @return StreamScanCursor scan cursor.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #sscan}.
      */
+    @Deprecated
     Mono<StreamScanCursor> sscan(ValueStreamingChannel<V> channel, K key, ScanCursor scanCursor, ScanArgs scanArgs);
 
     /**
@@ -301,6 +325,9 @@ public interface RedisSetReactiveCommands<K, V> {
      * @param key the key.
      * @param scanCursor cursor to resume from a previous scan, must not be {@code null}.
      * @return StreamScanCursor scan cursor.
+     * @deprecated since 6.0 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #sscan}.
      */
+    @Deprecated
     Mono<StreamScanCursor> sscan(ValueStreamingChannel<V> channel, K key, ScanCursor scanCursor);
 }