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

DATAREDIS-864 - Leave XREAD COUNT unconfigured if not configured through builder.

Browse Source 
XREAD does no longer apply a COUNT if the count was left unconfigured on StreamReadOperationsBuilder. This removes the limitation of batch size 1 and allows generally an increased throughput in default configuration.

Original Pull Request: #356
This commit is contained in:
Mark Paluch
2018-11-16 09:54:32 +01:00
committed by Christoph Strobl
parent eafcb331d8
commit 8858e0f1be
4 changed files with 28 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java
View File

@@ -89,13 +89,17 @@ class DefaultStreamMessageListenerContainer<K, V extends Record<K, ?>> implement
private static StreamReadOptions getStreamReadOptions(StreamMessageListenerContainerOptions<?, ?> options) {
StreamReadOptions streamReadOptions = StreamReadOptions.empty().count(options.getBatchSize());
StreamReadOptions readOptions = StreamReadOptions.empty();
if (!options.getPollTimeout().isZero()) {
streamReadOptions = streamReadOptions.block(options.getPollTimeout());
if (options.getBatchSize().isPresent()) {
readOptions = readOptions.count(options.getBatchSize().getAsInt());
}
return streamReadOptions;
if (!options.getPollTimeout().isZero()) {
readOptions = readOptions.block(options.getPollTimeout());
}
return readOptions;
}
private RedisTemplate<K, V> createRedisTemplate(RedisConnectionFactory connectionFactory,

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

@@ -73,7 +73,11 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
.key(options.getKeySerializer()).hashKey(options.getHashKeySerializer())
.hashValue(options.getHashValueSerializer()).build();
StreamReadOptions readOptions = StreamReadOptions.empty().count(options.getBatchSize());
StreamReadOptions readOptions = StreamReadOptions.empty();
if (options.getBatchSize().isPresent()) {
readOptions = readOptions.count(options.getBatchSize().getAsInt());
}
if (!options.getPollTimeout().isZero()) {
readOptions = readOptions.block(options.getPollTimeout());
}

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

@@ -16,6 +16,7 @@
package org.springframework.data.redis.stream;
import java.time.Duration;
import java.util.OptionalInt;
import java.util.concurrent.Executor;
import java.util.function.Predicate;
@@ -474,7 +475,7 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
class StreamMessageListenerContainerOptions<K, V extends Record<K, ?>> {
private final Duration pollTimeout;
private final int batchSize;
private final @Nullable Integer batchSize;
private final RedisSerializer<K> keySerializer;
private final RedisSerializer<Object> hashKeySerializer;
private final RedisSerializer<Object> hashValueSerializer;
@@ -484,7 +485,8 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
private final Executor executor;
@SuppressWarnings("unchecked")
private StreamMessageListenerContainerOptions(Duration pollTimeout, int batchSize, RedisSerializer<K> keySerializer,
private StreamMessageListenerContainerOptions(Duration pollTimeout, @Nullable Integer batchSize,
RedisSerializer<K> keySerializer,
RedisSerializer<Object> hashKeySerializer, RedisSerializer<Object> hashValueSerializer,
@Nullable Class<?> targetType, @Nullable HashMapper<V, ?, ?> hashMapper, ErrorHandler errorHandler,
Executor executor) {
@@ -518,10 +520,10 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
/**
* Batch size polling using the {@code COUNT} option during reads.
*
* @return
* @return the batch size.
*/
public int getBatchSize() {
return batchSize;
public OptionalInt getBatchSize() {
return batchSize != null ? OptionalInt.of(batchSize) : OptionalInt.empty();
}
public RedisSerializer<K> getKeySerializer() {
@@ -575,7 +577,7 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
class StreamMessageListenerContainerOptionsBuilder<K, V extends Record<K, ?>> {
private Duration pollTimeout = Duration.ofSeconds(2);
private int batchSize = 1;
private @Nullable Integer batchSize;
private RedisSerializer<K> keySerializer;
private RedisSerializer<Object> hashKeySerializer;
private RedisSerializer<Object> hashValueSerializer;

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

@@ -19,6 +19,7 @@ import reactor.core.publisher.Flux;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.OptionalInt;
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
@@ -186,14 +187,14 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
class StreamReceiverOptions<K, V extends Record<K, ?>> {
private final Duration pollTimeout;
private final int batchSize;
private final @Nullable Integer batchSize;
private final SerializationPair<K> keySerializer;
private final SerializationPair<Object> hashKeySerializer;
private final SerializationPair<Object> hashValueSerializer;
private final @Nullable Class<Object> targetType;
private final @Nullable HashMapper<Object, Object, Object> hashMapper;
private StreamReceiverOptions(Duration pollTimeout, int batchSize, SerializationPair<K> keySerializer,
private StreamReceiverOptions(Duration pollTimeout, @Nullable Integer batchSize, SerializationPair<K> keySerializer,
SerializationPair<Object> hashKeySerializer, SerializationPair<Object> hashValueSerializer,
@Nullable Class<?> targetType, @Nullable HashMapper<V, ?, ?> hashMapper) {
@@ -240,10 +241,10 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
/**
* Batch size polling using the {@code COUNT} option during reads.
*
* @return
* @return the batch size.
*/
public int getBatchSize() {
return batchSize;
public OptionalInt getBatchSize() {
return batchSize != null ? OptionalInt.of(batchSize) : OptionalInt.empty();
}
public SerializationPair<K> getKeySerializer() {
@@ -281,7 +282,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
class StreamReceiverOptionsBuilder<K, V extends Record<K, ?>> {
private Duration pollTimeout = Duration.ofSeconds(2);
private int batchSize = 1;
private @Nullable Integer batchSize;
private SerializationPair<K> keySerializer;
private SerializationPair<Object> hashKeySerializer;
private SerializationPair<Object> hashValueSerializer;