diff --git a/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java index b1db9699a..e1581bb3b 100644 --- a/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java @@ -211,6 +211,10 @@ class DefaultStreamMessageListenerContainer> implement */ @Override public Subscription register(StreamReadRequest streamRequest, StreamListener listener) { + + Assert.notNull(streamRequest, "StreamReadRequest must not be null"); + Assert.notNull(listener, "StreamListener must not be null"); + return doRegister(getReadTask(streamRequest, listener)); } diff --git a/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java index e0733155d..304b4d355 100644 --- a/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java @@ -346,6 +346,9 @@ public interface StreamMessageListenerContainer> exten Predicate cancelSubscriptionOnError = t -> true; StreamReadRequestBuilder(StreamOffset streamOffset) { + + Assert.notNull(streamOffset, "StreamOffset must not be null"); + this.streamOffset = streamOffset; } @@ -364,6 +367,8 @@ public interface StreamMessageListenerContainer> exten */ public StreamReadRequestBuilder errorHandler(ErrorHandler errorHandler) { + Assert.notNull(errorHandler, "ErrorHandler must not be null"); + this.errorHandler = errorHandler; return this; } @@ -377,6 +382,7 @@ public interface StreamMessageListenerContainer> exten */ public StreamReadRequestBuilder cancelOnError(Predicate cancelSubscriptionOnError) { + Assert.notNull(cancelSubscriptionOnError, "cancelSubscriptionOnError Predicate must not be null"); this.cancelSubscriptionOnError = cancelSubscriptionOnError; return this; } @@ -449,6 +455,8 @@ public interface StreamMessageListenerContainer> exten */ public ConsumerStreamReadRequestBuilder consumer(Consumer consumer) { + Assert.notNull(consumer, "Consumer must not be null"); + this.consumer = consumer; return this; }