[GH-3208] Feat: Allow configuring 'autoStartup' and 'phase' in StreamMessageListenerContainerOptions

Signed-off-by: Su Ko <[email protected]>

Author: bandalgomsu

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

@@ -94,6 +94,14 @@ class DefaultStreamMessageListenerContainer<K, V extends Record<K, ?>> implement
 		} else {
 			this.streamOperations = this.template.opsForStream();
 		}
+
+        if(containerOptions.isAutoStartup().isPresent()){
+            this.autoStartup = containerOptions.isAutoStartup().get();
+        }
+
+        if(containerOptions.getPhase().isPresent()){
+            this.phase = containerOptions.getPhase().getAsInt();
+        }
 	}
 
 	private static StreamReadOptions getStreamReadOptions(StreamMessageListenerContainerOptions<?, ?> options) {

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

@@ -16,6 +16,7 @@
 package org.springframework.data.redis.stream;
 
 import java.time.Duration;
+import java.util.Optional;
 import java.util.OptionalInt;
 import java.util.concurrent.Executor;
 import java.util.function.Predicate;
@@ -107,6 +108,7 @@
  * @author Christoph Strobl
  * @author Christian Rest
  * @author DongCheol Kim
+ * @author Su Ko
  * @param <K> Stream key and Stream field type.
  * @param <V> Stream value type.
  * @since 2.2
@@ -503,12 +505,14 @@ class StreamMessageListenerContainerOptions<K, V extends Record<K, ?>> {
 		private final @Nullable HashMapper<Object, Object, Object> hashMapper;
 		private final ErrorHandler errorHandler;
 		private final Executor executor;
+        private final @Nullable Integer phase;
+        private final @Nullable Boolean autoStartup;
 
 		@SuppressWarnings({ "unchecked", "rawtypes" })
 		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) {
+				@Nullable HashMapper<V, ?, ?> hashMapper, ErrorHandler errorHandler, Executor executor,@Nullable Integer phase, @Nullable Boolean autoStartup) {
 			this.pollTimeout = pollTimeout;
 			this.batchSize = batchSize;
 			this.keySerializer = keySerializer;
@@ -518,6 +522,8 @@ private StreamMessageListenerContainerOptions(Duration pollTimeout, @Nullable In
 			this.hashMapper = (HashMapper) hashMapper;
 			this.errorHandler = errorHandler;
 			this.executor = executor;
+            this.phase = phase;
+            this.autoStartup = autoStartup;
 		}
 
 		/**
@@ -598,6 +604,21 @@ public Executor getExecutor() {
 			return executor;
 		}
 
+        /**
+         * @return the phase.
+         * @since 4.0
+         */
+        public OptionalInt getPhase() {
+            return phase != null ? OptionalInt.of(phase) : OptionalInt.empty();
+        }
+
+        /**
+         * @return the autoStartup.
+         * @since 4.0
+         */
+        public Optional<Boolean> isAutoStartup() {
+            return autoStartup != null ? Optional.of(autoStartup) : Optional.empty();
+        }
 	}
 
 	/**
@@ -618,6 +639,8 @@ class StreamMessageListenerContainerOptionsBuilder<K, V extends Record<K, ?>> {
 		private @Nullable Class<?> targetType;
 		private ErrorHandler errorHandler = LoggingErrorHandler.INSTANCE;
 		private Executor executor = new SimpleAsyncTaskExecutor();
+        private @Nullable Integer phase;
+        private @Nullable Boolean autoStartup;
 
 		@SuppressWarnings("NullAway")
 		private StreamMessageListenerContainerOptionsBuilder() {}
@@ -679,6 +702,28 @@ public StreamMessageListenerContainerOptionsBuilder<K, V> errorHandler(ErrorHand
 			return this;
 		}
 
+        /**
+         * Configure a phase for the {@link SmartLifecycle}
+         *
+         * @return {@code this} {@link StreamMessageListenerContainerOptionsBuilder}.
+         * @since 4.0
+         */
+        public StreamMessageListenerContainerOptionsBuilder<K, V> phase(int phase) {
+            this.phase = phase;
+            return this;
+        }
+
+        /**
+         * Configure a autoStartup for the {@link SmartLifecycle}
+         *
+         * @return {@code this} {@link StreamMessageListenerContainerOptionsBuilder}.
+         * @since 4.0
+         */
+        public StreamMessageListenerContainerOptionsBuilder<K, V> autoStartup(boolean autoStartup) {
+            this.autoStartup = autoStartup;
+            return this;
+        }
+
 		/**
 		 * Configure a key, hash key and hash value serializer.
 		 *
@@ -796,7 +841,7 @@ public StreamMessageListenerContainerOptions<K, V> build() {
 			Assert.notNull(hashValueSerializer, "Hash Value Serializer must not be null");
 
 			return new StreamMessageListenerContainerOptions<>(pollTimeout, batchSize, keySerializer, hashKeySerializer,
-					hashValueSerializer, targetType, hashMapper, errorHandler, executor);
+					hashValueSerializer, targetType, hashMapper, errorHandler, executor,phase,autoStartup);
 		}
 
 	}

src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java

@@ -252,7 +252,7 @@ void shouldApplyConfiguredPhase() {
     }
 
     @Test // GH-3208
-    void defaultAutoStartupShouldBeMaxValue() {
+    void defaultAutoStartupShouldBeTrue() {
         assertThat(container.isAutoStartup()).isEqualTo(true);
     }
 

src/test/java/org/springframework/data/redis/stream/AbstractStreamMessageListenerContainerIntegrationTests.java

@@ -384,6 +384,44 @@ void containerRestartShouldRestartSubscription() throws InterruptedException {
 		cancelAwait(subscription);
 	}
 
+    @Test // GH-3208
+    void defaultPhaseShouldBeMaxValue() {
+        StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
+                .create(connectionFactory, containerOptions);
+
+        assertThat(container.getPhase()).isEqualTo(Integer.MAX_VALUE);
+    }
+
+    @Test // GH-3208
+    void shouldApplyConfiguredPhase() {
+        StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options = StreamMessageListenerContainerOptions.builder()
+                .phase(3208)
+                .build();
+        StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
+                .create(connectionFactory, options);
+
+        assertThat(container.getPhase()).isEqualTo(3208);
+    }
+
+    @Test // GH-3208
+    void defaultAutoStartupShouldBeFalse() {
+        StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
+                .create(connectionFactory, containerOptions);
+
+        assertThat(container.isAutoStartup()).isEqualTo(false);
+    }
+
+    @Test // GH-3208
+    void shouldApplyConfiguredAutoStartup() {
+        StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options = StreamMessageListenerContainerOptions.builder()
+                .autoStartup(true)
+                .build();
+        StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
+                .create(connectionFactory, options);
+
+        assertThat(container.isAutoStartup()).isEqualTo(true);
+    }
+
 	private static void cancelAwait(Subscription subscription) {
 
 		subscription.cancel();