SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Update SNAPSHOT to 3.2.1

Browse Source
This commit is contained in:
buildmaster
2021-12-01 16:39:57 +00:00
parent a9656d34f8
commit 060a67a716
8 changed files with 118 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

125
README.adoc
View File

@@ -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/#queue-affinity[Queue Affinity and the LocalizedQueueConnectionFactory] for more information.
See https://docs.spring.io/spring-amqp/reference/html/_reference.html#queue-affinity[Queue Affinity and the LocalizedQueueConnectionFactory] for more information.
+
Default: empty.
spring.cloud.stream.rabbit.binder.nodes::
@@ -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 <<rabbitmq-stream>>.
+
Default: `simple`
deadLetterQueueName::
@@ -434,6 +435,65 @@ 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<MessageListenerContainer> 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<Message<?>> 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,16 +1037,16 @@ 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 +1104,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.<binding name>.destination=myExhange`
* `spring.cloud.stream.bindings.<binding name>.destination=myExchange`
* `spring.cloud.stream.bindings.<binding name>.group=myQueue`
* `spring.cloud.stream.rabbit.bindings.<binding name>.consumer.bindQueue=false`
* `spring.cloud.stream.rabbit.bindings.<binding name>.consumer.declareExchange=false`
@@ -1172,6 +1232,41 @@ For negatively acknowledged confirmations, the payload is a `NackedAmqpMessageEx
There is no automatic handling of these exceptions (such as sending to a <<rabbit-dlq-processing, dead-letter queue>>).
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<MessageHandler> 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]]
@@ -1181,22 +1276,19 @@ You can consume these exceptions with your own Spring Integration flow.
=== Basic Compile and Test
Pre-requisites:
* To compile, JDK {jdkversion} installed.
* To run tests, RabbitMQ server running on `localhost:5672`
To build the source you will need to install JDK {jdkversion}.
The build uses the Maven wrapper so you don't have to install a specific
version of Maven. The main build command is
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
----
$ ./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
@@ -1212,6 +1304,11 @@ 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.

2
docs/pom.xml
View File

@@ -7,7 +7,7 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rabbit-parent</artifactId>
<version>3.2.1-SNAPSHOT</version>
<version>3.2.1</version>
</parent>
<packaging>jar</packaging>
<name>spring-cloud-stream-binder-rabbit-docs</name>

6
docs/src/main/asciidoc/_configprops.adoc
View File

@@ -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'
@@ -21,11 +20,6 @@
|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'.

4
pom.xml
View File

@@ -2,7 +2,7 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-cloud-stream-binder-rabbit-parent</artifactId>
<version>3.2.1-SNAPSHOT</version>
<version>3.2.1</version>
<packaging>pom</packaging>
<parent>
<groupId>org.springframework.cloud</groupId>
@@ -11,7 +11,7 @@
<relativePath />
</parent>
<properties>
<spring-cloud-stream.version>3.2.1-SNAPSHOT</spring-cloud-stream.version>
<spring-cloud-stream.version>3.2.1</spring-cloud-stream.version>
<java.version>1.8</java.version>
<spring-cloud-function.version>3.2.1</spring-cloud-function.version>
<maven-checkstyle-plugin.failsOnError>true</maven-checkstyle-plugin.failsOnError>

2
spring-cloud-starter-stream-rabbit/pom.xml
View File

@@ -4,7 +4,7 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rabbit-parent</artifactId>
<version>3.2.1-SNAPSHOT</version>
<version>3.2.1</version>
</parent>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
<description>Spring Cloud Starter Stream Rabbit</description>

2
spring-cloud-stream-binder-rabbit-core/pom.xml
View File

@@ -10,7 +10,7 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rabbit-parent</artifactId>
<version>3.2.1-SNAPSHOT</version>
<version>3.2.1</version>
</parent>
<dependencies>

2
spring-cloud-stream-binder-rabbit-test-support/pom.xml
View File

@@ -4,7 +4,7 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rabbit-parent</artifactId>
<version>3.2.1-SNAPSHOT</version>
<version>3.2.1</version>
</parent>
<artifactId>spring-cloud-stream-binder-rabbit-test-support</artifactId>
<description>Rabbit related test classes</description>

2
spring-cloud-stream-binder-rabbit/pom.xml
View File

@@ -10,7 +10,7 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rabbit-parent</artifactId>
<version>3.2.1-SNAPSHOT</version>
<version>3.2.1</version>
</parent>
<dependencies>