diff --git a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/support/StreamAdmin.java b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/support/StreamAdmin.java new file mode 100644 index 00000000..dc6e336a --- /dev/null +++ b/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 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 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; + } + +} diff --git a/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java b/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java index 0ea7fc50..d15fbdb1 100644 --- a/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java +++ b/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; } diff --git a/src/reference/asciidoc/stream.adoc b/src/reference/asciidoc/stream.adoc index ca3e07a9..264e58e4 100644 --- a/src/reference/asciidoc/stream.adoc +++ b/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 <>. +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