diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveGeoCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveGeoCommands.java index 0c325160d..b2510c8b7 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveGeoCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveGeoCommands.java @@ -417,7 +417,7 @@ public interface ReactiveGeoCommands { Assert.notNull(member, "Member must not be null!"); return geoHash(key, Collections.singletonList(member)) // - .then(vals -> vals.isEmpty() ? Mono.empty() : Mono.justOrEmpty(vals.iterator().next())); + .flatMap(vals -> vals.isEmpty() ? Mono.empty() : Mono.justOrEmpty(vals.iterator().next())); } /** @@ -524,7 +524,7 @@ public interface ReactiveGeoCommands { Assert.notNull(member, "Member must not be null!"); return geoPos(key, Collections.singletonList(member)) - .then(vals -> vals.isEmpty() ? Mono.empty() : Mono.justOrEmpty(vals.iterator().next())); + .flatMap(vals -> vals.isEmpty() ? Mono.empty() : Mono.justOrEmpty(vals.iterator().next())); } /** diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterKeyCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterKeyCommands.java index e37b550e8..8ba5cda69 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterKeyCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterKeyCommands.java @@ -88,7 +88,7 @@ public class LettuceReactiveClusterKeyCommands extends LettuceReactiveKeyCommand return super.rename(Mono.just(command)); } - Flux result = cmd.dump(command.getKey()) + Mono result = cmd.dump(command.getKey()) .otherwiseIfEmpty(Mono.error(new RedisSystemException("Cannot rename key that does not exist", new RedisException("ERR no such key.")))) .flatMap(value -> cmd.restore(command.getNewName(), 0, value).flatMap(res -> cmd.del(command.getKey()))) @@ -114,7 +114,7 @@ public class LettuceReactiveClusterKeyCommands extends LettuceReactiveKeyCommand return super.renameNX(Mono.just(command)); } - Flux result = cmd.exists(command.getNewName()).flatMap(exists -> { + Mono result = cmd.exists(command.getNewName()).flatMap(exists -> { if (exists == 1) { return Mono.just(Boolean.FALSE); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterListCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterListCommands.java index 228293509..6bec796ef 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterListCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterListCommands.java @@ -78,7 +78,7 @@ public class LettuceReactiveClusterListCommands extends LettuceReactiveListComma return super.rPopLPush(Mono.just(command)); } - Flux result = cmd.rpop(command.getKey()) + Mono result = cmd.rpop(command.getKey()) .flatMap(value -> cmd.lpush(command.getDestination(), value).map(x -> value)); return result.map(value -> new ByteBufferResponse<>(command, value)); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterSetCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterSetCommands.java index c7673b612..a74b78eaa 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterSetCommands.java @@ -224,7 +224,7 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand return super.sMove(Mono.just(command)); } - Flux result = cmd.exists(command.getKey()) + Mono result = cmd.exists(command.getKey()) .flatMap(nrKeys -> nrKeys == 0 ? Mono.empty() : cmd.sismember(command.getKey(), command.getValue())) .flatMap(exists -> { diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveGeoOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveGeoOperations.java index d7d996866..c0123e78f 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveGeoOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveGeoOperations.java @@ -240,7 +240,7 @@ public class DefaultReactiveGeoOperations implements ReactiveGeoOperations Assert.notNull(within, "Circle must not be null!"); return createMono(connection -> connection.geoRadius(rawKey(key), within) // - .flatMap(Flux::fromIterable) // + .flatMapMany(Flux::fromIterable) // .map(location -> new GeoLocation<>(readValue(location.getName()), location.getPoint())) // .collectList()); } @@ -256,7 +256,7 @@ public class DefaultReactiveGeoOperations implements ReactiveGeoOperations Assert.notNull(args, "GeoRadiusCommandArgs must not be null!"); return createMono(connection -> connection.geoRadius(rawKey(key), within, args) // - .flatMap(Flux::fromIterable) // + .flatMapMany(Flux::fromIterable) // .map(geoResult -> new GeoResult<>( new GeoLocation<>(readValue(geoResult.getContent().getName()), geoResult.getContent().getPoint()), geoResult.getDistance())) // @@ -274,7 +274,7 @@ public class DefaultReactiveGeoOperations implements ReactiveGeoOperations Assert.notNull(member, "Member must not be null!"); return createMono(connection -> connection.geoRadiusByMember(rawKey(key), rawValue(member), new Distance(radius)) // - .flatMap(Flux::fromIterable) // + .flatMapMany(Flux::fromIterable) // .map(geoLocation -> new GeoLocation<>(readValue(geoLocation.getName()), geoLocation.getPoint())) // .collectList()); } @@ -290,7 +290,7 @@ public class DefaultReactiveGeoOperations implements ReactiveGeoOperations Assert.notNull(distance, "Distance must not be null!"); return createMono(connection -> connection.geoRadiusByMember(rawKey(key), rawValue(member), distance) // - .flatMap(Flux::fromIterable) // + .flatMapMany(Flux::fromIterable) // .map(geoLocation -> new GeoLocation<>(readValue(geoLocation.getName()), geoLocation.getPoint())) // .collectList()); } @@ -308,7 +308,7 @@ public class DefaultReactiveGeoOperations implements ReactiveGeoOperations Assert.notNull(args, "GeoRadiusCommandArgs must not be null!"); return createMono(connection -> connection.geoRadiusByMember(rawKey(key), rawValue(member), distance, args) // - .flatMap(Flux::fromIterable) // + .flatMapMany(Flux::fromIterable) // .map(geoResult -> new GeoResult<>( new GeoLocation<>(readValue(geoResult.getContent().getName()), geoResult.getContent().getPoint()), geoResult.getDistance())) // diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveHashOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveHashOperations.java index 05be4200b..66df09dca 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveHashOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveHashOperations.java @@ -74,7 +74,7 @@ public class DefaultReactiveHashOperations implements ReactiveHashOpe return createMono(connection -> Flux.fromArray(hashKeys) // .map(o -> (HK) o).map(this::rawHashKey) // .collectList() // - .then(hks -> connection.hDel(rawKey(key), hks))); + .flatMap(hks -> connection.hDel(rawKey(key), hks))); } /* (non-Javadoc) @@ -116,7 +116,7 @@ public class DefaultReactiveHashOperations implements ReactiveHashOpe return createMono(connection -> Flux.fromIterable(hashKeys) // .map(this::rawHashKey) // .collectList() // - .then(hks -> connection.hMGet(rawKey(key), hks)).map(this::deserializeHashValues)); + .flatMap(hks -> connection.hMGet(rawKey(key), hks)).map(this::deserializeHashValues)); } /* (non-Javadoc) @@ -156,7 +156,7 @@ public class DefaultReactiveHashOperations implements ReactiveHashOpe Assert.notNull(key, "Key must not be null!"); return createMono(connection -> connection.hKeys(rawKey(key)) // - .flatMap(Flux::fromIterable) // + .flatMapMany(Flux::fromIterable) // .map(this::readHashKey) // .collectList()); } @@ -221,7 +221,7 @@ public class DefaultReactiveHashOperations implements ReactiveHashOpe Assert.notNull(key, "Key must not be null!"); return createMono(connection -> connection.hVals(rawKey(key)) // - .flatMap(Flux::fromIterable) // + .flatMapMany(Flux::fromIterable) // .map(this::readHashValue) // .collectList()); } diff --git a/src/test/java/org/springframework/data/redis/core/DefaultReactiveListOperationsIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/DefaultReactiveListOperationsIntegrationTests.java index b2162886a..ab07445ca 100644 --- a/src/test/java/org/springframework/data/redis/core/DefaultReactiveListOperationsIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/core/DefaultReactiveListOperationsIntegrationTests.java @@ -125,7 +125,7 @@ public class DefaultReactiveListOperationsIntegrationTests { StepVerifier.create(listOperations.leftPush(key, value1)).expectNext(1L).verifyComplete(); StepVerifier.create(listOperations.leftPush(key, value2)).expectNext(2L).verifyComplete(); - StepVerifier.create(listOperations.range(key, 0, -1).flatMap(Flux::fromIterable)).expectNext(value2) + StepVerifier.create(listOperations.range(key, 0, -1).flatMapMany(Flux::fromIterable)).expectNext(value2) .expectNext(value1).verifyComplete(); } @@ -140,7 +140,7 @@ public class DefaultReactiveListOperationsIntegrationTests { StepVerifier.create(listOperations.leftPushAll(key, value1, value2)).expectNext(2L).verifyComplete(); - StepVerifier.create(listOperations.range(key, 0, -1).flatMap(Flux::fromIterable)).expectNext(value2) + StepVerifier.create(listOperations.range(key, 0, -1).flatMapMany(Flux::fromIterable)).expectNext(value2) .expectNext(value1).verifyComplete(); } @@ -170,7 +170,7 @@ public class DefaultReactiveListOperationsIntegrationTests { StepVerifier.create(listOperations.leftPush(key, value1, value3)).expectNext(3L).verifyComplete(); - StepVerifier.create(listOperations.range(key, 0, -1).flatMap(Flux::fromIterable)).expectNext(value2) + StepVerifier.create(listOperations.range(key, 0, -1).flatMapMany(Flux::fromIterable)).expectNext(value2) .expectNext(value3).expectNext(value1).verifyComplete(); } @@ -186,7 +186,7 @@ public class DefaultReactiveListOperationsIntegrationTests { StepVerifier.create(listOperations.rightPush(key, value1)).expectNext(1L).verifyComplete(); StepVerifier.create(listOperations.rightPush(key, value2)).expectNext(2L).verifyComplete(); - StepVerifier.create(listOperations.range(key, 0, -1).flatMap(Flux::fromIterable)).expectNext(value1) + StepVerifier.create(listOperations.range(key, 0, -1).flatMapMany(Flux::fromIterable)).expectNext(value1) .expectNext(value2).verifyComplete(); } @@ -201,7 +201,7 @@ public class DefaultReactiveListOperationsIntegrationTests { StepVerifier.create(listOperations.rightPushAll(key, value1, value2)).expectNext(2L).verifyComplete(); - StepVerifier.create(listOperations.range(key, 0, -1).flatMap(Flux::fromIterable)).expectNext(value1) + StepVerifier.create(listOperations.range(key, 0, -1).flatMapMany(Flux::fromIterable)).expectNext(value1) .expectNext(value2).verifyComplete(); } @@ -231,7 +231,7 @@ public class DefaultReactiveListOperationsIntegrationTests { StepVerifier.create(listOperations.rightPush(key, value1, value3)).expectNext(3L).verifyComplete(); - StepVerifier.create(listOperations.range(key, 0, -1).flatMap(Flux::fromIterable)).expectNext(value1) + StepVerifier.create(listOperations.range(key, 0, -1).flatMapMany(Flux::fromIterable)).expectNext(value1) .expectNext(value3).expectNext(value2).verifyComplete(); } @@ -248,7 +248,7 @@ public class DefaultReactiveListOperationsIntegrationTests { StepVerifier.create(listOperations.set(key, 1, value1)).expectNext(true).verifyComplete(); - StepVerifier.create(listOperations.range(key, 0, -1).flatMap(Flux::fromIterable)).expectNext(value1) + StepVerifier.create(listOperations.range(key, 0, -1).flatMapMany(Flux::fromIterable)).expectNext(value1) .expectNext(value1).verifyComplete(); } @@ -265,7 +265,7 @@ public class DefaultReactiveListOperationsIntegrationTests { StepVerifier.create(listOperations.remove(key, 1, value1)).expectNext(1L).verifyComplete(); - StepVerifier.create(listOperations.range(key, 0, -1).flatMap(Flux::fromIterable)).expectNext(value2) + StepVerifier.create(listOperations.range(key, 0, -1).flatMapMany(Flux::fromIterable)).expectNext(value2) .verifyComplete(); } diff --git a/src/test/java/reactor/test/TestSubscriber.java b/src/test/java/reactor/test/TestSubscriber.java index b23b55115..5a3456d59 100644 --- a/src/test/java/reactor/test/TestSubscriber.java +++ b/src/test/java/reactor/test/TestSubscriber.java @@ -37,8 +37,6 @@ import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; import reactor.core.Fuseable; -import reactor.core.Receiver; -import reactor.core.Trackable; import reactor.core.publisher.Operators; /** @@ -95,7 +93,7 @@ import reactor.core.publisher.Operators; * @author Stephane Maldini * @author Brian Clozel */ -public class TestSubscriber implements Subscriber, Subscription, Trackable, Receiver { +public class TestSubscriber implements Subscriber, Subscription { /** * Default timeout for waiting next values to be received @@ -839,17 +837,14 @@ public class TestSubscriber implements Subscriber, Subscription, Trackable } } - @Override public final boolean isCancelled() { return s == Operators.cancelledSubscription(); } - @Override public final boolean isStarted() { return s != null; } - @Override public final boolean isTerminated() { return isCancelled(); } @@ -953,7 +948,6 @@ public class TestSubscriber implements Subscriber, Subscription, Trackable } } - @Override public final long requestedFromDownstream() { return requested; } @@ -969,7 +963,6 @@ public class TestSubscriber implements Subscriber, Subscription, Trackable return this; } - @Override public Subscription upstream() { return s; }