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

Going back to snapshots

Browse Source
This commit is contained in:
buildmaster
2022-02-16 18:16:18 +00:00
parent 0e0999f5b1
commit 7bfe00519b
8 changed files with 30 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

144
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/_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 <<rabbitmq-stream>>.
+
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<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.
@@ -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.<binding name>.destination=myExchange`
* `spring.cloud.stream.bindings.<binding name>.destination=myExhange`
* `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`
@@ -1232,60 +1172,6 @@ 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.
=======
[[rabbit-binder-health-indicator]]
== Rabbit Binder Health Indicator
The health indicator for Rabbit binder delegates to the one provided from Spring Boot.
For more information on this, see https://docs.spring.io/spring-boot/docs/current/reference/htmlsingle/#actuator.endpoints.health.auto-configured-health-indicators[this].
You can disable this health indicator at the binder level by using the property - `management.health.binders.enabled` and set this to `false`.
In the case of multibinder environements, this has to be set on the binder's environment properties.
When the health indicator is disabled, you should see something like the below in the health actuator endpoint:
```
"rabbit": {
"status": "UNKNOWN"
}
```
At the Spring Boot level, if you want to disable the Rabbit health indicator, you need to use the property `management.health.rabbit.enabled` and set to `false`.
= Appendices
[appendix]
[[building]]
@@ -1295,19 +1181,22 @@ At the Spring Boot level, if you want to disable the Rabbit health indicator, yo
=== 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
@@ -1323,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.

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.2</version>
<version>3.2.2-SNAPSHOT</version>
</parent>
<packaging>jar</packaging>
<name>spring-cloud-stream-binder-rabbit-docs</name>

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

@@ -9,7 +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.input-bindings | | A semi-colon delimited string to explicitly define input bindings (specifically for cases when there is no implicit trigger to create such bindings such as Function, Supplier or Consumer).
|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'
@@ -19,9 +19,13 @@
|spring.cloud.stream.metrics.meter-filter | | Pattern to control the 'meters' one wants to capture. By default all 'meters' will be captured. For example, 'spring.integration.*' will only capture metric information for meters whose name starts with 'spring.integration'.
|spring.cloud.stream.metrics.properties | | Application properties that should be added to the metrics payload For example: `spring.application**`.
|spring.cloud.stream.metrics.schedule-interval | `60s` | Interval expressed as Duration for scheduling metrics snapshots publishing. Defaults to 60 seconds
|spring.cloud.stream.output-bindings | | A semi-colon delimited string to explicitly define output bindings (specifically for cases when there is no implicit trigger to create such bindings such as Function, Supplier or Consumer).
|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'.
@@ -29,6 +33,6 @@
|spring.cloud.stream.rabbit.binder.nodes | `[]` | Cluster member node names; only needed for queue affinity.
|spring.cloud.stream.rabbit.bindings | |
|spring.cloud.stream.sendto.destination | `none` | The name of the header used to determine the name of the output destination
|spring.cloud.stream.source | | A semi-colon delimited string representing the names of the sources based on which source bindings will be created. This is primarily to support cases where source binding may be required without providing a corresponding Supplier. (e.g., for cases where the actual source of data is outside of scope of spring-cloud-stream - HTTP -> Stream) @deprecated use {@link #outputBindings}
|spring.cloud.stream.source | | A colon delimited string representing the names of the sources based on which source bindings will be created. This is primarily to support cases where source binding may be required without providing a corresponding Supplier. (e.g., for cases where the actual source of data is outside of scope of spring-cloud-stream - HTTP -> Stream)
|===

8
pom.xml
View File

@@ -2,18 +2,18 @@
<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.2</version>
<version>3.2.2-SNAPSHOT</version>
<packaging>pom</packaging>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-build</artifactId>
<version>3.1.1</version>
<version>3.1.0</version>
<relativePath />
</parent>
<properties>
<spring-cloud-stream.version>3.2.2</spring-cloud-stream.version>
<spring-cloud-stream.version>3.2.2-SNAPSHOT</spring-cloud-stream.version>
<java.version>1.8</java.version>
<spring-cloud-function.version>3.2.2</spring-cloud-function.version>
<spring-cloud-function.version>3.2.2-SNAPSHOT</spring-cloud-function.version>
<maven-checkstyle-plugin.failsOnError>true</maven-checkstyle-plugin.failsOnError>
<maven-checkstyle-plugin.failsOnViolation>true
</maven-checkstyle-plugin.failsOnViolation>

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.2</version>
<version>3.2.2-SNAPSHOT</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.2</version>
<version>3.2.2-SNAPSHOT</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.2</version>
<version>3.2.2-SNAPSHOT</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.2</version>
<version>3.2.2-SNAPSHOT</version>
</parent>
<dependencies>