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

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

Author: bandalgomsu

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

@@ -50,6 +50,7 @@
  *
  * @author Mark Paluch
  * @author Christoph Strobl
+ * @author Su Ko
  * @since 2.2
  */
 class DefaultStreamMessageListenerContainer<K, V extends Record<K, ?>> implements StreamMessageListenerContainer<K, V> {
@@ -67,6 +68,9 @@ class DefaultStreamMessageListenerContainer<K, V extends Record<K, ?>> implement
 
 	private boolean running = false;
 
+    private int phase = Integer.MAX_VALUE;
+    private boolean autoStartup = false;
+
 	/**
 	 * Create a new {@link DefaultStreamMessageListenerContainer}.
 	 *
@@ -123,9 +127,21 @@ private RedisTemplate<K, V> createRedisTemplate(RedisConnectionFactory connectio
 
 	@Override
 	public boolean isAutoStartup() {
-		return false;
+		return this.autoStartup;
 	}
 
+    /**
+     * Configure if this Lifecycle connection factory should get started automatically by the container at the time that
+     * the containing ApplicationContext gets refreshed.
+     * The default is {@code true}.
+     *
+     * @see org.springframework.context.SmartLifecycle#isAutoStartup()
+     * @since 4.0.0
+     */
+    public void setAutoStartup(boolean autoStartup) {
+        this.autoStartup = autoStartup;
+    }
+
 	@Override
 	public void stop(Runnable callback) {
 
@@ -177,9 +193,21 @@ public boolean isRunning() {
 
 	@Override
 	public int getPhase() {
-		return Integer.MAX_VALUE;
+		return this.phase;
 	}
 
+    /**
+     * Specify the lifecycle phase for this container.
+     * Lower values start earlier and stop later.
+     * The default is {@code Integer.MAX_VALUE}.
+     *
+     * @see org.springframework.context.SmartLifecycle#getPhase()
+     * @since 4.0.0
+     */
+    public void setPhase(int phase) {
+        this.phase = phase;
+    }
+
 	@Override
 	public Subscription register(StreamReadRequest<K> streamRequest, StreamListener<K, V> listener) {