diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/StreamConverters.java b/src/main/java/org/springframework/data/redis/connection/lettuce/StreamConverters.java index 82ca4329a..9e6e9df34 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/StreamConverters.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/StreamConverters.java @@ -26,7 +26,6 @@ import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.function.BiFunction; -import java.util.stream.Collectors; import org.springframework.core.convert.converter.Converter; import org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions; @@ -38,6 +37,7 @@ import org.springframework.data.redis.connection.stream.RecordId; import org.springframework.data.redis.connection.stream.StreamReadOptions; import org.springframework.data.redis.connection.stream.StreamRecords; import org.springframework.data.redis.util.ByteUtils; +import org.springframework.lang.Nullable; import org.springframework.util.NumberUtils; /** @@ -69,7 +69,7 @@ class StreamConverters { return new org.springframework.data.redis.connection.stream.PendingMessage(id, consumer, Duration.ofMillis(it.getMsSinceLastDelivery()), it.getRedeliveryCount()); - }).collect(Collectors.toList()); + }).toList(); return new org.springframework.data.redis.connection.stream.PendingMessages(groupName, messages); @@ -111,18 +111,10 @@ class StreamConverters { return (it) -> StreamRecords.newRecord().in(it.getStream()).withId(it.getId()).ofBytes(it.getBody()); } - static Converter>, List> byteRecordListConverter() { - return new ListConverter<>(byteRecordConverter()); - } - static Converter, RecordId> messageToIdConverter() { return (it) -> RecordId.of(it.getId()); } - static Converter>, List> messagesToIds() { - return MESSAGEs_TO_IDs; - } - /** * Convert the raw Lettuce xpending result to {@link PendingMessages}. * @@ -157,7 +149,7 @@ class StreamConverters { * @param value dont't get me started om this. * @return preconverted values that Lettuce parsers are able to understand \รถ/. */ - private static Object preConvertNativeValues(Object value) { + private static Object preConvertNativeValues(@Nullable Object value) { if (value instanceof ByteBuffer || value instanceof byte[]) { diff --git a/src/main/java/org/springframework/data/redis/connection/stream/ByteBufferRecord.java b/src/main/java/org/springframework/data/redis/connection/stream/ByteBufferRecord.java index 281ec32a0..fe1185d06 100644 --- a/src/main/java/org/springframework/data/redis/connection/stream/ByteBufferRecord.java +++ b/src/main/java/org/springframework/data/redis/connection/stream/ByteBufferRecord.java @@ -17,8 +17,8 @@ package org.springframework.data.redis.connection.stream; import java.nio.ByteBuffer; import java.util.Collections; +import java.util.LinkedHashMap; import java.util.Map; -import java.util.stream.Collectors; import org.springframework.data.redis.hash.HashMapper; import org.springframework.data.redis.serializer.RedisSerializer; @@ -65,16 +65,17 @@ public interface ByteBufferRecord extends MapRecord MapRecord deserialize(@Nullable RedisSerializer streamSerializer, @Nullable RedisSerializer fieldSerializer, @Nullable RedisSerializer valueSerializer) { - return mapEntries(it -> Collections. singletonMap( - fieldSerializer != null ? fieldSerializer.deserialize(ByteUtils.getBytes(it.getKey())) : (HK) it.getKey(), - valueSerializer != null ? valueSerializer.deserialize(ByteUtils.getBytes(it.getValue())) : (HV) it.getValue()) - .entrySet().iterator().next()).withStreamKey( - streamSerializer != null ? streamSerializer.deserialize(ByteUtils.getBytes(getStream())) : (K) getStream()); + return mapEntries(it -> { + + Map map = Collections.singletonMap(StreamSerialization.deserialize(fieldSerializer, it.getKey()), + StreamSerialization.deserialize(valueSerializer, it.getValue())); + + return map.entrySet().iterator().next(); + }).withStreamKey(StreamSerialization.deserialize(streamSerializer, getRequiredStream())); } /** @@ -84,7 +85,7 @@ public interface ByteBufferRecord extends MapRecord source) { - return StreamRecords.newRecord().in(source.getStream()).withId(source.getId()).ofBuffer(source.getValue()); + return StreamRecords.newRecord().in(source.getRequiredStream()).withId(source.getId()).ofBuffer(source.getValue()); } /** @@ -97,10 +98,13 @@ public interface ByteBufferRecord extends MapRecord ObjectRecord toObjectRecord( HashMapper mapper) { - Map targetMap = getValue().entrySet().stream().collect( - Collectors.toMap(entry -> ByteUtils.getBytes(entry.getKey()), entry -> ByteUtils.getBytes(entry.getValue()))); + Map value = getValue(); + Map targetMap = new LinkedHashMap<>(value.size()); + + value.forEach((k, v) -> targetMap.put(ByteUtils.getBytes(k), ByteUtils.getBytes(v))); return Record. of((OV) (mapper).fromHash((Map) targetMap)).withId(getId()) - .withStreamKey(getStream()); + .withStreamKey(getRequiredStream()); } + } diff --git a/src/main/java/org/springframework/data/redis/connection/stream/ByteRecord.java b/src/main/java/org/springframework/data/redis/connection/stream/ByteRecord.java index 18fb7d9cb..f05936324 100644 --- a/src/main/java/org/springframework/data/redis/connection/stream/ByteRecord.java +++ b/src/main/java/org/springframework/data/redis/connection/stream/ByteRecord.java @@ -16,8 +16,8 @@ package org.springframework.data.redis.connection.stream; import java.util.Collections; +import java.util.Map; -import org.springframework.data.redis.connection.RedisStreamCommands; import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.lang.Nullable; @@ -65,11 +65,13 @@ public interface ByteRecord extends MapRecord { @Nullable RedisSerializer fieldSerializer, @Nullable RedisSerializer valueSerializer) { - return mapEntries(it -> Collections - . singletonMap(fieldSerializer != null ? fieldSerializer.deserialize(it.getKey()) : (HK) it.getKey(), - valueSerializer != null ? valueSerializer.deserialize(it.getValue()) : (HV) it.getValue()) - .entrySet().iterator().next()) - .withStreamKey(streamSerializer != null ? streamSerializer.deserialize(getStream()) : (K) getStream()); + return mapEntries(it -> { + + Map map = Collections.singletonMap(StreamSerialization.deserialize(fieldSerializer, it.getKey()), + StreamSerialization.deserialize(valueSerializer, it.getValue())); + + return map.entrySet().iterator().next(); + }).withStreamKey(StreamSerialization.deserialize(streamSerializer, getRequiredStream())); } /** @@ -79,6 +81,6 @@ public interface ByteRecord extends MapRecord { * @return new instance of {@link ByteRecord}. */ static ByteRecord of(MapRecord source) { - return StreamRecords.newRecord().in(source.getStream()).withId(source.getId()).ofBytes(source.getValue()); + return StreamRecords.newRecord().in(source.getRequiredStream()).withId(source.getId()).ofBytes(source.getValue()); } } diff --git a/src/main/java/org/springframework/data/redis/connection/stream/MapRecord.java b/src/main/java/org/springframework/data/redis/connection/stream/MapRecord.java index 9f9d62cf2..4f9513324 100644 --- a/src/main/java/org/springframework/data/redis/connection/stream/MapRecord.java +++ b/src/main/java/org/springframework/data/redis/connection/stream/MapRecord.java @@ -78,7 +78,7 @@ public interface MapRecord extends Record>, Iterable extends Record>, Iterable extends Record>, Iterable ObjectRecord toObjectRecord(HashMapper mapper) { - return Record. of((OV) mapper.fromHash((Map) getValue())).withId(getId()).withStreamKey(getStream()); + return Record. of((OV) mapper.fromHash((Map) getValue())).withId(getId()).withStreamKey(getRequiredStream()); } } diff --git a/src/main/java/org/springframework/data/redis/connection/stream/Record.java b/src/main/java/org/springframework/data/redis/connection/stream/Record.java index a6556707b..931490263 100644 --- a/src/main/java/org/springframework/data/redis/connection/stream/Record.java +++ b/src/main/java/org/springframework/data/redis/connection/stream/Record.java @@ -39,6 +39,24 @@ public interface Record { @Nullable S getStream(); + /** + * The id of the stream (aka the {@literal key} in Redis). + * + * @return can be {@literal null}. + * @throws IllegalStateException if the stream is {@literal null}. + * @since 3.0 + */ + default S getRequiredStream() { + + S stream = getStream(); + + if (stream == null) { + throw new IllegalStateException("Stream is not available"); + } + + return stream; + } + /** * The id of the entry inside the stream. * 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 b181bccbb..57b93a9d7 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 @@ -27,6 +27,7 @@ import java.util.function.Function; import java.util.stream.Stream; import org.springframework.data.redis.connection.convert.Converters; +import org.springframework.data.util.Streamable; import org.springframework.lang.Nullable; /** @@ -67,6 +68,16 @@ public class StreamInfo { return value == null ? null : type.cast(value); } + T getRequired(String entry, Class type) { + + T value = get(entry, type); + + if (value == null) { + throw new IllegalStateException("Value for entry '%s' is null.".formatted(entry)); + } + return value; + } + @Nullable T getAndMap(String entry, Class type, Function f) { @@ -124,8 +135,8 @@ public class StreamInfo { * * @return */ - public Long streamLength() { - return get("length", Long.class); + public long streamLength() { + return getRequired("length", Long.class); } /** @@ -133,8 +144,8 @@ public class StreamInfo { * * @return */ - public Long radixTreeKeySize() { - return get("radix-tree-keys", Long.class); + public long radixTreeKeySize() { + return getRequired("radix-tree-keys", Long.class); } /** @@ -142,8 +153,8 @@ public class StreamInfo { * * @return */ - public Long radixTreeNodesSize() { - return get("radix-tree-nodes", Long.class); + public long radixTreeNodesSize() { + return getRequired("radix-tree-nodes", Long.class); } /** @@ -151,8 +162,8 @@ public class StreamInfo { * * @return */ - public Long groupCount() { - return get("groups", Long.class); + public long groupCount() { + return getRequired("groups", Long.class); } /** @@ -162,7 +173,7 @@ public class StreamInfo { * @return */ public String lastGeneratedId() { - return get("last-generated-id", String.class); + return getRequired("last-generated-id", String.class); } /** @@ -209,7 +220,7 @@ public class StreamInfo { * * @author Christoph Strobl */ - public static class XInfoGroups { + public static class XInfoGroups implements Streamable { private final List groupInfoList; @@ -253,6 +264,7 @@ public class StreamInfo { /** * @return {@literal true} if no groups associated. */ + @Override public boolean isEmpty() { return groupInfoList.isEmpty(); } @@ -262,6 +274,7 @@ public class StreamInfo { * * @return */ + @Override public Iterator iterator() { return groupInfoList.iterator(); } @@ -281,6 +294,7 @@ public class StreamInfo { * * @return */ + @Override public Stream stream() { return groupInfoList.stream(); } @@ -290,6 +304,7 @@ public class StreamInfo { * * @param action */ + @Override public void forEach(Consumer action) { groupInfoList.forEach(action); } @@ -317,7 +332,7 @@ public class StreamInfo { * @return */ public String groupName() { - return get("name", String.class); + return getRequired("name", String.class); } /** @@ -326,7 +341,7 @@ public class StreamInfo { * @return */ public Long consumerCount() { - return get("consumers", Long.class); + return getRequired("consumers", Long.class); } /** @@ -335,7 +350,7 @@ public class StreamInfo { * @return */ public Long pendingCount() { - return get("pending", Long.class); + return getRequired("pending", Long.class); } /** @@ -344,11 +359,11 @@ public class StreamInfo { * @return */ public String lastDeliveredId() { - return get("last-delivered-id", String.class); + return getRequired("last-delivered-id", String.class); } } - public static class XInfoConsumers { + public static class XInfoConsumers implements Streamable { private final List consumerInfoList; @@ -386,6 +401,7 @@ public class StreamInfo { /** * @return {@literal true} if no groups associated. */ + @Override public boolean isEmpty() { return consumerInfoList.isEmpty(); } @@ -395,6 +411,7 @@ public class StreamInfo { * * @return */ + @Override public Iterator iterator() { return consumerInfoList.iterator(); } @@ -414,6 +431,7 @@ public class StreamInfo { * * @return */ + @Override public Stream stream() { return consumerInfoList.stream(); } @@ -423,6 +441,7 @@ public class StreamInfo { * * @param action */ + @Override public void forEach(Consumer action) { consumerInfoList.forEach(action); } @@ -458,7 +477,7 @@ public class StreamInfo { * @return */ public String consumerName() { - return get("name", String.class); + return getRequired("name", String.class); } /** @@ -466,8 +485,8 @@ public class StreamInfo { * * @return */ - public Long idleTimeMs() { - return get("idle", Long.class); + public long idleTimeMs() { + return getRequired("idle", Long.class); } /** @@ -484,8 +503,8 @@ public class StreamInfo { * * @return */ - public Long pendingCount() { - return get("pending", Long.class); + public long pendingCount() { + return getRequired("pending", Long.class); } } } diff --git a/src/main/java/org/springframework/data/redis/connection/stream/StreamRecords.java b/src/main/java/org/springframework/data/redis/connection/stream/StreamRecords.java index 6b6dcf7dd..bf6f8cbcd 100644 --- a/src/main/java/org/springframework/data/redis/connection/stream/StreamRecords.java +++ b/src/main/java/org/springframework/data/redis/connection/stream/StreamRecords.java @@ -232,8 +232,8 @@ public class StreamRecords { */ static class MapBackedRecord implements MapRecord { - private @Nullable S stream; - private RecordId recordId; + private final @Nullable S stream; + private final RecordId recordId; private final Map kvMap; MapBackedRecord(@Nullable S stream, RecordId recordId, Map kvMap) { @@ -249,7 +249,6 @@ public class StreamRecords { return stream; } - @Nullable @Override public RecordId getId() { return recordId; @@ -322,7 +321,7 @@ public class StreamRecords { */ static class ByteMapBackedRecord extends MapBackedRecord implements ByteRecord { - ByteMapBackedRecord(byte[] stream, RecordId recordId, Map map) { + ByteMapBackedRecord(@Nullable byte[] stream, RecordId recordId, Map map) { super(stream, recordId, map); } @@ -343,7 +342,8 @@ public class StreamRecords { static class ByteBufferMapBackedRecord extends MapBackedRecord implements ByteBufferRecord { - ByteBufferMapBackedRecord(ByteBuffer stream, RecordId recordId, Map map) { + ByteBufferMapBackedRecord(@Nullable ByteBuffer stream, @Nullable RecordId recordId, + Map map) { super(stream, recordId, map); } @@ -363,7 +363,7 @@ public class StreamRecords { */ static class StringMapBackedRecord extends MapBackedRecord implements StringRecord { - StringMapBackedRecord(String stream, RecordId recordId, Map stringStringMap) { + StringMapBackedRecord(@Nullable String stream, @Nullable RecordId recordId, Map stringStringMap) { super(stream, recordId, stringStringMap); } diff --git a/src/main/java/org/springframework/data/redis/connection/stream/StreamSerialization.java b/src/main/java/org/springframework/data/redis/connection/stream/StreamSerialization.java index 3f04237d4..938d63d59 100644 --- a/src/main/java/org/springframework/data/redis/connection/stream/StreamSerialization.java +++ b/src/main/java/org/springframework/data/redis/connection/stream/StreamSerialization.java @@ -15,7 +15,10 @@ */ package org.springframework.data.redis.connection.stream; +import java.nio.ByteBuffer; + import org.springframework.data.redis.serializer.RedisSerializer; +import org.springframework.data.redis.util.ByteUtils; import org.springframework.lang.Nullable; /** @@ -34,11 +37,27 @@ class StreamSerialization { * @param value the value to serialize. * @return the serialized (binary) representation of {@code value}. */ - @SuppressWarnings("unchecked") + @SuppressWarnings({ "unchecked", "rawtypes" }) static byte[] serialize(@Nullable RedisSerializer serializer, Object value) { return canSerialize(serializer, value) ? ((RedisSerializer) serializer).serialize(value) : (byte[]) value; } + /** + * Deserialize the {@code value using the optional {@link RedisSerializer}. If no conversion is possible, return + * {@code value}. @param serializer @param value @param @return + */ + static T deserialize(@Nullable RedisSerializer serializer, ByteBuffer value) { + return deserialize(serializer, ByteUtils.getBytes(value)); + } + + /** + * Deserialize the {@code value using the optional {@link RedisSerializer}. If no conversion is possible, return + * {@code value}. @param serializer @param value @param @return + */ + static T deserialize(@Nullable RedisSerializer serializer, byte[] value) { + return serializer != null ? serializer.deserialize(value) : (T) value; + } + /** * Returns whether the given {@link RedisSerializer} is capable of serializing the {@code value} to {@literal byte[]}. * @@ -47,7 +66,7 @@ class StreamSerialization { * @return {@literal true} if the given {@link RedisSerializer} is capable of serializing the {@code value} to * {@literal byte[]}. */ - private static boolean canSerialize(@Nullable RedisSerializer serializer, Object value) { + private static boolean canSerialize(@Nullable RedisSerializer serializer, @Nullable Object value) { return serializer != null && (value == null || serializer.canSerialize(value.getClass())); } } diff --git a/src/main/java/org/springframework/data/redis/connection/stream/StringRecord.java b/src/main/java/org/springframework/data/redis/connection/stream/StringRecord.java index 6e893ad60..51af565c4 100644 --- a/src/main/java/org/springframework/data/redis/connection/stream/StringRecord.java +++ b/src/main/java/org/springframework/data/redis/connection/stream/StringRecord.java @@ -54,6 +54,6 @@ public interface StringRecord extends MapRecord { * @return new instance of {@link StringRecord}. */ static StringRecord of(MapRecord source) { - return StreamRecords.newRecord().in(source.getStream()).withId(source.getId()).ofStrings(source.getValue()); + return StreamRecords.newRecord().in(source.getRequiredStream()).withId(source.getId()).ofStrings(source.getValue()); } } diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java index ff6e662f3..c265e63da 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java @@ -77,7 +77,7 @@ public interface ReactiveStreamOperations extends HashMapperProvider< * @see Redis Documentation: XACK */ default Mono acknowledge(String group, Record record) { - return acknowledge(record.getStream(), group, record.getId()); + return acknowledge(record.getRequiredStream(), group, record.getId()); } /** diff --git a/src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java b/src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java index 0a7b2d6b1..8cbfb035e 100644 --- a/src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java +++ b/src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java @@ -106,12 +106,11 @@ class StreamObjectMapper { @SuppressWarnings({ "unchecked", "rawtypes" }) static MapRecord toMapRecord(HashMapperProvider provider, Record source) { - if (source instanceof ObjectRecord) { - - ObjectRecord entry = ((ObjectRecord) source); + if (source instanceof ObjectRecord entry) { if (entry.getValue() instanceof Map) { - return StreamRecords.newRecord().in(source.getStream()).withId(source.getId()).ofMap((Map) entry.getValue()); + return StreamRecords.newRecord().in(source.getRequiredStream()).withId(source.getId()) + .ofMap((Map) entry.getValue()); } return entry.toMapRecord(provider.getHashMapper(entry.getValue().getClass())); @@ -122,7 +121,7 @@ class StreamObjectMapper { } return Record.of(((HashMapper) provider.getHashMapper(source.getClass())).toHash(source)) - .withStreamKey(source.getStream()); + .withStreamKey(source.getRequiredStream()); } /** diff --git a/src/main/java/org/springframework/data/redis/core/StreamOperations.java b/src/main/java/org/springframework/data/redis/core/StreamOperations.java index e0f758cc2..1eb5b673c 100644 --- a/src/main/java/org/springframework/data/redis/core/StreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/StreamOperations.java @@ -77,7 +77,7 @@ public interface StreamOperations extends HashMapperProvider * @see Redis Documentation: XACK */ default Long acknowledge(String group, Record record) { - return acknowledge(record.getStream(), group, record.getId()); + return acknowledge(record.getRequiredStream(), group, record.getId()); } /**