diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java index fec9e7b73..e51ffaee5 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java @@ -1041,24 +1041,24 @@ public interface ReactiveStreamCommands { private final @Nullable String groupName; - public XInfoCommand(@Nullable ByteBuffer key, @Nullable String groupName) { + private XInfoCommand(ByteBuffer key, @Nullable String groupName) { super(key); this.groupName = groupName; } - public static XInfoCommand xInfo() { - return new XInfoCommand(null, null); - } + public static XInfoCommand of(ByteBuffer key) { - public XInfoCommand of(ByteBuffer key) { - return new XInfoCommand(key, groupName); + Assert.notNull(key, "Key must not be null"); + + return new XInfoCommand(key, null); } public XInfoCommand consumersIn(String groupName) { return new XInfoCommand(getKey(), groupName); } + @Nullable public String getGroupName() { return groupName; } @@ -1072,7 +1072,7 @@ public interface ReactiveStreamCommands { * @since 2.3 */ default Mono xInfo(ByteBuffer key) { - return xInfo(Mono.just(XInfoCommand.xInfo().of(key))).next().map(CommandResponse::getOutput); + return xInfo(Mono.just(XInfoCommand.of(key))).next().map(CommandResponse::getOutput); } /** @@ -1092,7 +1092,7 @@ public interface ReactiveStreamCommands { * @since 2.3 */ default Flux xInfoGroups(ByteBuffer key) { - return xInfoGroups(Mono.just(XInfoCommand.xInfo().of(key))).next().flatMapMany(CommandResponse::getOutput); + return xInfoGroups(Mono.just(XInfoCommand.of(key))).next().flatMapMany(CommandResponse::getOutput); } /** @@ -1114,7 +1114,7 @@ public interface ReactiveStreamCommands { * @since 2.3 */ default Flux xInfoConsumers(ByteBuffer key, String groupName) { - return xInfoConsumers(Mono.just(XInfoCommand.xInfo().of(key).consumersIn(groupName))).next() + return xInfoConsumers(Mono.just(XInfoCommand.of(key).consumersIn(groupName))).next() .flatMapMany(CommandResponse::getOutput); } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java index c0b19f9df..f5c59dd48 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java @@ -20,7 +20,6 @@ import io.lettuce.core.XClaimArgs; import io.lettuce.core.XReadArgs; import io.lettuce.core.XReadArgs.StreamOffset; import io.lettuce.core.cluster.api.reactive.RedisClusterReactiveCommands; -import org.springframework.data.redis.connection.stream.StreamInfo.XInfoConsumer; import reactor.core.publisher.Flux; import java.nio.ByteBuffer; @@ -40,6 +39,7 @@ import org.springframework.data.redis.connection.stream.Consumer; import org.springframework.data.redis.connection.stream.PendingMessages; import org.springframework.data.redis.connection.stream.PendingMessagesSummary; import org.springframework.data.redis.connection.stream.RecordId; +import org.springframework.data.redis.connection.stream.StreamInfo.XInfoConsumer; import org.springframework.data.redis.connection.stream.StreamInfo.XInfoGroup; import org.springframework.data.redis.connection.stream.StreamInfo.XInfoStream; import org.springframework.data.redis.connection.stream.StreamReadOptions; @@ -256,7 +256,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { // end. return StreamConverters.toPendingMessagesInfo(command.getGroupName(), target); - }).map(value -> new CommandResponse(command, value)); + }).map(value -> new CommandResponse<>(command, value)); })); } @@ -283,7 +283,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { return publisher.collectList().map(it -> { return StreamConverters.toPendingMessages(command.getGroupName(), command.getRange(), it); - }).map(value -> new CommandResponse(command, value)); + }).map(value -> new CommandResponse<>(command, value)); })); } @@ -347,6 +347,10 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { .map(it -> StreamRecords.newRecord().in(it.getStream()).withId(it.getId()).ofBuffer(it.getBody())); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveStreamCommands#xInfo(org.reactivestreams.Publisher) + */ @Override public Flux> xInfo(Publisher commands) { @@ -360,16 +364,26 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveStreamCommands#xInfoGroups(org.reactivestreams.Publisher) + */ @Override public Flux>> xInfoGroups(Publisher commands) { + return connection.execute(cmd -> Flux.from(commands).map(command -> { Assert.notNull(command.getKey(), "Key must not be null!"); - return new CommandResponse(command, cmd.xinfoGroups(command.getKey()).map(it -> XInfoGroup.fromList((List) it))); + return new CommandResponse<>(command, + cmd.xinfoGroups(command.getKey()).map(it -> XInfoGroup.fromList((List) it))); })); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveStreamCommands#xInfoConsumers(org.reactivestreams.Publisher) + */ @Override public Flux>> xInfoConsumers(Publisher commands) { @@ -378,7 +392,8 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { Assert.notNull(command.getKey(), "Key must not be null!"); ByteBuffer groupName = ByteUtils.getByteBuffer(command.getGroupName()); - return new CommandResponse(command, cmd.xinfoConsumers(command.getKey(), groupName).map(it -> new XInfoConsumer(command.getGroupName(), (List) it))); + return new CommandResponse<>(command, cmd.xinfoConsumers(command.getKey(), groupName) + .map(it -> new XInfoConsumer(command.getGroupName(), (List) it))); })); } diff --git a/src/main/java/org/springframework/data/redis/connection/stream/StreamInfo.java b/src/main/java/org/springframework/data/redis/connection/stream/StreamInfo.java index b6ae09a29..b23b7dbfc 100644 --- a/src/main/java/org/springframework/data/redis/connection/stream/StreamInfo.java +++ b/src/main/java/org/springframework/data/redis/connection/stream/StreamInfo.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * https://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, @@ -23,12 +23,15 @@ import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.function.Consumer; +import java.util.function.Function; import java.util.stream.Stream; import org.springframework.data.redis.connection.convert.Converters; +import org.springframework.lang.Nullable; /** * @author Christoph Strobl + * @author Mark Paluch * @since 2.3 */ public class StreamInfo { @@ -56,8 +59,20 @@ public class StreamInfo { this.raw = raw; } - protected T get(String entry, Class type) { - return type.cast(raw.get(entry)); + @Nullable + T get(String entry, Class type) { + + Object value = raw.get(entry); + + return value == null ? null : type.cast(value); + } + + @Nullable + T getAndMap(String entry, Class type, Function f) { + + I value = get(entry, type); + + return value == null ? null : f.apply(value); } public Map getRaw() { @@ -156,13 +171,7 @@ public class StreamInfo { * @return */ public String firstEntryId() { - - Map firstEntryMap = get("first-entry", Map.class); - if (firstEntryMap == null) { - return null; - } - - return firstEntryMap.keySet().iterator().next().toString(); + return getAndMap("first-entry", Map.class, it -> it.keySet().iterator().next().toString()); } /** @@ -171,13 +180,7 @@ public class StreamInfo { * @return */ public Map getFirstEntry() { - - Map lastEntryMap = get("first-entry", Map.class); - if (lastEntryMap == null) { - return null; - } - - return Collections.unmodifiableMap(lastEntryMap); + return getAndMap("first-entry", Map.class, Collections::unmodifiableMap); } /** @@ -186,13 +189,7 @@ public class StreamInfo { * @return */ public String lastEntryId() { - - Map lastEntryMap = get("last-entry", Map.class); - if (lastEntryMap == null) { - return null; - } - - return lastEntryMap.keySet().iterator().next().toString(); + return getAndMap("last-entry", Map.class, it -> it.keySet().iterator().next().toString()); } /** @@ -201,13 +198,7 @@ public class StreamInfo { * @return */ public Map getLastEntry() { - - Map firstEntryMap = get("last-entry", Map.class); - if (firstEntryMap == null) { - return null; - } - - return Collections.unmodifiableMap(firstEntryMap); + return getAndMap("last-entry", Map.class, Collections::unmodifiableMap); } } @@ -315,7 +306,7 @@ public class StreamInfo { private XInfoGroup(List raw) { super(raw, DEFAULT_TYPE_HINTS); } - + public static XInfoGroup fromList(List raw) { return new XInfoGroup(raw); } diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java index 2de72bc0b..0659a322c 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -3339,7 +3339,6 @@ public abstract class AbstractConnectionIntegrationTests { XInfoGroups info = (XInfoGroups) results.get(2); assertThat(info.size()).isZero(); - } @Test // DATAREDIS-1119