From 4fbe76571f7656f33dfd028769f5593a3366a7cf Mon Sep 17 00:00:00 2001 From: buildmaster Date: Fri, 30 Jul 2021 14:58:30 +0000 Subject: [PATCH] Update SNAPSHOT to 3.2.0-M1 --- README.adoc | 66 +++++++++++++++++-- docs/pom.xml | 2 +- docs/src/main/asciidoc/_configprops.adoc | 1 - pom.xml | 6 +- spring-cloud-starter-stream-rabbit/pom.xml | 2 +- .../pom.xml | 2 +- .../pom.xml | 2 +- spring-cloud-stream-binder-rabbit/pom.xml | 2 +- 8 files changed, 69 insertions(+), 14 deletions(-) diff --git a/README.adoc b/README.adoc index e12abbfce..6c4369ad0 100644 --- a/README.adoc +++ b/README.adoc @@ -185,6 +185,7 @@ Default: none - the broker will generate random consumer tags. containerType:: Select the type of listener container to be used. See https://docs.spring.io/spring-amqp/reference/html/_reference.html#choose-container[Choosing a Container] in the Spring AMQP documentation for more information. +Also see <>. + Default: `simple` deadLetterQueueName:: @@ -434,6 +435,62 @@ Not supported when the `containerType` is `direct`. + Default: `1`. +[[rabbitmq-stream]] +=== Initial Support for the RabbitMQ Stream Plugin + +Basic support for the https://rabbitmq.com/stream.html[RabbitMQ Stream Plugin] is now provided. +To enable this feature, you must add the `spring-rabbit-stream` jar to the class path - it must be the same version as `spring-amqp` and `spring-rabbit`. + +IMPORTANT: The consumer properties described above are not supported when you set the `containerType` property to `stream`; `concurrency` is also not supported at this time. +Only a single stream queue can be consumed by each binding. + +To configure the binder to use `containerType=stream`, you must add an `Environment` `@Bean` and, optionally, a customizer to customize the listener container. + +==== +[source, java] +---- +@Bean +Environment streamEnv() { + return Environment.builder() + .build(); +} + +@Bean +ListenerContainerCustomizer customizer() { + return (cont, dest, group) -> { + StreamListenerContainer container = (StreamListenerContainer) cont; + container.setConsumerCustomizer(builder -> { + builder.offset(OffsetSpecification.first()); + }); + // ... + }; +} +---- +==== + +The stream name (for the purpose of offset tracking) is set to the binding `destination + '.' + group`. +If you decide to use manual offset tracking, the `Context` is available as a message header: + +==== +[source, java] +---- +int count; + +@Bean +public Consumer> input() { + return msg -> { + System.out.println(msg); + if (++count % 1000 == 0) { + Context context = msg.getHeaders().get("rabbitmq_streamContext", Context.class); + context.consumer().store(context.offset()); + } + }; +} +---- +==== + +Refer to the https://rabbitmq.github.io/rabbitmq-stream-java-client/stable/htmlsingle/[RabbitMQ Stream Java Client documentation] for information about configuring the environment and consumer builder. + === Advanced Listener Container Configuration To set listener container properties that are not exposed as binder or binding properties, add a single bean of type `ListenerContainerCustomizer` to the application context. @@ -977,18 +1034,17 @@ public class Application { if (correlation.getReturnedMessage() != null) { log.error("Message for " + correlation.getPayload() + " was returned "); - // try to re-publish, send a DLQ, etc + // throw some exception to invoke binder retry/error handling } } catch (InterruptedException e) { Thread.currentThread().interrupt(); - e.printStackTrace(); + throw new IllegalStateException(e); } catch (ExecutionException | TimeoutException e) { - e.printStackTrace(); + throw new IllegalStateException(e); } - }); }; } @@ -1044,7 +1100,7 @@ There are a number of rabbit-specific binding properties that allow you to modif If you have an existing exchange/queue that you wish to use, you can completely disable automatic provisioning as follows, assuming the exchange is named `myExchange` and the queue is named `myQueue`: -* `spring.cloud.stream.bindings..destination=myExhange` +* `spring.cloud.stream.bindings..destination=myExchange` * `spring.cloud.stream.bindings..group=myQueue` * `spring.cloud.stream.rabbit.bindings..consumer.bindQueue=false` * `spring.cloud.stream.rabbit.bindings..consumer.declareExchange=false` diff --git a/docs/pom.xml b/docs/pom.xml index 716e65be1..196e4e32c 100644 --- a/docs/pom.xml +++ b/docs/pom.xml @@ -7,7 +7,7 @@ org.springframework.cloud spring-cloud-stream-binder-rabbit-parent - 3.2.0-SNAPSHOT + 3.2.0-M1 jar spring-cloud-stream-binder-rabbit-docs diff --git a/docs/src/main/asciidoc/_configprops.adoc b/docs/src/main/asciidoc/_configprops.adoc index 88ac8ca3b..4a79f45ca 100644 --- a/docs/src/main/asciidoc/_configprops.adoc +++ b/docs/src/main/asciidoc/_configprops.adoc @@ -9,7 +9,6 @@ |spring.cloud.stream.dynamic-destinations | `[]` | A list of destinations that can be bound dynamically. If set, only listed destinations can be bound. |spring.cloud.stream.function.batch-mode | `false` | |spring.cloud.stream.function.bindings | | -|spring.cloud.stream.function.definition | | Definition of functions to bind. If several functions need to be composed into one, use pipes (e.g., 'fooFunc\|barFunc') |spring.cloud.stream.instance-count | `1` | The number of deployed instances of an application. Default: 1. NOTE: Could also be managed per individual binding "spring.cloud.stream.bindings.foo.consumer.instance-count" where 'foo' is the name of the binding. |spring.cloud.stream.instance-index | `0` | The instance id of the application: a number from 0 to instanceCount-1. Used for partitioning and with Kafka. NOTE: Could also be managed per individual binding "spring.cloud.stream.bindings.foo.consumer.instance-index" where 'foo' is the name of the binding. |spring.cloud.stream.instance-index-list | | A list of instance id's from 0 to instanceCount-1. Used for partitioning and with Kafka. NOTE: Could also be managed per individual binding "spring.cloud.stream.bindings.foo.consumer.instance-index-list" where 'foo' is the name of the binding. This setting will override the one set in 'spring.cloud.stream.instance-index' diff --git a/pom.xml b/pom.xml index 0a4f019d2..7fbbe2376 100644 --- a/pom.xml +++ b/pom.xml @@ -2,16 +2,16 @@ 4.0.0 spring-cloud-stream-binder-rabbit-parent - 3.2.0-SNAPSHOT + 3.2.0-M1 pom org.springframework.cloud spring-cloud-build - 3.1.0-SNAPSHOT + 3.1.0-M1 - 3.2.0-SNAPSHOT + 3.2.0-M1 1.8 true true diff --git a/spring-cloud-starter-stream-rabbit/pom.xml b/spring-cloud-starter-stream-rabbit/pom.xml index a8fe1082d..3b3abf1da 100644 --- a/spring-cloud-starter-stream-rabbit/pom.xml +++ b/spring-cloud-starter-stream-rabbit/pom.xml @@ -4,7 +4,7 @@ org.springframework.cloud spring-cloud-stream-binder-rabbit-parent - 3.2.0-SNAPSHOT + 3.2.0-M1 spring-cloud-starter-stream-rabbit Spring Cloud Starter Stream Rabbit diff --git a/spring-cloud-stream-binder-rabbit-core/pom.xml b/spring-cloud-stream-binder-rabbit-core/pom.xml index ee954e02a..51c54c8c0 100644 --- a/spring-cloud-stream-binder-rabbit-core/pom.xml +++ b/spring-cloud-stream-binder-rabbit-core/pom.xml @@ -10,7 +10,7 @@ org.springframework.cloud spring-cloud-stream-binder-rabbit-parent - 3.2.0-SNAPSHOT + 3.2.0-M1 diff --git a/spring-cloud-stream-binder-rabbit-test-support/pom.xml b/spring-cloud-stream-binder-rabbit-test-support/pom.xml index ed10c0e8b..861d4fe65 100644 --- a/spring-cloud-stream-binder-rabbit-test-support/pom.xml +++ b/spring-cloud-stream-binder-rabbit-test-support/pom.xml @@ -4,7 +4,7 @@ org.springframework.cloud spring-cloud-stream-binder-rabbit-parent - 3.2.0-SNAPSHOT + 3.2.0-M1 spring-cloud-stream-binder-rabbit-test-support Rabbit related test classes diff --git a/spring-cloud-stream-binder-rabbit/pom.xml b/spring-cloud-stream-binder-rabbit/pom.xml index 45875901a..8c55f1714 100644 --- a/spring-cloud-stream-binder-rabbit/pom.xml +++ b/spring-cloud-stream-binder-rabbit/pom.xml @@ -10,7 +10,7 @@ org.springframework.cloud spring-cloud-stream-binder-rabbit-parent - 3.2.0-SNAPSHOT + 3.2.0-M1