Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAREDIS-1119 - Polishing.

Browse Source 
Add getAndMap(…) method to simplify post-retrieval mapping of non-null values. Require key on XInfoCommand command construction.

Tweak javadoc. Reformat code.

Original pull request: #519.
This commit is contained in:
Mark Paluch
2020-04-20 11:53:56 +02:00
parent 3f5533ea9a
commit 254d60e820
4 changed files with 52 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java
View File

@@ -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<XInfoStream> 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<XInfoGroup> 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<XInfoConsumer> 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);
}

25
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java
View File

@@ -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<PendingRecordsCommand, PendingMessagesSummary>(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<PendingRecordsCommand, PendingMessages>(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<CommandResponse<XInfoCommand, XInfoStream>> xInfo(Publisher<XInfoCommand> 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<CommandResponse<XInfoCommand, Flux<XInfoGroup>>> xInfoGroups(Publisher<XInfoCommand> 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<Object>) it)));
return new CommandResponse<>(command,
cmd.xinfoGroups(command.getKey()).map(it -> XInfoGroup.fromList((List<Object>) it)));
}));
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.ReactiveStreamCommands#xInfoConsumers(org.reactivestreams.Publisher)
*/
@Override
public Flux<CommandResponse<XInfoCommand, Flux<XInfoConsumer>>> xInfoConsumers(Publisher<XInfoCommand> 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<Object>) it)));
return new CommandResponse<>(command, cmd.xinfoConsumers(command.getKey(), groupName)
.map(it -> new XInfoConsumer(command.getGroupName(), (List<Object>) it)));
}));
}

55
src/main/java/org/springframework/data/redis/connection/stream/StreamInfo.java
View File

@@ -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> T get(String entry, Class<T> type) {
return type.cast(raw.get(entry));
@Nullable
<T> T get(String entry, Class<T> type) {
Object value = raw.get(entry);
return value == null ? null : type.cast(value);
}
@Nullable
<I, T> T getAndMap(String entry, Class<I> type, Function<I, T> f) {
I value = get(entry, type);
return value == null ? null : f.apply(value);
}
public Map<String, Object> 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<Object, Object> getFirstEntry() {
Map<Object, Object> 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<Object, Object> getLastEntry() {
Map<Object, Object> 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<Object> raw) {
super(raw, DEFAULT_TYPE_HINTS);
}
public static XInfoGroup fromList(List<Object> raw) {
return new XInfoGroup(raw);
}