Add assertions to fail early on absent values using StreamMessageListenerContainer.

Closes #2472

Author: mp911de

src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java

@@ -183,6 +183,10 @@ public int getPhase() {
 
 	@Override
 	public Subscription register(StreamReadRequest<K> streamRequest, StreamListener<K, V> listener) {
+
+		Assert.notNull(streamRequest, "StreamReadRequest must not be null");
+		Assert.notNull(listener, "StreamListener must not be null");
+
 		return doRegister(getReadTask(streamRequest, listener));
 	}
 

src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java

@@ -337,6 +337,9 @@ class StreamReadRequestBuilder<K> {
 		Predicate<Throwable> cancelSubscriptionOnError = t -> true;
 
 		StreamReadRequestBuilder(StreamOffset<K> streamOffset) {
+
+			Assert.notNull(streamOffset, "StreamOffset must not be null");
+
 			this.streamOffset = streamOffset;
 		}
 
@@ -355,6 +358,8 @@ class StreamReadRequestBuilder<K> {
 		 */
 		public StreamReadRequestBuilder<K> errorHandler(ErrorHandler errorHandler) {
 
+			Assert.notNull(errorHandler, "ErrorHandler must not be null");
+
 			this.errorHandler = errorHandler;
 			return this;
 		}
@@ -368,6 +373,7 @@ public StreamReadRequestBuilder<K> errorHandler(ErrorHandler errorHandler) {
 		 */
 		public StreamReadRequestBuilder<K> cancelOnError(Predicate<Throwable> cancelSubscriptionOnError) {
 
+			Assert.notNull(cancelSubscriptionOnError, "cancelSubscriptionOnError Predicate must not be null");
 			this.cancelSubscriptionOnError = cancelSubscriptionOnError;
 			return this;
 		}
@@ -440,6 +446,8 @@ public ConsumerStreamReadRequestBuilder<K> cancelOnError(Predicate<Throwable> ca
 		 */
 		public ConsumerStreamReadRequestBuilder<K> consumer(Consumer consumer) {
 
+			Assert.notNull(consumer, "Consumer must not be null");
+
 			this.consumer = consumer;
 			return this;
 		}