diff --git a/README.adoc b/README.adoc index 991ab0f4c..8bf85c9d5 100644 --- a/README.adoc +++ b/README.adoc @@ -116,7 +116,7 @@ A comma-separated list of RabbitMQ management plugin URLs. Only used when `nodes` contains more than one entry. Each entry in this list must have a corresponding entry in `spring.rabbitmq.addresses`. Only needed if you use a RabbitMQ cluster and wish to consume from the node that hosts the queue. -See https://docs.spring.io/spring-amqp/reference/html/_reference.html#queue-affinity[Queue Affinity and the LocalizedQueueConnectionFactory] for more information. +See https://docs.spring.io/spring-amqp/reference/html/#queue-affinity[Queue Affinity and the LocalizedQueueConnectionFactory] for more information. + Default: empty. spring.cloud.stream.rabbit.binder.nodes:: @@ -185,7 +185,6 @@ 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:: @@ -435,65 +434,6 @@ Not supported when the `containerType` is `direct`. + Default: `1`. -[[rabbitmq-stream-consumer]] -=== Initial Consumer 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((name, builder) -> { - builder.offset(OffsetSpecification.first()); - }); - // ... - }; -} ----- -==== - -The `name` argument passed to the customizer is `destination + '.' + group + '.container'`. - -The stream `name()` (for the purpose of offset tracking) is set to the binding `destination + '.' + group`. -It can be changed using a `ConsumerCustomizer` shown above. -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. @@ -1037,16 +977,16 @@ public class Application { if (correlation.getReturnedMessage() != null) { log.error("Message for " + correlation.getPayload() + " was returned "); - // throw some exception to invoke binder retry/error handling + // try to re-publish, send a DLQ, etc } } catch (InterruptedException e) { Thread.currentThread().interrupt(); - throw new IllegalStateException(e); + e.printStackTrace(); } catch (ExecutionException | TimeoutException e) { - throw new IllegalStateException(e); + e.printStackTrace(); } }); }; @@ -1104,7 +1044,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=myExchange` +* `spring.cloud.stream.bindings..destination=myExhange` * `spring.cloud.stream.bindings..group=myQueue` * `spring.cloud.stream.rabbit.bindings..consumer.bindQueue=false` * `spring.cloud.stream.rabbit.bindings..consumer.declareExchange=false` @@ -1232,41 +1172,6 @@ For negatively acknowledged confirmations, the payload is a `NackedAmqpMessageEx There is no automatic handling of these exceptions (such as sending to a <>). You can consume these exceptions with your own Spring Integration flow. -[[rabbitmq-stream-producer]] -=== Initial Producer 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 producer properties described above are not supported when you set the `producerType` property to `STREAM_SYNC` or `STREAM_ASYNC`. - -To configure the binder to use a stream `ProducerType`, you must add an `Environment` `@Bean` and, optionally, a customizer to customize the message handler. - -==== -[source, java] ----- -@Bean -Environment streamEnv() { - return Environment.builder() - .build(); -} - -@Bean -ProducerMessageHandlerCustomizer handlerCustomizer() { - return (hand, dest) -> { - RabbitStreamMessageHandler handler = (RabbitStreamMessageHandler) hand; - handler.setConfirmTimeout(5000); - ((RabbitStreamTemplate) handler.getStreamOperations()).setProducerCustomizer( - (name, builder) -> { - ... - }); - }; -} ----- -==== - -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 producer builder. - = Appendices [appendix] [[building]] @@ -1276,19 +1181,22 @@ Refer to the https://rabbitmq.github.io/rabbitmq-stream-java-client/stable/htmls === Basic Compile and Test -To build the source you will need to install JDK {jdkversion}. +Pre-requisites: + +* To compile, JDK {jdkversion} installed. +* To run tests, RabbitMQ server running on `localhost:5672` + The build uses the Maven wrapper so you don't have to install a specific -version of Maven. To enable the tests, you should have RabbitMQ server running -on localhost and the default port (5672) -before building. - -The main build command is +version of Maven. The main build command is ---- $ ./mvnw clean install ---- +NOTE: There are scripts in `./ci-docker-compose` that use https://docs.docker.com/compose//[Docker Compose] to +start/stop a local RabbitMQ server. + You can also add '-DskipTests' if you like, to avoid running the tests. NOTE: You can also install Maven (>=3.3.3) yourself and run the `mvn` command @@ -1304,11 +1212,6 @@ build succeed, please raise a ticket to get the settings added to source control. -The projects that require middleware generally include a -`docker-compose.yml`, so consider using -https://compose.docker.io/[Docker Compose] to run the middeware servers -in Docker containers. - === Documentation There is a "docs" profile that will generate documentation. diff --git a/docs/pom.xml b/docs/pom.xml index 4276bece9..716e65be1 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-M3 + 3.2.0-SNAPSHOT jar spring-cloud-stream-binder-rabbit-docs diff --git a/docs/src/main/asciidoc/_configprops.adoc b/docs/src/main/asciidoc/_configprops.adoc index 6b575783b..88ac8ca3b 100644 --- a/docs/src/main/asciidoc/_configprops.adoc +++ b/docs/src/main/asciidoc/_configprops.adoc @@ -9,6 +9,7 @@ |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' @@ -20,6 +21,11 @@ |spring.cloud.stream.metrics.schedule-interval | `60s` | Interval expressed as Duration for scheduling metrics snapshots publishing. Defaults to 60 seconds |spring.cloud.stream.override-cloud-connectors | `false` | This property is only applicable when the cloud profile is active and Spring Cloud Connectors are provided with the application. If the property is false (the default), the binder detects a suitable bound service (for example, a RabbitMQ service bound in Cloud Foundry for the RabbitMQ binder) and uses it for creating connections (usually through Spring Cloud Connectors). When set to true, this property instructs binders to completely ignore the bound services and rely on Spring Boot properties (for example, relying on the spring.rabbitmq.* properties provided in the environment for the RabbitMQ binder). The typical usage of this property is to be nested in a customized environment when connecting to multiple systems. |spring.cloud.stream.pollable-source | `none` | A semi-colon delimited list of binding names of pollable sources. Binding names follow the same naming convention as functions. For example, name '...pollable-source=foobar' will be accessible as 'foobar-iin-0'' binding +|spring.cloud.stream.poller.cron | | Cron expression value for the Cron Trigger. +|spring.cloud.stream.poller.fixed-delay | `1000` | Fixed delay for default poller. +|spring.cloud.stream.poller.initial-delay | `0` | Initial delay for periodic triggers. +|spring.cloud.stream.poller.max-messages-per-poll | `1` | Maximum messages per poll for the default poller. +|spring.cloud.stream.poller.time-unit | | The TimeUnit to apply to delay values. |spring.cloud.stream.rabbit.binder.admin-addresses | `[]` | Urls for management plugins; only needed for queue affinity. |spring.cloud.stream.rabbit.binder.admin-adresses | | |spring.cloud.stream.rabbit.binder.compression-level | `0` | Compression level for compressed bindings; see 'java.util.zip.Deflator'. diff --git a/pom.xml b/pom.xml index 34a4173f0..9a2916d7e 100644 --- a/pom.xml +++ b/pom.xml @@ -2,18 +2,18 @@ 4.0.0 spring-cloud-stream-binder-rabbit-parent - 3.2.0-M3 + 3.2.0-SNAPSHOT pom org.springframework.cloud spring-cloud-build - 3.1.0-M3 + 3.1.0-SNAPSHOT - 3.2.0-M3 + 3.2.0-SNAPSHOT 1.8 - 3.2.0-M3 + 3.2.0-SNAPSHOT true true diff --git a/spring-cloud-starter-stream-rabbit/pom.xml b/spring-cloud-starter-stream-rabbit/pom.xml index fe53de3f0..a8fe1082d 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-M3 + 3.2.0-SNAPSHOT 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 61be3837a..ee954e02a 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-M3 + 3.2.0-SNAPSHOT diff --git a/spring-cloud-stream-binder-rabbit-test-support/pom.xml b/spring-cloud-stream-binder-rabbit-test-support/pom.xml index 3e2ebb00f..ed10c0e8b 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-M3 + 3.2.0-SNAPSHOT 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 78652a5d4..134fd2b3b 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-M3 + 3.2.0-SNAPSHOT