GH-2451: Add StreamAdmin

Resolves https://github.com/spring-projects/spring-amqp/issues/2451

**cherry-pick to 2.4.x**

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/support/StreamAdmin.java

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.rabbit.stream.support;
+
+import java.util.function.Consumer;
+
+import org.springframework.context.SmartLifecycle;
+import org.springframework.util.Assert;
+
+import com.rabbitmq.stream.Environment;
+import com.rabbitmq.stream.StreamCreator;
+
+/**
+ * Used to provision streams.
+ *
+ * @author Gary Russell
+ * @since 2.4.13
+ *
+ */
+public class StreamAdmin implements SmartLifecycle {
+
+	private final StreamCreator streamCreator;
+
+	private final Consumer<StreamCreator> callback;
+
+	private boolean autoStartup = true;
+
+	private int phase;
+
+	private volatile boolean running;
+
+	/**
+	 * Construct with the provided parameters.
+	 * @param env the environment.
+	 * @param callback the callback to receive the {@link StreamCreator}.
+	 */
+	public StreamAdmin(Environment env, Consumer<StreamCreator> callback) {
+		Assert.notNull(env, "Environment cannot be null");
+		Assert.notNull(callback, "'callback' cannot be null");
+		this.streamCreator = env.streamCreator();
+		this.callback = callback;
+	}
+
+	@Override
+	public int getPhase() {
+		return this.phase;
+	}
+
+	/**
+	 * Set the phase; default is 0.
+	 * @param phase the phase.
+	 */
+	public void setPhase(int phase) {
+		this.phase = phase;
+	}
+
+	/**
+	 * Set to false to prevent automatic startup.
+	 * @param autoStartup the autoStartup.
+	 */
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	@Override
+	public void start() {
+		this.callback.accept(this.streamCreator);
+		this.running = true;
+	}
+
+	@Override
+	public void stop() {
+		this.running = false;
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.running;
+	}
+
+	@Override
+	public boolean isAutoStartup() {
+		return this.autoStartup;
+	}
+
+}

spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2022 the original author or authors.
+ * Copyright 2021-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -52,6 +52,7 @@ import org.springframework.http.MediaType;
 import org.springframework.rabbit.stream.config.StreamRabbitListenerContainerFactory;
 import org.springframework.rabbit.stream.producer.RabbitStreamTemplate;
 import org.springframework.rabbit.stream.retry.StreamRetryOperationsInterceptorFactoryBean;
+import org.springframework.rabbit.stream.support.StreamAdmin;
 import org.springframework.rabbit.stream.support.StreamMessageProperties;
 import org.springframework.retry.interceptor.RetryOperationsInterceptor;
 import org.springframework.test.annotation.DirtiesContext;
@@ -170,7 +171,17 @@ public class RabbitListenerTests extends AbstractTestContainerTests {
 		}
 
 		@Bean
-		SmartLifecycle creator(Environment env) {
+		StreamAdmin streamAdmin(Environment env) {
+			StreamAdmin streamAdmin = new StreamAdmin(env, sc -> {
+				sc.stream("test.stream.queue1").create();
+				sc.stream("test.stream.queue2").create();
+			});
+			streamAdmin.setAutoStartup(false);
+			return streamAdmin;
+		}
+
+		@Bean
+		SmartLifecycle creator(Environment env, StreamAdmin admin) {
 			return new SmartLifecycle() {
 
 				boolean running;
@@ -184,8 +195,7 @@ public class RabbitListenerTests extends AbstractTestContainerTests {
 				@Override
 				public void start() {
 					clean(env);
-					env.streamCreator().stream("test.stream.queue1").create();
-					env.streamCreator().stream("test.stream.queue2").create();
+					admin.start();
 					this.running = true;
 				}
 

src/reference/asciidoc/stream.adoc

@@ -28,8 +28,38 @@ compile 'org.springframework.amqp:spring-rabbit-stream:{project-version}'
 ----
 ====
 
-Provision the queues as normal, using a `RabbitAdmin` bean, using the `QueueBuilder.stream()` method to designate the queue type.
-See <<stream-examples>>.
+You can provision the queues as normal, using a `RabbitAdmin` bean, using the `QueueBuilder.stream()` method to designate the queue type.
+For example:
+
+====
+[source, java]
+----
+@Bean
+Queue stream() {
+    return QueueBuilder.durable("stream.queue1")
+            .stream()
+            .build();    
+}
+----
+====
+
+However, this will only work if you are also using non-stream components (such as the `SimpleMessageListenerContainer` or `DirectMessageListenerContainer`) because the admin is triggered to declare the defined beans when an AMQP connection is opened.
+If your application only uses stream components, or you wish to use advanced stream configuration features, you should configure a `StreamAdmin` instead:
+
+====
+[source, java]
+----
+@Bean
+StreamAdmin streamAdmin(Environment env) {
+    return new StreamAdmin(env, sc -> {
+        sc.stream("stream.queue1").maxAge(Duration.ofHours(2)).create();
+        sc.stream("stream.queue2").create();
+    });
+}
+----
+====
+
+Refer to the RabbitMQ documentation for more information about the `StreamCreator`.
 
 ==== Sending Messages