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

DATAREDIS-864 - Polishing.

Browse Source 
Update Javadoc, remove unused code and fix some warnings.

Original Pull Request: #356
This commit is contained in:
Christoph Strobl
2018-11-22 10:41:42 +01:00
parent 8858e0f1be
commit dc0ce19cd0
12 changed files with 75 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -15,7 +15,6 @@
*/
package org.springframework.data.redis.connection;
import org.springframework.data.redis.connection.stream.StreamRecords;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -31,13 +30,14 @@ import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse;
import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
import org.springframework.data.redis.connection.stream.ByteBufferRecord;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.connection.stream.StreamReadOptions;
import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
@@ -139,7 +139,7 @@ public interface ReactiveStreamCommands {
* @param key the stream key.
* @param group name of the consumer group.
* @param recordIds record Id's to acknowledge.
* @return
* @return {@link Mono} emitting the nr of acknowledged messages.
* @see <a href="http://redis.io/commands/xadd">Redis Documentation: XADD</a>
*/
default Mono<Long> xAck(ByteBuffer key, String group, String... recordIds) {
@@ -157,7 +157,7 @@ public interface ReactiveStreamCommands {
* @param key the stream key.
* @param group name of the consumer group.
* @param recordIds record Id's to acknowledge.
* @return
* @return {@link Mono} emitting the nr of acknowledged messages.
* @see <a href="http://redis.io/commands/xadd">Redis Documentation: XADD</a>
*/
default Mono<Long> xAck(ByteBuffer key, String group, RecordId... recordIds) {
@@ -173,7 +173,7 @@ public interface ReactiveStreamCommands {
* Acknowledge one or more records as processed.
*
* @param commands must not be {@literal null}.
* @return
* @return {@link Flux} emitting the nr of acknowledged messages per {@link AcknowledgeCommand}.
* @see <a href="http://redis.io/commands/xack">Redis Documentation: XACK</a>
*/
Flux<NumericResponse<AcknowledgeCommand, Long>> xAck(Publisher<AcknowledgeCommand> commands);
@@ -230,7 +230,7 @@ public interface ReactiveStreamCommands {
}
/**
* @return
* @return the actual {@link ByteBufferRecord#getValue()}
*/
public Map<ByteBuffer, ByteBuffer> getBody() {
return record.getValue();
@@ -246,7 +246,7 @@ public interface ReactiveStreamCommands {
*
* @param key must not be {@literal null}.
* @param body must not be {@literal null}.
* @return
* @return {@link Mono} emitting the server generated {@link RecordId id}.
* @see <a href="http://redis.io/commands/xadd">Redis Documentation: XADD</a>
*/
default Mono<RecordId> xAdd(ByteBuffer key, Map<ByteBuffer, ByteBuffer> body) {
@@ -261,7 +261,7 @@ public interface ReactiveStreamCommands {
* Add stream record with given {@literal body} to {@literal key}.
*
* @param record must not be {@literal null}.
* @return
* @return {@link Mono} the {@link RecordId id}.
* @see <a href="http://redis.io/commands/xadd">Redis Documentation: XADD</a>
*/
default Mono<RecordId> xAdd(ByteBufferRecord record) {
@@ -275,7 +275,7 @@ public interface ReactiveStreamCommands {
* Add stream record with given {@literal body} to {@literal key}.
*
* @param commands must not be {@literal null}.
* @return
* @return {@link Flux} emitting the {@link RecordId} on by for for the given {@link AddStreamRecord} commands.
* @see <a href="http://redis.io/commands/xadd">Redis Documentation: XADD</a>
*/
Flux<CommandResponse<AddStreamRecord, RecordId>> xAdd(Publisher<AddStreamRecord> commands);
@@ -349,7 +349,7 @@ public interface ReactiveStreamCommands {
*
* @param key the stream key.
* @param recordIds stream record Id's.
* @return number of removed entries.
* @return {@link Mono} emitting the number of removed entries.
* @see <a href="http://redis.io/commands/xdel">Redis Documentation: XDEL</a>
*/
default Mono<Long> xDel(ByteBuffer key, String... recordIds) {
@@ -366,7 +366,7 @@ public interface ReactiveStreamCommands {
*
* @param key the stream key.
* @param recordIds stream record Id's.
* @return number of removed entries.
* @return {@link Mono} emitting the number of removed entries.
* @see <a href="http://redis.io/commands/xdel">Redis Documentation: XDEL</a>
*/
default Mono<Long> xDel(ByteBuffer key, RecordId... recordIds) {
@@ -382,7 +382,7 @@ public interface ReactiveStreamCommands {
* number of IDs passed in case certain IDs do not exist.
*
* @param commands must not be {@literal null}.
* @return
* @return {@link Mono} emitting the number of removed entries.
* @see <a href="http://redis.io/commands/xdel">Redis Documentation: XDEL</a>
*/
Flux<CommandResponse<DeleteCommand, Long>> xDel(Publisher<DeleteCommand> commands);
@@ -391,7 +391,7 @@ public interface ReactiveStreamCommands {
* Get the size of the stream stored at {@literal key}.
*
* @param key must not be {@literal null}.
* @return length of the stream.
* @return {@link Mono} emitting the length of the stream.
* @see <a href="http://redis.io/commands/xlen">Redis Documentation: XLEN</a>
*/
default Mono<Long> xLen(ByteBuffer key) {
@@ -405,7 +405,7 @@ public interface ReactiveStreamCommands {
* Get the size of the stream stored at {@link KeyCommand#getKey()}
*
* @param commands must not be {@literal null}.
* @return
* @return {@link Flux} emitting the length of the stream per {@link KeyCommand}.
* @see <a href="http://redis.io/commands/xlen">Redis Documentation: XLEN</a>
*/
Flux<NumericResponse<KeyCommand, Long>> xLen(Publisher<KeyCommand> commands);
@@ -501,7 +501,7 @@ public interface ReactiveStreamCommands {
*
* @param key the stream key.
* @param range must not be {@literal null}.
* @return list with members of the resulting stream.
* @return {@link Flux} emitting with members of the stream.
* @see <a href="http://redis.io/commands/xrange">Redis Documentation: XRANGE</a>
*/
default Flux<ByteBufferRecord> xRange(ByteBuffer key, Range<String> range) {
@@ -514,7 +514,7 @@ public interface ReactiveStreamCommands {
* @param key the stream key.
* @param range must not be {@literal null}.
* @param limit must not be {@literal null}.
* @return list with members of the resulting stream.
* @return {@link Flux} emitting with members of the stream.
* @see <a href="http://redis.io/commands/xrange">Redis Documentation: XRANGE</a>
*/
default Flux<ByteBufferRecord> xRange(ByteBuffer key, Range<String> range, Limit limit) {
@@ -531,7 +531,7 @@ public interface ReactiveStreamCommands {
* Read records from a stream within a specific {@link Range} applying a {@link Limit}.
*
* @param commands must not be {@literal null}.
* @return
* @return {@link Flux} emitting with members of the stream per {@link RangeCommand}.
* @see <a href="http://redis.io/commands/xrange">Redis Documentation: XRANGE</a>
*/
Flux<CommandResponse<RangeCommand, Flux<ByteBufferRecord>>> xRange(Publisher<RangeCommand> commands);
@@ -633,7 +633,7 @@ public interface ReactiveStreamCommands {
* Read records from one or more {@link StreamOffset}s.
*
* @param streams the streams to read from.
* @return list with members of the resulting stream.
* @return {@link Flux} emitting with members of the stream.
* @see <a href="http://redis.io/commands/xread">Redis Documentation: XREAD</a>
*/
default Flux<ByteBufferRecord> xRead(StreamOffset<ByteBuffer>... streams) {
@@ -645,7 +645,7 @@ public interface ReactiveStreamCommands {
*
* @param readOptions read arguments.
* @param streams the streams to read from.
* @return list with members of the resulting stream.
* @return {@link Flux} emitting with members of the stream.
* @see <a href="http://redis.io/commands/xread">Redis Documentation: XREAD</a>
*/
default Flux<ByteBufferRecord> xRead(StreamReadOptions readOptions, StreamOffset<ByteBuffer>... streams) {
@@ -661,7 +661,7 @@ public interface ReactiveStreamCommands {
* Read records from one or more {@link StreamOffset}s.
*
* @param commands must not be {@literal null}.
* @return list with members of the resulting stream.
* @return {@link Flux} emitting the members of the stream per {@link ReadCommand}.
* @see <a href="http://redis.io/commands/xread">Redis Documentation: XREAD</a>
* @see <a href="http://redis.io/commands/xreadgroup">Redis Documentation: XREADGROUP</a>
*/
@@ -675,7 +675,7 @@ public interface ReactiveStreamCommands {
private final @Nullable ReadOffset offset;
public GroupCommand(@Nullable ByteBuffer key, GroupCommandAction action, @Nullable String groupName,
@Nullable String consumerName, ReadOffset offset) {
@Nullable String consumerName, @Nullable ReadOffset offset) {
super(key);
this.action = action;
@@ -732,7 +732,7 @@ public interface ReactiveStreamCommands {
}
public enum GroupCommandAction {
CREATE, SET_ID, DESTROY, DELETE_CONSUMER;
CREATE, SET_ID, DESTROY, DELETE_CONSUMER
}
}
@@ -800,7 +800,7 @@ public interface ReactiveStreamCommands {
* Execute the given {@link GroupCommand} to {@literal create, destroy,... } groups.
*
* @param commands
* @return
* @return {@link Flux} emitting the results of the {@link GroupCommand} one by one.
*/
Flux<CommandResponse<GroupCommand, String>> xGroup(Publisher<GroupCommand> commands);
@@ -809,7 +809,7 @@ public interface ReactiveStreamCommands {
*
* @param consumer consumer/group.
* @param streams the streams to read from.
* @return list with members of the resulting stream.
* @return {@link Flux} emitting the members of the stream
* @see <a href="http://redis.io/commands/xreadgroup">Redis Documentation: XREADGROUP</a>
*/
default Flux<ByteBufferRecord> xReadGroup(Consumer consumer, StreamOffset<ByteBuffer>... streams) {
@@ -822,7 +822,7 @@ public interface ReactiveStreamCommands {
* @param consumer consumer/group.
* @param readOptions read arguments.
* @param streams the streams to read from.
* @return list with members of the resulting stream.
* @return {@link Flux} emitting the members of the stream.
* @see <a href="http://redis.io/commands/xreadgroup">Redis Documentation: XREADGROUP</a>
*/
default Flux<ByteBufferRecord> xReadGroup(Consumer consumer, StreamReadOptions readOptions,
@@ -841,7 +841,7 @@ public interface ReactiveStreamCommands {
*
* @param key the stream key.
* @param range must not be {@literal null}.
* @return list with members of the resulting stream.
* @return {@link Flux} emitting the members of the stream in reverse.
* @see <a href="http://redis.io/commands/xrevrange">Redis Documentation: XREVRANGE</a>
*/
default Flux<ByteBufferRecord> xRevRange(ByteBuffer key, Range<String> range) {
@@ -854,7 +854,7 @@ public interface ReactiveStreamCommands {
* @param key the stream key.
* @param range must not be {@literal null}.
* @param limit must not be {@literal null}.
* @return list with members of the resulting stream.
* @return {@link Flux} emitting the members of the stream in reverse.
* @see <a href="http://redis.io/commands/xrevrange">Redis Documentation: XREVRANGE</a>
*/
default Flux<ByteBufferRecord> xRevRange(ByteBuffer key, Range<String> range, Limit limit) {
@@ -871,7 +871,7 @@ public interface ReactiveStreamCommands {
* Read records from a stream within a specific {@link Range} applying a {@link Limit} in reverse order.
*
* @param commands must not be {@literal null}.
* @return
* @return {@link Flux} emitting the members of the stream in reverse.
* @see <a href="http://redis.io/commands/xrevrange">Redis Documentation: XREVRANGE</a>
*/
Flux<CommandResponse<RangeCommand, Flux<ByteBufferRecord>>> xRevRange(Publisher<RangeCommand> commands);
@@ -929,7 +929,7 @@ public interface ReactiveStreamCommands {
*
* @param key the stream key.
* @param count length of the stream.
* @return number of removed entries.
* @return {@link Mono} emitting the number of removed entries.
* @see <a href="http://redis.io/commands/xtrim">Redis Documentation: XTRIM</a>
*/
default Mono<Long> xTrim(ByteBuffer key, long count) {
@@ -943,7 +943,7 @@ public interface ReactiveStreamCommands {
* Trims the stream to {@code count} elements.
*
* @param commands must not be {@literal null}.
* @return
* @return {@link Flux} emitting the number of removed entries per {@link TrimCommand}.
* @see <a href="http://redis.io/commands/xtrim">Redis Documentation: XTRIM</a>
*/
Flux<NumericResponse<KeyCommand, Long>> xTrim(Publisher<TrimCommand> commands);

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

@@ -120,6 +120,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands {
}
@Override
@SuppressWarnings({"unchecked", "rawtypes"})
public Flux<CommandResponse<GroupCommand, String>> xGroup(Publisher<GroupCommand> commands) {
return connection.execute(cmd -> Flux.from(commands).concatMap(command -> {

2
src/main/java/org/springframework/data/redis/connection/stream/StreamRecords.java
View File

@@ -94,7 +94,7 @@ public class StreamRecords {
/**
* Obtain new instance of {@link RecordBuilder} to fluently create {@link Record records}.
*
* @return
* @return new instance of {@link RecordBuilder}.
*/
public static RecordBuilder<?> newRecord() {
return new RecordBuilder<>(null, RecordId.autoGenerate());

11
src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java
View File

@@ -123,6 +123,7 @@ class DefaultStreamOperations<K, HK, HV> extends AbstractOperations<K, Object> i
*/
@Nullable
@Override
@SuppressWarnings("unchecked")
public RecordId add(Record<K, ?> record) {
Assert.notNull(record, "Record must not be null");
@@ -278,11 +279,7 @@ class DefaultStreamOperations<K, HK, HV> extends AbstractOperations<K, Object> i
return hashValueSerializer() != null;
}
protected HV deserializeHashValue(byte[] bytes, Class<HV> targetType) {
return hashValueSerializerPresent() ? (HV) hashValueSerializer().deserialize(bytes)
: objectMapper.getConversionService().convert(bytes, targetType);
}
@SuppressWarnings({"rawtypes", "unchecked"})
private byte[] serialize(Object value, RedisSerializer serializer) {
Object _value = value;
@@ -302,9 +299,13 @@ class DefaultStreamOperations<K, HK, HV> extends AbstractOperations<K, Object> i
abstract class RecordDeserializingRedisCallback implements RedisCallback<List<MapRecord<K, HK, HV>>> {
@SuppressWarnings("unchecked")
public final List<MapRecord<K, HK, HV>> doInRedis(RedisConnection connection) {
List<ByteRecord> raw = inRedis(connection);
if(raw == null) {
return Collections.emptyList();
}
List<MapRecord<K, HK, HV>> result = new ArrayList<>();
for (ByteRecord record : raw) {

8
src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java
View File

@@ -164,7 +164,7 @@ public interface ReactiveStreamOperations<K, HK, HV> extends HashMapperProvider<
/**
* Create a consumer group at the {@link ReadOffset#latest() latest offset}.
*
* @param key
* @param key the {@literal key} the stream is stored at.
* @param group name of the consumer group.
* @return the {@link Mono} emitting {@literal OK} if successful.. {@literal null} when used in pipeline /
* transaction.
@@ -176,8 +176,8 @@ public interface ReactiveStreamOperations<K, HK, HV> extends HashMapperProvider<
/**
* Create a consumer group.
*
* @param key
* @param readOffset
* @param key the {@literal key} the stream is stored at.
* @param readOffset the {@link ReadOffset} to apply.
* @param group name of the consumer group.
* @return the {@link Mono} emitting {@literal OK} if successful.
*/
@@ -270,6 +270,7 @@ public interface ReactiveStreamOperations<K, HK, HV> extends HashMapperProvider<
* @return the {@link Flux} emitting records one by one.
* @see <a href="http://redis.io/commands/xread">Redis Documentation: XREAD</a>
*/
@SuppressWarnings("unchecked")
default Flux<MapRecord<K, HK, HV>> read(StreamOffset<K> stream) {
Assert.notNull(stream, "StreamOffset must not be null");
@@ -285,6 +286,7 @@ public interface ReactiveStreamOperations<K, HK, HV> extends HashMapperProvider<
* @return the {@link Flux} emitting records one by one.
* @see <a href="http://redis.io/commands/xread">Redis Documentation: XREAD</a>
*/
@SuppressWarnings("unchecked")
default <V> Flux<ObjectRecord<K, V>> read(Class<V> targetType, StreamOffset<K> stream) {
Assert.notNull(stream, "StreamOffset must not be null");

9
src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java
View File

@@ -137,7 +137,7 @@ class StreamObjectMapper {
* single-element and multi-element list transformation.l
*
* @param records the {@link MapRecord} that should be mapped.
* @param hashMapperProvider
* @param hashMapperProvider the provider to obtain the actual {@link HashMapper} from. Must not be {@literal null}.
* @param targetType the requested {@link Class target type}.
* @return the resulting {@link List} of {@link ObjectRecord} or {@literal null} if {@code records} was
* {@literal null}.
@@ -178,14 +178,17 @@ class StreamObjectMapper {
*
* @param conversionService the used {@link ConversionService}.
* @param targetType the target type.
* @return
* @return obtain the {@link HashMapper} for a certain type.
*/
protected HashMapper<?, ?, ?> doGetHashMapper(ConversionService conversionService, Class<?> targetType) {
return this.objectHashMapper != null ? objectHashMapper : this.mapper;
}
/**
* @param targetType
* Check if the given type is a simple type as in
* {@link org.springframework.data.convert.CustomConversions#isSimpleType(Class)}.
*
* @param targetType the type to inspect. Must not be {@literal null}.
* @return {@literal true} if {@link Class targetType} is a simple type.
* @see org.springframework.data.convert.CustomConversions#isSimpleType(Class)
*/

6
src/main/java/org/springframework/data/redis/core/StreamOperations.java
View File

@@ -162,7 +162,7 @@ public interface StreamOperations<K, HK, HV> extends HashMapperProvider<HK, HV>
/**
* Create a consumer group at the {@link ReadOffset#latest() latest offset}.
*
* @param key
* @param key the {@literal key} the stream is stored at.
* @param group name of the consumer group.
* @return {@literal OK} if successful. {@literal null} when used in pipeline / transaction.
*/
@@ -173,8 +173,8 @@ public interface StreamOperations<K, HK, HV> extends HashMapperProvider<HK, HV>
/**
* Create a consumer group.
*
* @param key
* @param readOffset
* @param key the {@literal key} the stream is stored at.
* @param readOffset the {@link ReadOffset} to apply.
* @param group name of the consumer group.
* @return {@literal OK} if successful. {@literal null} when used in pipeline / transaction.
*/

27
src/main/java/org/springframework/data/redis/hash/ObjectHashMapper.java
View File

@@ -152,33 +152,6 @@ public class ObjectHashMapper implements HashMapper<Object, byte[], byte[]> {
return type.cast(fromHash(hash));
}
public Map<String, Object> toObjectHash(Object source) {
Map<byte[], byte[]> raw = toHash(source);
RedisPersistentEntity<?> entity = converter.getMappingContext().getPersistentEntity(source.getClass());
Map<String, Object> result = new LinkedHashMap<>();
for(Map.Entry<byte[], byte[]> entry : raw.entrySet()) {
String key = converter.fromBytes(entry.getKey(), String.class);
Object value = entry.getValue();
try {
value = converter.fromBytes(entry.getValue(), PropertyPath.from(key, entity.getTypeInformation()).getType());
} catch (PropertyReferenceException e) {
value = converter.fromBytes(entry.getValue(), String.class);
} catch (ConverterNotFoundException cnfe) {
// value = fromHash(entry)
// TODO: nested ones!
}
result.put(key, value);
}
return result;
}
/**
* {@link ReferenceResolver} implementation always returning an empty {@link Map}.
*

20
src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java
View File

@@ -193,8 +193,6 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
/**
* Arm the subscription so {@link Subscription#request(long) demand} activates polling.
*
* @return
*/
void arm() {
@@ -456,8 +454,8 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
/**
* Create a new state object for standalone-read.
*
* @param offset
* @return
* @param offset the {@link ReadOffset} from where to start. Must not be {@literal null}.
* @return new instance of {@link PollState}.
*/
static PollState standalone(ReadOffset offset) {
@@ -468,9 +466,9 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
/**
* Create a new state object for consumergroup-read.
*
* @param consumer
* @param offset
* @return
* @param consumer the {@link Consumer}. Must not be {@literal null}.
* @param offset the {@link ReadOffset} from where to start. Must not be {@literal null}.
* @return new instance of {@link PollState}.
*/
static PollState consumer(Consumer consumer, ReadOffset offset) {
@@ -496,7 +494,7 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
/**
* Decrement request count to indicate that an element was emitted.
*
* @return
* @return {@literal true} if there are still requests pending and the decrement has been successful.
*/
boolean decrementRequested() {
@@ -526,9 +524,9 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
/**
* Update demand.
*
* @param expect
* @param update
* @return
* @param expect the number of requests still pending.
* @param update the increment to apply
* @return {@literal true} if successful.
*/
boolean setRequested(long expect, long update) {
return requestsPending.compareAndSet(expect, update);

2
src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java
View File

@@ -511,7 +511,7 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
/**
* Timeout for blocking polling using the {@code BLOCK} option during reads.
*
* @return
* @return the timeout.
*/
public Duration getPollTimeout() {
return pollTimeout;

10
src/main/java/org/springframework/data/redis/stream/StreamPollTask.java
View File

@@ -184,8 +184,8 @@ class StreamPollTask<K, V extends Record<K, ?>> implements Task {
/**
* Create a new state object for standalone-read.
*
* @param offset
* @return
* @param offset the {@link ReadOffset} to use.
* @return new instance of {@link PollState}.
*/
static PollState standalone(ReadOffset offset) {
@@ -196,9 +196,9 @@ class StreamPollTask<K, V extends Record<K, ?>> implements Task {
/**
* Create a new state object for consumergroup-read.
*
* @param consumer
* @param offset
* @return
* @param consumer the {@link Consumer} to use.
* @param offset the {@link ReadOffset} to apply.
* @return new instance of {@link PollState}.
*/
static PollState consumer(Consumer consumer, ReadOffset offset) {

10
src/main/java/org/springframework/data/redis/stream/StreamReceiver.java
View File

@@ -194,6 +194,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
private final @Nullable Class<Object> targetType;
private final @Nullable HashMapper<Object, Object, Object> hashMapper;
@SuppressWarnings({ "unchecked", "rawtypes" })
private StreamReceiverOptions(Duration pollTimeout, @Nullable Integer batchSize, SerializationPair<K> keySerializer,
SerializationPair<Object> hashKeySerializer, SerializationPair<Object> hashValueSerializer,
@Nullable Class<?> targetType, @Nullable HashMapper<V, ?, ?> hashMapper) {
@@ -232,7 +233,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
/**
* Timeout for blocking polling using the {@code BLOCK} option during reads.
*
* @return
* @return the actual timeout.
*/
public Duration getPollTimeout() {
return pollTimeout;
@@ -326,6 +327,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
* @param pair must not be {@literal null}.
* @return {@code this} {@link StreamReceiverOptionsBuilder}.
*/
@SuppressWarnings("unchecked")
public <T> StreamReceiverOptionsBuilder<T, MapRecord<T, T, T>> serializer(SerializationPair<T> pair) {
Assert.notNull(pair, "SerializationPair must not be null");
@@ -339,9 +341,10 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
/**
* Configure a key, hash key and hash value serializer.
*
* @param pair must not be {@literal null}.
* @param serializationContext must not be {@literal null}.
* @return {@code this} {@link StreamReceiverOptionsBuilder}.
*/
@SuppressWarnings("unchecked")
public <T> StreamReceiverOptionsBuilder<T, MapRecord<T, T, T>> serializer(
RedisSerializationContext<T, ?> serializationContext) {
@@ -360,6 +363,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
* @param pair must not be {@literal null}.
* @return {@code this} {@link StreamReceiverOptionsBuilder}.
*/
@SuppressWarnings("unchecked")
public <NK, NV extends Record<NK, ?>> StreamReceiverOptionsBuilder<NK, NV> keySerializer(
SerializationPair<NK> pair) {
@@ -375,6 +379,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
* @param pair must not be {@literal null}.
* @return {@code this} {@link StreamReceiverOptionsBuilder}.
*/
@SuppressWarnings("unchecked")
public <HK, HV> StreamReceiverOptionsBuilder<K, MapRecord<K, HK, HV>> hashKeySerializer(
SerializationPair<HK> pair) {
@@ -390,6 +395,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
* @param pair must not be {@literal null}.
* @return {@code this} {@link StreamReceiverOptionsBuilder}.
*/
@SuppressWarnings("unchecked")
public <HK, HV> StreamReceiverOptionsBuilder<K, MapRecord<K, HK, HV>> hashValueSerializer(
SerializationPair<HK> pair) {