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

Merge branch '2.2.x'

Browse Source
This commit is contained in:
Oleg Zhurakousky
2019-02-04 17:05:48 +01:00
parent a8cbf77794 5b39e57ac9
commit 38bdccf965
27 changed files with 429 additions and 332 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/pom.xml
View File

@@ -7,7 +7,7 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-parent</artifactId>
<version>2.1.1.BUILD-SNAPSHOT</version>
<version>2.2.0.BUILD-SNAPSHOT</version>
</parent>
<packaging>pom</packaging>
<name>spring-cloud-stream-core-docs</name>

34
docs/src/main/asciidoc/preface.adoc
View File

@@ -151,8 +151,8 @@ You can also build and package your application into a boot jar (by using `./mvn
Now you have a working (albeit very basic) Spring Cloud Stream application.
== What's New in 2.1?
Spring Cloud Stream introduces a number of new features, enhancements, and changes in addition to the once already introduced in
== What's New in 2.2?
Spring Cloud Stream introduces a number of new features, enhancements, and changes in addition to the once already introduced in
https://docs.spring.io/spring-cloud-stream/docs/Elmhurst.SR2/reference/htmlsingle/#_what_s_new_in_2_0[version 2.0]
@@ -164,42 +164,18 @@ The following sections outline the most notable ones:
[[spring-cloud-stream-preface-new-features]]
=== New Features and Components
* *Spring Cloud Function*: One of the core themes of the 2.1.x release is the introduction of programming model based on https://cloud.spring.io/spring-cloud-function/[Spring Cloud Function] project.
For more details you can jump right into the <<spring_cloud_function, relevant section>>.
You can also read https://spring.io/blog/2018/10/30/spring-cloud-stream-fishtown-rc1-2-1-0-rc1-release-announcement[this blog post] for more details.
* *Simplified Test Binder*: In addition to an already existing testing support via `spring-cloud-stream-test-support`, this release also introduces a simpler implementation of a test binder that is
more aligned with the current binder API providing for a better integration testing as it is touches on all aspects of binding API.
This binder was primarily designed for internal use, but found its usages outside. For more information on how to use it and how it can help you please refer to <<spring_integration_test_binder, this section>> of user guide.
[[spring-cloud-stream-preface-notable-enhancements]]
=== Notable Enhancements
* *Improved Reactive Support*: Given that https://projectreactor.io/[Project Reactor] primitives such as `Flux` and `Mono` are at the core
of https://cloud.spring.io/spring-cloud-function/[Spring Cloud Function] project, you no longer
have to use or draw any distinction between _reactive_ and _conventional_ stream handler design,
hence you no longer need to explicitly rely on `spring-cloud-stream-reactive` module, which we're now
considering for deprecation. For more details please refer to <<spring_cloud_function, Spring Cloud Function>> section of this user guide.
* *Enhanced properties binding support*: This version of Spring Cloud Stream introduces significant
enhancements to configuration properties bindings primarily to ensure consistency between the default and binding specific properties.
A particular emphasis was given to maintaining the _precedence_ and _inheritance_ aspects where:
- _precedence_ - binding specific properties always take precedence over the default properties, effectively allowing binding specific properties to override the default ones
- _inheritance_ - default will propagate to individual binding properties unless explicitly overriden by the binding specific properties
* *Additional Content-Type Negotiation Improvements*: One of the core themes for 2.0.x release was an improved content-type negotiation.
This release introduces few more significant enhancements to introduce more consistency. One such enhancement is
the delegation of type conversion to MessageConverters in _all_ cases, including the ones where the target type of the handler method is not known.
To you (the end user) it simply means that starting with this release extending content-type negotiation via `@StreamMessageConverter` is available for all type conversion cases.
NOTE: Keep in mind that most of the content-type work at the moment also preserves compatibility with 1.3.x version of Spring Cloud Stream, thus will be further simplified once 1.3.x line goes EOL.
[[spring-cloud-stream-preface-notable-deprecations]]
=== Notable Deprecations
As of version 2.1, the following items have been deprecated:
As of version 2.2, the following items have been deprecated:
- Aggregator Builder support is deprecated in favor of application composition via <<spring_cloud_function,Spring Cloud Function>> programming model.
- As mentioned earlier we're also considering the deprecation of `spring-cloud-stream-reactive` module in favor of the adequate support already provided by <<spring_cloud_function,Spring Cloud Function>>.
- The spring-cloud-stream-reactive module is deprecated in favor of native support
via <<spring_cloud_function,Spring Cloud Function>> programming model.
== Notes on migrating from 1.x to 2.x?
- Due to the improvements in content-type negotiation, the `originalContentType` header is not used (ignored) since 2.x and only exists for maintaining compatibility with 1.x versions

189
docs/src/main/asciidoc/spring-cloud-stream.adoc
View File

@@ -659,7 +659,27 @@ public static class SinkFromConsumer {
}
}
----
===== Reactive Functions support
Since _Spring Cloud Function_ is build on top of https://projectreactor.io/[Project Reactor] there isn't much you need to do
to benefit from reactive programming model while implementing `Supplier`, `Function` or `Consumer`.
For example:
[source,java]
----
@EnableAutoConfiguration
@EnableBinding(Processor.class)
public static class SinkFromConsumer {
public static void main(String[] args) {
SpringApplication.run(SinkFromConsumer.class, "--spring.cloud.stream.function.definition=reactiveUpperCase");
}
@Bean
public Function<Flux<String>, Flux<String>> reactiveUpperCase() {
return flux -> flux.map(val -> val.toUpperCase());
}
}
----
===== Functional Composition
Using this programming model you can also benefit from functional composition where you can dynamically compose complex handlers from a set of simple functions.
@@ -676,7 +696,17 @@ To do that Spring Cloud Function allows you to use `|` (pipe) symbol. So to fini
[source,java]
----
—spring.cloud.stream.function.definition=toUpperCase|wrapInQuotes
--spring.cloud.stream.function.definition=toUpperCase|wrapInQuotes
----
NOTE: One of the great benefits of functional composition support provided by _Spring Cloud Function_ is
the fact that you can compose _reactive_ and _imperative_ functions.
For example, the above composition could be defined as such (if both functions present):
[source,java]
----
--spring.cloud.stream.function.definition=reactiveUpperCase|wrapInQuotes
----
@@ -1024,161 +1054,14 @@ public RetryTemplate myRetryTemplate() {
----
As you can see from the above example you don't need to annotate it with `@Bean` since `@StreamRetryTemplate` is a qualified `@Bean`.
[[spring-cloud-stream-overview-reactive-programming-support]]
=== Reactive Programming Support
If you need to be more precise with your `RetryTemplate`, you can specify the bean by name in your `ConsumerProperties` to associate
the specific retry bean per binding.
Spring Cloud Stream also supports the use of reactive APIs where incoming and outgoing data is handled as continuous data flows.
Support for reactive APIs is available through `spring-cloud-stream-reactive`, which needs to be added explicitly to your project.
The programming model with reactive APIs is declarative. Instead of specifying how each individual message should be handled, you can use operators that describe functional transformations from inbound to outbound data flows.
At present Spring Cloud Stream supports the only the https://projectreactor.io/[Reactor API].
In the future, we intend to support a more generic model based on Reactive Streams.
The reactive programming model also uses the `@StreamListener` annotation for setting up reactive handlers.
The differences are that:
* The `@StreamListener` annotation must not specify an input or output, as they are provided as arguments and return values from the method.
* The arguments of the method must be annotated with `@Input` and `@Output`, indicating which input or output the incoming and outgoing data flows connect to, respectively.
* The return value of the method, if any, is annotated with `@Output`, indicating the input where data should be sent.
NOTE: Reactive programming support requires Java 1.8.
NOTE: As of Spring Cloud Stream 1.1.1 and later (starting with release train Brooklyn.SR2), reactive programming support requires the use of Reactor 3.0.4.RELEASE and higher.
Earlier Reactor versions (including 3.0.1.RELEASE, 3.0.2.RELEASE and 3.0.3.RELEASE) are not supported.
`spring-cloud-stream-reactive` transitively retrieves the proper version, but it is possible for the project structure to manage the version of the `io.projectreactor:reactor-core` to an earlier release, especially when using Maven.
This is the case for projects generated by using Spring Initializr with Spring Boot 1.x, which overrides the Reactor version to `2.0.8.RELEASE`.
In such cases, you must ensure that the proper version of the artifact is released.
You can do so by adding a direct dependency on `io.projectreactor:reactor-core` with a version of `3.0.4.RELEASE` or later to your project.
NOTE: The use of term, "`reactive`", currently refers to the reactive APIs being used and not to the execution model being reactive (that is, the bound endpoints still use a 'push' rather than a 'pull' model). While some backpressure support is provided by the use of Reactor, we do intend, in a future release, to support entirely reactive pipelines by the use of native reactive clients for the connected middleware.
===== Reactor-based Handlers
A Reactor-based handler can have the following argument types:
* For arguments annotated with `@Input`, it supports the Reactor `Flux` type.
The parameterization of the inbound Flux follows the same rules as in the case of individual message handling: It can be the entire `Message`, a POJO that can be the `Message` payload, or a POJO that is the result of a transformation based on the `Message` content-type header. Multiple inputs are provided.
* For arguments annotated with `Output`, it supports the `FluxSender` type, which connects a `Flux` produced by the method with an output. Generally speaking, specifying outputs as arguments is only recommended when the method can have multiple outputs.
A Reactor-based handler supports a return type of `Flux`. In that case, it must be annotated with `@Output`. We recommend using the return value of the method when a single output `Flux` is available.
The following example shows a Reactor-based `Processor`:
[source, java]
[source]
----
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class UppercaseTransformer {
@StreamListener
@Output(Processor.OUTPUT)
public Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
return input.map(s -> s.toUpperCase());
}
}
spring.cloud.stream.bindings.<foo>.consumer.retry-template-name=<your-retry-template-bean-name>
----
The same processor using output arguments looks like the following example:
[source, java]
----
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class UppercaseTransformer {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<String> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(s -> s.toUpperCase()));
}
}
----
===== Reactive Sources
Spring Cloud Stream reactive support also provides the ability for creating reactive sources through the `@StreamEmitter` annotation.
By using the `@StreamEmitter` annotation, a regular source may be converted to a reactive one.
`@StreamEmitter` is a method level annotation that marks a method to be an emitter to outputs declared with `@EnableBinding`.
You cannot use the `@Input` annotation along with `@StreamEmitter`, as the methods marked with this annotation are not listening for any input. Rather, methods marked with `@StreamEmitter` generate output.
Following the same programming model used in `@StreamListener`, `@StreamEmitter` also allows flexible ways of using the `@Output` annotation, depending on whether the method has any arguments, a return type, and other considerations.
The remainder of this section contains examples of using the `@StreamEmitter` annotation in various styles.
The following example emits the `Hello, World` message every millisecond and publishes to a Reactor `Flux`:
[source, java]
----
@EnableBinding(Source.class)
@EnableAutoConfiguration
public static class HelloWorldEmitter {
@StreamEmitter
@Output(Source.OUTPUT)
public Flux<String> emit() {
return Flux.intervalMillis(1)
.map(l -> "Hello World");
}
}
----
In the preceding example, the resulting messages in the `Flux` are sent to the output channel of the `Source`.
The next example is another flavor of an `@StreamEmmitter` that sends a Reactor `Flux`.
Instead of returning a `Flux`, the following method uses a `FluxSender` to programmatically send a `Flux` from a source:
[source, java]
----
@EnableBinding(Source.class)
@EnableAutoConfiguration
public static class HelloWorldEmitter {
@StreamEmitter
@Output(Source.OUTPUT)
public void emit(FluxSender output) {
output.send(Flux.intervalMillis(1)
.map(l -> "Hello World"));
}
}
----
The next example is exactly same as the above snippet in functionality and style.
However, instead of using an explicit `@Output` annotation on the method, it uses the annotation on the method parameter.
[source, java]
----
@EnableBinding(Source.class)
@EnableAutoConfiguration
public static class HelloWorldEmitter {
@StreamEmitter
public void emit(@Output(Source.OUTPUT) FluxSender output) {
output.send(Flux.intervalMillis(1)
.map(l -> "Hello World"));
}
}
----
The last example in this section is yet another flavor of writing reacting sources by using the Reactive Streams Publisher API and taking advantage of the support for it in https://github.com/spring-projects/spring-integration-java-dsl/wiki/Spring-Integration-Java-DSL-Reference[Spring Integration Java DSL].
The `Publisher` in the following example still uses Reactor `Flux` under the hood, but, from an application perspective, that is transparent to the user and only needs Reactive Streams and Java DSL for Spring Integration:
[source, java]
----
@EnableBinding(Source.class)
@EnableAutoConfiguration
public static class HelloWorldEmitter {
@StreamEmitter
@Output(Source.OUTPUT)
@Bean
public Publisher<Message<String>> emit() {
return IntegrationFlows.from(() ->
new GenericMessage<>("Hello World"),
e -> e.poller(p -> p.fixedDelay(1)))
.toReactivePublisher();
}
}
----
[[spring-cloud-stream-overview-binders]]
== Binders
@@ -1392,7 +1275,7 @@ To visualize the current bindings, access the following URL:
`http://<host>:<port>/actuator/bindings`
Alternative, to see a single binding, access one of the URLs similar to the following:
`http://<host>:<port>/actuator/bindings/myBindingName`
`http://<host>:<port>/actuator/bindings/<bindingName>`
You can also stop, start, pause, and resume individual bindings by posting to the same URL while providing a `state` argument as JSON, as shown in the following examples:

2
pom.xml
View File

@@ -4,7 +4,7 @@
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-cloud-stream-parent</artifactId>
<version>2.1.1.BUILD-SNAPSHOT</version>
<version>2.2.0.BUILD-SNAPSHOT</version>
<packaging>pom</packaging>
<parent>
<groupId>org.springframework.cloud</groupId>

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

@@ -13,7 +13,7 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-parent</artifactId>
<version>2.1.1.BUILD-SNAPSHOT</version>
<version>2.2.0.BUILD-SNAPSHOT</version>
</parent>
<dependencies>

2
spring-cloud-stream-integration-tests/pom.xml
View File

@@ -12,7 +12,7 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-parent</artifactId>
<version>2.1.1.BUILD-SNAPSHOT</version>
<version>2.2.0.BUILD-SNAPSHOT</version>
</parent>
<dependencies>

2
spring-cloud-stream-reactive/pom.xml
View File

@@ -5,7 +5,7 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-parent</artifactId>
<version>2.1.1.BUILD-SNAPSHOT</version>
<version>2.2.0.BUILD-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>

2
spring-cloud-stream-schema-server/pom.xml
View File

@@ -8,7 +8,7 @@
<parent>
<artifactId>spring-cloud-stream-parent</artifactId>
<groupId>org.springframework.cloud</groupId>
<version>2.1.1.BUILD-SNAPSHOT</version>
<version>2.2.0.BUILD-SNAPSHOT</version>
</parent>
<dependencies>

2
spring-cloud-stream-schema/pom.xml
View File

@@ -5,7 +5,7 @@
<parent>
<artifactId>spring-cloud-stream-parent</artifactId>
<groupId>org.springframework.cloud</groupId>
<version>2.1.1.BUILD-SNAPSHOT</version>
<version>2.2.0.BUILD-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>

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

@@ -6,7 +6,7 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-parent</artifactId>
<version>2.1.1.BUILD-SNAPSHOT</version>
<version>2.2.0.BUILD-SNAPSHOT</version>
</parent>
<artifactId>spring-cloud-stream-test-support-internal</artifactId>
<description>Set of classes and utility code that may assist in testing both

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

@@ -6,7 +6,7 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-parent</artifactId>
<version>2.1.1.BUILD-SNAPSHOT</version>
<version>2.2.0.BUILD-SNAPSHOT</version>
</parent>
<artifactId>spring-cloud-stream-test-support</artifactId>
<description>A set of classes to ease testing of Spring Cloud Stream modules.

84
spring-cloud-stream-tools/pom.xml Normal file
View File

@@ -0,0 +1,84 @@
<?xml version="1.0" encoding="UTF-8"?>
<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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<version>2.2.0.BUILD-SNAPSHOT</version>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-build</artifactId>
<version>2.1.2.BUILD-SNAPSHOT</version>
<relativePath />
</parent>
<artifactId>spring-cloud-stream-tools</artifactId>
<name>spring-cloud-stream-build-tools</name>
<description>Spring Cloud Stream Build Tools</description>
<properties>
<java.version>1.8</java.version>
</properties>
<profiles>
<profile>
<id>spring</id>
<repositories>
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</repository>
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
<repository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/release</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
</repositories>
<pluginRepositories>
<pluginRepository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</pluginRepository>
<pluginRepository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
<pluginRepository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/libs-release-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
</pluginRepositories>
</profile>
</profiles>
</project>

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

@@ -12,7 +12,7 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-parent</artifactId>
<version>2.1.1.BUILD-SNAPSHOT</version>
<version>2.2.0.BUILD-SNAPSHOT</version>
</parent>
<dependencies>

2
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/StreamMessageConverter.java
View File

@@ -23,6 +23,7 @@ import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
/**
* Marker to tag {@link org.springframework.messaging.converter.MessageConverter} beans
@@ -36,6 +37,7 @@ import org.springframework.beans.factory.annotation.Qualifier;
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Qualifier
@Bean
public @interface StreamMessageConverter {
}

19
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2018 the original author or authors.
* Copyright 2013-2019 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.
@@ -16,6 +16,8 @@
package org.springframework.cloud.stream.binder;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -52,6 +54,7 @@ import org.springframework.util.StringUtils;
* @author Soby Chacko
* @author Vinicius Carvalho
* @author Oleg Zhurakousky
* @author Nicolas Homble
*/
public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends ProducerProperties>
implements ApplicationContextAware, InitializingBean, Binder<T, C, P> {
@@ -68,10 +71,9 @@ public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends
private volatile EvaluationContext evaluationContext;
@Autowired(required = false) // this would need to be refactored into constructor in
// the future
@Autowired(required = false)
@StreamRetryTemplate
private RetryTemplate consumerBindingRetryTemplate;
private Map<String, RetryTemplate> consumerBindingRetryTemplates;
/**
* For binder implementations that support a prefix, apply the prefix to the name.
@@ -196,8 +198,8 @@ public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends
* @return The retry template
*/
protected RetryTemplate buildRetryTemplate(ConsumerProperties properties) {
RetryTemplate rt = this.consumerBindingRetryTemplate;
if (rt == null) {
RetryTemplate rt;
if (CollectionUtils.isEmpty(this.consumerBindingRetryTemplates)) {
rt = new RetryTemplate();
SimpleRetryPolicy retryPolicy = CollectionUtils
.isEmpty(properties.getRetryableExceptions())
@@ -213,6 +215,11 @@ public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends
rt.setRetryPolicy(retryPolicy);
rt.setBackOffPolicy(backOffPolicy);
}
else {
rt = StringUtils.hasText(properties.getRetryTemplateName())
? this.consumerBindingRetryTemplates.get(properties.getRetryTemplateName())
: this.consumerBindingRetryTemplates.values().iterator().next();
}
return rt;
}

92
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
View File

@@ -42,6 +42,7 @@ import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.context.Lifecycle;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.AbstractSubscribableChannel;
import org.springframework.integration.channel.DirectChannel;
@@ -199,6 +200,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
"Exception thrown while building outbound endpoint", e);
}
}
if (producerProperties.isAutoStartup()
&& producerMessageHandler instanceof Lifecycle) {
((Lifecycle) producerMessageHandler).start();
@@ -541,10 +543,9 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
*/
private SubscribableChannel registerErrorInfrastructure(
ProducerDestination destination) {
ConfigurableListableBeanFactory beanFactory = getApplicationContext()
.getBeanFactory();
String errorChannelName = errorsBaseName(destination);
SubscribableChannel errorChannel = null;
SubscribableChannel errorChannel;
if (getApplicationContext().containsBean(errorChannelName)) {
Object errorChannelObject = getApplicationContext().getBean(errorChannelName);
if (!(errorChannelObject instanceof SubscribableChannel)) {
@@ -555,9 +556,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
}
else {
errorChannel = new PublishSubscribeChannel();
this.registerComponentWithBeanFactory(errorChannelName, errorChannel);
errorChannel = (PublishSubscribeChannel) beanFactory
.initializeBean(errorChannel, errorChannelName);
((GenericApplicationContext)getApplicationContext()).
registerBean(errorChannelName, SubscribableChannel.class, () -> errorChannel);
}
MessageChannel defaultErrorChannel = null;
if (getApplicationContext()
@@ -571,8 +571,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
errorBridge.setOutputChannel(defaultErrorChannel);
errorChannel.subscribe(errorBridge);
String errorBridgeHandlerName = getErrorBridgeName(destination);
this.registerComponentWithBeanFactory(errorBridgeHandlerName, errorBridge);
beanFactory.initializeBean(errorBridge, errorBridgeHandlerName);
((GenericApplicationContext)getApplicationContext()).registerBean(errorBridgeHandlerName, BridgeHandler.class, () -> errorBridge);
}
return errorChannel;
}
@@ -607,23 +606,21 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
boolean polled) {
ErrorMessageStrategy errorMessageStrategy = getErrorMessageStrategy();
ConfigurableListableBeanFactory beanFactory = getApplicationContext()
.getBeanFactory();
String errorChannelName = errorsBaseName(destination, group, consumerProperties);
SubscribableChannel errorChannel = null;
SubscribableChannel errorChannel;
if (getApplicationContext().containsBean(errorChannelName)) {
Object errorChannelObject = getApplicationContext().getBean(errorChannelName);
if (!(errorChannelObject instanceof SubscribableChannel)) {
throw new IllegalStateException("Error channel '" + errorChannelName
+ "' must be a SubscribableChannel");
}
Assert.isInstanceOf(SubscribableChannel.class, errorChannelObject,
"Error channel '" + errorChannelName + "' must be a SubscribableChannel");
errorChannel = (SubscribableChannel) errorChannelObject;
}
else {
errorChannel = new BinderErrorChannel();
this.registerComponentWithBeanFactory(errorChannelName, errorChannel);
errorChannel = (LastSubscriberAwareChannel) beanFactory
.initializeBean(errorChannel, errorChannelName);
((GenericApplicationContext)getApplicationContext())
.registerBean(errorChannelName, SubscribableChannel.class, () -> errorChannel);
}
ErrorMessageSendingRecoverer recoverer;
if (errorMessageStrategy == null) {
@@ -633,10 +630,10 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
recoverer = new ErrorMessageSendingRecoverer(errorChannel,
errorMessageStrategy);
}
String recovererBeanName = getErrorRecovererName(destination, group,
consumerProperties);
this.registerComponentWithBeanFactory(recovererBeanName, recoverer);
beanFactory.initializeBean(recoverer, recovererBeanName);
String recovererBeanName = getErrorRecovererName(destination, group, consumerProperties);
((GenericApplicationContext)getApplicationContext())
.registerBean(recovererBeanName, ErrorMessageSendingRecoverer.class, () -> recoverer);
MessageHandler handler;
if (polled) {
handler = getPolledConsumerErrorMessageHandler(destination, group,
@@ -662,8 +659,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
if (handler != null) {
if (this.isSubscribable(errorChannel)) {
this.registerComponentWithBeanFactory(errorMessageHandlerName, handler);
beanFactory.initializeBean(handler, errorMessageHandlerName);
MessageHandler errorHandler = handler;
((GenericApplicationContext)getApplicationContext()).registerBean(errorMessageHandlerName, MessageHandler.class, () -> errorHandler);
errorChannel.subscribe(handler);
}
else {
@@ -680,11 +677,10 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
BridgeHandler errorBridge = new BridgeHandler();
errorBridge.setOutputChannel(defaultErrorChannel);
errorChannel.subscribe(errorBridge);
String errorBridgeHandlerName = getErrorBridgeName(destination, group,
consumerProperties);
this.registerComponentWithBeanFactory(errorBridgeHandlerName,
errorBridge);
beanFactory.initializeBean(errorBridge, errorBridgeHandlerName);
String errorBridgeHandlerName = getErrorBridgeName(destination, group, consumerProperties);
((GenericApplicationContext)getApplicationContext())
.registerBean(errorBridgeHandlerName, BridgeHandler.class, () -> errorBridge);
}
else {
this.logger.warn("The provided errorChannel '" + errorChannelName
@@ -731,10 +727,9 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
C properties) {
try {
String recoverer = getErrorRecovererName(destination, group, properties);
if (getApplicationContext().containsBean(recoverer)) {
((DefaultSingletonBeanRegistry) getApplicationContext().getBeanFactory())
.destroySingleton(recoverer);
}
destroyBean(recoverer);
String errorChannelName = errorsBaseName(destination, group, properties);
String errorMessageHandlerName = getErrorMessageHandlerName(destination,
group, properties);
@@ -755,16 +750,13 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
.getBean(errorChannelName, SubscribableChannel.class);
if (bridgeHandler != null) {
channel.unsubscribe(bridgeHandler);
((DefaultSingletonBeanRegistry) getApplicationContext()
.getBeanFactory()).destroySingleton(errorBridgeHandlerName);
destroyBean(errorBridgeHandlerName);
}
if (handler != null) {
channel.unsubscribe(handler);
((DefaultSingletonBeanRegistry) getApplicationContext()
.getBeanFactory()).destroySingleton(errorMessageHandlerName);
destroyBean(errorMessageHandlerName);
}
((DefaultSingletonBeanRegistry) getApplicationContext().getBeanFactory())
.destroySingleton(errorChannelName);
destroyBean(errorChannelName);
}
}
catch (IllegalStateException e) {
@@ -772,6 +764,13 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
}
}
private void destroyBean(String beanName) {
if (getApplicationContext().containsBean(beanName)) {
((DefaultSingletonBeanRegistry) getApplicationContext().getBeanFactory()).destroySingleton(beanName);
((GenericApplicationContext)getApplicationContext()).removeBeanDefinition(beanName);
}
}
/**
* Binders can return a message handler to be subscribed to the error channel.
* Examples might be if the user wishes to (re)publish messages to a DLQ.
@@ -865,20 +864,6 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
}
}
private void registerComponentWithBeanFactory(String name, Object component) {
if (getApplicationContext().getBeanFactory().containsBean(name)) {
throw new IllegalStateException("Failed to register bean with name '" + name
+ "', since bean with the same name already exists. Possible reason: "
+ "You may have multiple bindings with the same 'destination' and 'group' name (consumer side) "
+ "and multiple bindings with the same 'destination' name (producer side). "
+ "Solution: ensure each binding uses different group name (consumer side) "
+ "or 'destination' name (producer side).");
}
else {
getApplicationContext().getBeanFactory().registerSingleton(name, component);
}
}
/*
* FUNCTION-TO-EXISTING-APP section
*
@@ -938,6 +923,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
moveChannelInterceptors((AbstractMessageChannel) inputChannel,
actualInputChannel);
}
this.integrationFlowFunctionSupport.andThenFunction(
MessageChannelReactiveUtils.toPublisher(actualInputChannel),
inputChannel, this.streamFunctionProperties);

17
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java
View File

@@ -94,13 +94,25 @@ public interface Binding<T> extends Pausable {
}
/**
* Returns the name of this binding (i.e., channel name).
* @return binding name
* Returns the name of the destination for this binding
*
* @return destination name
*/
default String getName() {
return null;
}
/**
* Returns the name of the target for this binding (i.e., channel name)
*
* @return binding name
*
* @since 2.2
*/
default String getBindingName() {
return null;
}
/**
* Unbinds the target component represented by this instance and stops any active
* components. Implementations must be idempotent. After this method is invoked, the
@@ -120,5 +132,4 @@ public interface Binding<T> extends Pausable {
"Binding implementation `" + this.getClass().getName()
+ "` must implement this operation before it is called");
}
}

35
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2018 the original author or authors.
* Copyright 2016-2019 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.
@@ -24,13 +24,14 @@ import javax.validation.constraints.Min;
import com.fasterxml.jackson.annotation.JsonInclude;
/**
* Common consumer properties.
* Common consumer properties - spring.cloud.stream.bindings.[destinationName].consumer.
*
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Gary Russell
* @author Soby Chacko
* @author Oleg Zhurakousky
* @author Nicolas Homble
*/
@JsonInclude(JsonInclude.Include.NON_DEFAULT)
public class ConsumerProperties {
@@ -71,11 +72,22 @@ public class ConsumerProperties {
private int instanceIndex = -1;
/**
<<<<<<< HEAD
* The number of attempts to process the message (including the first) in the event of
* processing failures. This is a RetryTemplate configuration which is provided by the
* framework. Default: 3. Set to 1 to disable retry. You can also provide custom
* RetryTemplate in the event you want to take complete control of the RetryTemplate.
* Simply configure it as @Bean inside your application configuration.
=======
* The number of attempts to process the message (including the first)
* in the event of processing failures. This is a RetryTemplate configuration
* which is provided by the framework.
* Default: 3. Set to 1 to disable retry. You can also provide custom RetryTemplate
* in the event you want to take complete control of the RetryTemplate. Simply configure
* it as @Bean inside your application configuration.
* If you need to be binding specific, then you can reference a specific RetryTemplate by name
* with the retry-template configuration.
>>>>>>> 2.2.x
*/
private int maxAttempts = 3;
@@ -110,8 +122,18 @@ public class ConsumerProperties {
private boolean defaultRetryable = true;
/**
<<<<<<< HEAD
* A map of Throwable class names in the key and a boolean in the value. Specify those
* exceptions (and subclasses) that will or won't be retried.
=======
* Allows you to further qualify which RetryTemplate to use for a specific consumer binding..
*/
private String retryTemplateName;
/**
* A map of Throwable class names in the key and a boolean in the value.
* Specify those exceptions (and subclasses) that will or won't be retried.
>>>>>>> 2.2.x
*/
private Map<Class<? extends Throwable>, Boolean> retryableExceptions = new LinkedHashMap<>();
@@ -150,6 +172,14 @@ public class ConsumerProperties {
*/
private boolean multiplex;
public String getRetryTemplateName(){
return retryTemplateName;
}
public void setRetryTemplateName(String retryTemplateName){
this.retryTemplateName = retryTemplateName;
}
@Min(value = 1, message = "Concurrency should be greater than zero.")
public int getConcurrency() {
return this.concurrency;
@@ -269,5 +299,4 @@ public class ConsumerProperties {
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
}
}

5
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java
View File

@@ -287,8 +287,9 @@ public class DefaultBinderFactory
binderEnvironment.getPropertySources().remove("configurationProperties");
springApplicationBuilder.environment(binderEnvironment);
}
ConfigurableApplicationContext binderProducingContext = springApplicationBuilder
.run(args.toArray(new String[0]));
ConfigurableApplicationContext binderProducingContext =
springApplicationBuilder.run(args.toArray(new String[0]));
Binder<T, ?, ?> binder = binderProducingContext.getBean(Binder.class);
/*

12
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2018 the original author or authors.
* Copyright 2013-2019 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.
@@ -22,6 +22,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.context.Lifecycle;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.endpoint.Pausable;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.util.Assert;
@@ -39,7 +40,8 @@ import org.springframework.util.StringUtils;
* @author Oleg Zhurakousky
* @see org.springframework.cloud.stream.annotation.EnableBinding
*/
@JsonPropertyOrder({ "name", "group", "pausable", "state" })
@JsonPropertyOrder({"bindingName", "name", "group", "pausable", "state"})
@JsonIgnoreProperties("running")
public class DefaultBinding<T> implements Binding<T> {
@@ -84,6 +86,12 @@ public class DefaultBinding<T> implements Binding<T> {
return this.name;
}
public String getBindingName() {
String resolvedName = (this.target instanceof IntegrationObjectSupport)
? ((IntegrationObjectSupport)this.target).getComponentName() : getName();
return resolvedName == null ? getName() : resolvedName;
}
public String getGroup() {
return this.group;
}

1
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java
View File

@@ -43,7 +43,6 @@ public class ProducerProperties {
/**
* Signals if this producer needs to be started automatically.
*
* Default: true
*/
private boolean autoStartup = true;

1
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
View File

@@ -352,7 +352,6 @@ public class MessageConverterConfigurer
@Override
public Message<?> doPreSend(Message<?> message, MessageChannel channel) {
// If handler is a function, FunctionInvoker will already perform message
// conversion.
// In fact in the future we should consider propagating knowledge of the

2
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java
View File

@@ -189,6 +189,7 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
MimeType.valueOf(this.bindingServiceProperties
.getBindingProperties("output")
.getContentType()));
}
Assert.notNull(returnMessage,
"Failed to convert result value '" + value + "' to message.");
@@ -196,6 +197,7 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
return returnMessage;
}
@SuppressWarnings("unchecked")
private <T> Message<O> wrapOutputToMessage(T value, Message<I> originalMessage) {
Message<O> returnMessage = (Message<O>) MessageBuilder.withPayload(value)

1
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java
View File

@@ -403,7 +403,6 @@ public class PollableConsumerTests {
count.incrementAndGet();
throw new RequeueCurrentMessageException("test retry");
})).isTrue();
// fail("Expected exception"); // see GH-1578
}
catch (Exception e) {
// no op

119
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/RetryTemplateTests.java Normal file
View File

@@ -0,0 +1,119 @@
/*
* Copyright 2019 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
*
* http://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.cloud.stream.config;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.Map;
import org.junit.Test;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamRetryTemplate;
import org.springframework.cloud.stream.binder.AbstractBinder;
import org.springframework.cloud.stream.binder.ConsumerProperties;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.retry.support.RetryTemplate;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
/**
*
* @author Oleg Zhurakousky
*
*/
public class RetryTemplateTests {
@SuppressWarnings("rawtypes")
@Test
public void testSingleCustomRetryTemplate() throws Exception {
ApplicationContext context = new SpringApplicationBuilder(SingleCustomRetryTemplateConfiguration.class)
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false");
AbstractBinder binder = context.getBean(AbstractBinder.class);
Field f = AbstractBinder.class.getDeclaredField("consumerBindingRetryTemplates");
f.setAccessible(true);
@SuppressWarnings("unchecked")
Map<String, RetryTemplate> consumerBindingRetryTemplates = (Map<String, RetryTemplate>) f.get(binder);
assertTrue(consumerBindingRetryTemplates.size() == 1);
}
@EnableBinding(Processor.class)
@Import(TestChannelBinderConfiguration.class)
@EnableAutoConfiguration
public static class SingleCustomRetryTemplateConfiguration {
@StreamRetryTemplate
public RetryTemplate retryTemplate() {
return new RetryTemplate();
}
@Bean
public RetryTemplate otherRetryTemplate() {
return new RetryTemplate();
}
}
@SuppressWarnings("rawtypes")
@Test
public void testSpecificCustomRetryTemplate() throws Exception {
ApplicationContext context = new SpringApplicationBuilder(SpecificCustomRetryTemplateConfiguration.class)
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.consumer.retry-template-name=retryTemplateTwo");
RetryTemplate retryTemplateTwo = context.getBean("retryTemplateTwo", RetryTemplate.class);
BindingServiceProperties bindingServiceProperties = context.getBean(BindingServiceProperties.class);
ConsumerProperties consumerProperties = bindingServiceProperties.getConsumerProperties("input");
AbstractBinder binder = context.getBean(AbstractBinder.class);
Method m = AbstractBinder.class.getDeclaredMethod("buildRetryTemplate", ConsumerProperties.class);
m.setAccessible(true);
RetryTemplate retryTemplate = (RetryTemplate) m.invoke(binder, consumerProperties);
assertEquals(retryTemplateTwo, retryTemplate);
}
@EnableBinding(Processor.class)
@Import(TestChannelBinderConfiguration.class)
@EnableAutoConfiguration
public static class SpecificCustomRetryTemplateConfiguration {
@StreamRetryTemplate
public RetryTemplate retryTemplate() {
return new RetryTemplate();
}
@StreamRetryTemplate
public RetryTemplate retryTemplateTwo() {
return new RetryTemplate();
}
@Bean
public RetryTemplate otherRetryTemplate() {
return new RetryTemplate();
}
}
}

127
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java
View File

@@ -110,6 +110,66 @@ public class FunctionInvokerTests {
}
}
@EnableAutoConfiguration
@EnableBinding(Processor.class)
public static class ConverterDoesNotProduceCTConfiguration {
@Bean
public Function<String, String> func() {
return x -> x;
}
@StreamMessageConverter
public MessageConverter customConverter() {
return new MessageConverter() {
@Override
public Message<?> toMessage(Object payload, MessageHeaders headers) {
return new GenericMessage<byte[]>(((String)payload).getBytes());
}
@Override
public Object fromMessage(Message<?> message, Class<?> targetClass) {
String contentType = (String) message.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString();
if (contentType.equals("foo/bar")) {
return new String((byte[])message.getPayload());
}
return null;
}
};
}
}
@EnableAutoConfiguration
@EnableBinding(Processor.class)
public static class ConverterInjectingCTConfiguration {
@Bean
public Function<String, String> func() {
return x -> x;
}
@StreamMessageConverter
public MessageConverter customConverter() {
return new MessageConverter() {
@Override
public Message<?> toMessage(Object payload, MessageHeaders headers) {
return MessageBuilder.withPayload(((String)payload).getBytes()).setHeader(MessageHeaders.CONTENT_TYPE, "ping/pong").build();
}
@Override
public Object fromMessage(Message<?> message, Class<?> targetClass) {
String contentType = (String) message.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString();
if (contentType.equals("foo/bar")) {
return new String((byte[])message.getPayload());
}
return null;
}
};
}
}
@Test
public void testSameMessageTypesAreNotConverted() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
@@ -304,73 +364,6 @@ public class FunctionInvokerTests {
}
}
@EnableAutoConfiguration
@EnableBinding(Processor.class)
public static class ConverterDoesNotProduceCTConfiguration {
@Bean
public Function<String, String> func() {
return x -> x;
}
@StreamMessageConverter
@Bean
public MessageConverter customConverter() {
return new MessageConverter() {
@Override
public Message<?> toMessage(Object payload, MessageHeaders headers) {
return new GenericMessage<byte[]>(((String) payload).getBytes());
}
@Override
public Object fromMessage(Message<?> message, Class<?> targetClass) {
String contentType = (String) message.getHeaders()
.get(MessageHeaders.CONTENT_TYPE).toString();
if (contentType.equals("foo/bar")) {
return new String((byte[]) message.getPayload());
}
return null;
}
};
}
}
@EnableAutoConfiguration
@EnableBinding(Processor.class)
public static class ConverterInjectingCTConfiguration {
@Bean
public Function<String, String> func() {
return x -> x;
}
@StreamMessageConverter
@Bean
public MessageConverter customConverter() {
return new MessageConverter() {
@Override
public Message<?> toMessage(Object payload, MessageHeaders headers) {
return MessageBuilder.withPayload(((String) payload).getBytes())
.setHeader(MessageHeaders.CONTENT_TYPE, "ping/pong").build();
}
@Override
public Object fromMessage(Message<?> message, Class<?> targetClass) {
String contentType = (String) message.getHeaders()
.get(MessageHeaders.CONTENT_TYPE).toString();
if (contentType.equals("foo/bar")) {
return new String((byte[]) message.getPayload());
}
return null;
}
};
}
}
@EnableAutoConfiguration
public static class MyFunctionsConfiguration {

1
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java
View File

@@ -134,7 +134,6 @@ public class GreenfieldFunctionEnableBindingTests {
"http://localhost:"
+ context.getEnvironment().getProperty("local.server.port"),
"hello");
OutputDestination target = context.getBean(OutputDestination.class);
String result = new String(target.receive(10000).getPayload());
System.out.println(result);