diff --git a/docs/pom.xml b/docs/pom.xml index 33199fd55..27a0e2ef5 100644 --- a/docs/pom.xml +++ b/docs/pom.xml @@ -7,7 +7,7 @@ org.springframework.cloud spring-cloud-stream-parent - 2.1.1.BUILD-SNAPSHOT + 2.2.0.BUILD-SNAPSHOT pom spring-cloud-stream-core-docs diff --git a/docs/src/main/asciidoc/preface.adoc b/docs/src/main/asciidoc/preface.adoc index 4a3b22d8b..f2e3060d4 100644 --- a/docs/src/main/asciidoc/preface.adoc +++ b/docs/src/main/asciidoc/preface.adoc @@ -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 <>. -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 <> 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 <> 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 <> 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 <>. +- The spring-cloud-stream-reactive module is deprecated in favor of native support + via <> 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 diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 74021c2d2..abc4e923e 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -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> 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 receive(@Input(Processor.INPUT) Flux input) { - return input.map(s -> s.toUpperCase()); - } -} +spring.cloud.stream.bindings..consumer.retry-template-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 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 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> 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://:/actuator/bindings` Alternative, to see a single binding, access one of the URLs similar to the following: -`http://:/actuator/bindings/myBindingName` +`http://:/actuator/bindings/` 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: diff --git a/pom.xml b/pom.xml index 2609cc4f9..4bbe13d7e 100644 --- a/pom.xml +++ b/pom.xml @@ -4,7 +4,7 @@ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> 4.0.0 spring-cloud-stream-parent - 2.1.1.BUILD-SNAPSHOT + 2.2.0.BUILD-SNAPSHOT pom org.springframework.cloud diff --git a/spring-cloud-stream-binder-test/pom.xml b/spring-cloud-stream-binder-test/pom.xml index 7b94c05a3..154ed8453 100644 --- a/spring-cloud-stream-binder-test/pom.xml +++ b/spring-cloud-stream-binder-test/pom.xml @@ -13,7 +13,7 @@ org.springframework.cloud spring-cloud-stream-parent - 2.1.1.BUILD-SNAPSHOT + 2.2.0.BUILD-SNAPSHOT diff --git a/spring-cloud-stream-integration-tests/pom.xml b/spring-cloud-stream-integration-tests/pom.xml index d91fa5b58..afa33f28f 100644 --- a/spring-cloud-stream-integration-tests/pom.xml +++ b/spring-cloud-stream-integration-tests/pom.xml @@ -12,7 +12,7 @@ org.springframework.cloud spring-cloud-stream-parent - 2.1.1.BUILD-SNAPSHOT + 2.2.0.BUILD-SNAPSHOT diff --git a/spring-cloud-stream-reactive/pom.xml b/spring-cloud-stream-reactive/pom.xml index b0fd9df5d..4566fbcb3 100644 --- a/spring-cloud-stream-reactive/pom.xml +++ b/spring-cloud-stream-reactive/pom.xml @@ -5,7 +5,7 @@ org.springframework.cloud spring-cloud-stream-parent - 2.1.1.BUILD-SNAPSHOT + 2.2.0.BUILD-SNAPSHOT 4.0.0 diff --git a/spring-cloud-stream-schema-server/pom.xml b/spring-cloud-stream-schema-server/pom.xml index 919dde135..24906e094 100644 --- a/spring-cloud-stream-schema-server/pom.xml +++ b/spring-cloud-stream-schema-server/pom.xml @@ -8,7 +8,7 @@ spring-cloud-stream-parent org.springframework.cloud - 2.1.1.BUILD-SNAPSHOT + 2.2.0.BUILD-SNAPSHOT diff --git a/spring-cloud-stream-schema/pom.xml b/spring-cloud-stream-schema/pom.xml index 13c3081b0..84f04ba5e 100644 --- a/spring-cloud-stream-schema/pom.xml +++ b/spring-cloud-stream-schema/pom.xml @@ -5,7 +5,7 @@ spring-cloud-stream-parent org.springframework.cloud - 2.1.1.BUILD-SNAPSHOT + 2.2.0.BUILD-SNAPSHOT 4.0.0 diff --git a/spring-cloud-stream-test-support-internal/pom.xml b/spring-cloud-stream-test-support-internal/pom.xml index 45137dbe8..db2c08e2c 100644 --- a/spring-cloud-stream-test-support-internal/pom.xml +++ b/spring-cloud-stream-test-support-internal/pom.xml @@ -6,7 +6,7 @@ org.springframework.cloud spring-cloud-stream-parent - 2.1.1.BUILD-SNAPSHOT + 2.2.0.BUILD-SNAPSHOT spring-cloud-stream-test-support-internal Set of classes and utility code that may assist in testing both diff --git a/spring-cloud-stream-test-support/pom.xml b/spring-cloud-stream-test-support/pom.xml index bbb7998bb..ddcf23c58 100644 --- a/spring-cloud-stream-test-support/pom.xml +++ b/spring-cloud-stream-test-support/pom.xml @@ -6,7 +6,7 @@ org.springframework.cloud spring-cloud-stream-parent - 2.1.1.BUILD-SNAPSHOT + 2.2.0.BUILD-SNAPSHOT spring-cloud-stream-test-support A set of classes to ease testing of Spring Cloud Stream modules. diff --git a/spring-cloud-stream-tools/pom.xml b/spring-cloud-stream-tools/pom.xml new file mode 100644 index 000000000..78c2298f0 --- /dev/null +++ b/spring-cloud-stream-tools/pom.xml @@ -0,0 +1,84 @@ + + + 4.0.0 + 2.2.0.BUILD-SNAPSHOT + + org.springframework.cloud + spring-cloud-build + 2.1.2.BUILD-SNAPSHOT + + + + spring-cloud-stream-tools + spring-cloud-stream-build-tools + Spring Cloud Stream Build Tools + + + 1.8 + + + + + spring + + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + true + + + false + + + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + false + + + + spring-releases + Spring Releases + https://repo.spring.io/release + + false + + + + + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + true + + + false + + + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + false + + + + spring-releases + Spring Releases + https://repo.spring.io/libs-release-local + + false + + + + + + + diff --git a/spring-cloud-stream/pom.xml b/spring-cloud-stream/pom.xml index 8622dbd99..b1c5dc0d7 100644 --- a/spring-cloud-stream/pom.xml +++ b/spring-cloud-stream/pom.xml @@ -12,7 +12,7 @@ org.springframework.cloud spring-cloud-stream-parent - 2.1.1.BUILD-SNAPSHOT + 2.2.0.BUILD-SNAPSHOT diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/StreamMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/StreamMessageConverter.java index 3c1b309f3..022bdf7b9 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/StreamMessageConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/StreamMessageConverter.java @@ -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 { } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java index c8a2871ab..0c5d2b62f 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java @@ -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 implements ApplicationContextAware, InitializingBean, Binder { @@ -68,10 +71,9 @@ public abstract class AbstractBinder consumerBindingRetryTemplates; /** * For binder implementations that support a prefix, apply the prefix to the name. @@ -196,8 +198,8 @@ public abstract class AbstractBinder errorChannel); } MessageChannel defaultErrorChannel = null; if (getApplicationContext() @@ -571,8 +571,7 @@ public abstract class AbstractMessageChannelBinder errorBridge); } return errorChannel; } @@ -607,23 +606,21 @@ public abstract class AbstractMessageChannelBinder errorChannel); } ErrorMessageSendingRecoverer recoverer; if (errorMessageStrategy == null) { @@ -633,10 +630,10 @@ public abstract class AbstractMessageChannelBinder recoverer); MessageHandler handler; if (polled) { handler = getPolledConsumerErrorMessageHandler(destination, group, @@ -662,8 +659,8 @@ public abstract class AbstractMessageChannelBinder errorHandler); errorChannel.subscribe(handler); } else { @@ -680,11 +677,10 @@ public abstract class AbstractMessageChannelBinder errorBridge); } else { this.logger.warn("The provided errorChannel '" + errorChannelName @@ -731,10 +727,9 @@ public abstract class AbstractMessageChannelBinder 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 extends Pausable { "Binding implementation `" + this.getClass().getName() + "` must implement this operation before it is called"); } - } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java index 662a6eb37..a858675f2 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java @@ -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, 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; } - } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java index 9a3795c28..4ffcb7cdb 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java @@ -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 binder = binderProducingContext.getBean(Binder.class); /* diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java index 9736d06c7..64161f4a2 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java @@ -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 implements Binding { @@ -84,6 +86,12 @@ public class DefaultBinding implements Binding { 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; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java index 6a58f4336..cbae6964b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java @@ -43,7 +43,6 @@ public class ProducerProperties { /** * Signals if this producer needs to be started automatically. - * * Default: true */ private boolean autoStartup = true; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java index 44cdf9e5e..759b92ab7 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java @@ -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 diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java index 966ca5d7a..e1d296f9c 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java @@ -189,6 +189,7 @@ class FunctionInvoker implements Function>, Flux implements Function>, Flux Message wrapOutputToMessage(T value, Message originalMessage) { Message returnMessage = (Message) MessageBuilder.withPayload(value) diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java index 9c424ec9d..02c5eb38a 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java @@ -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 diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/RetryTemplateTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/RetryTemplateTests.java new file mode 100644 index 000000000..f60be955e --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/RetryTemplateTests.java @@ -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 consumerBindingRetryTemplates = (Map) 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(); + } + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java index aca260cc1..f28f20bcf 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java @@ -110,6 +110,66 @@ public class FunctionInvokerTests { } } + @EnableAutoConfiguration + @EnableBinding(Processor.class) + public static class ConverterDoesNotProduceCTConfiguration { + + @Bean + public Function func() { + return x -> x; + } + + @StreamMessageConverter + public MessageConverter customConverter() { + return new MessageConverter() { + + @Override + public Message toMessage(Object payload, MessageHeaders headers) { + return new GenericMessage(((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 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 func() { - return x -> x; - } - - @StreamMessageConverter - @Bean - public MessageConverter customConverter() { - return new MessageConverter() { - - @Override - public Message toMessage(Object payload, MessageHeaders headers) { - return new GenericMessage(((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 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 { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java index ed906fcdf..f04f1ba46 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java @@ -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);