From 0dad98e1e6b58ed7049adf4ed787c0663208c0c0 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Fri, 18 Jan 2019 15:31:36 +0100 Subject: [PATCH 01/16] Created 2.2.x module --- docs/pom.xml | 2 +- pom.xml | 4 ++-- spring-cloud-stream-binder-test/pom.xml | 2 +- spring-cloud-stream-integration-tests/pom.xml | 2 +- spring-cloud-stream-schema-server/pom.xml | 2 +- spring-cloud-stream-schema/pom.xml | 2 +- spring-cloud-stream-test-support-internal/pom.xml | 2 +- spring-cloud-stream-test-support/pom.xml | 2 +- spring-cloud-stream-tools/pom.xml | 4 ++-- spring-cloud-stream/pom.xml | 2 +- 10 files changed, 12 insertions(+), 12 deletions(-) diff --git a/docs/pom.xml b/docs/pom.xml index 7485e6976..c83c0a9b7 100644 --- a/docs/pom.xml +++ b/docs/pom.xml @@ -5,7 +5,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/pom.xml b/pom.xml index 6b5a6ef4d..db2d31afe 100644 --- a/pom.xml +++ b/pom.xml @@ -3,12 +3,12 @@ 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 spring-cloud-build - 2.1.0.BUILD-SNAPSHOT + 2.1.2.BUILD-SNAPSHOT diff --git a/spring-cloud-stream-binder-test/pom.xml b/spring-cloud-stream-binder-test/pom.xml index dcc11a58f..cf6aa4afc 100644 --- a/spring-cloud-stream-binder-test/pom.xml +++ b/spring-cloud-stream-binder-test/pom.xml @@ -11,7 +11,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 539149029..1e51ccf84 100644 --- a/spring-cloud-stream-integration-tests/pom.xml +++ b/spring-cloud-stream-integration-tests/pom.xml @@ -11,7 +11,7 @@ org.springframework.cloud spring-cloud-stream-parent - 2.1.1.BUILD-SNAPSHOT + 2.2.0.BUILD-SNAPSHOT diff --git a/spring-cloud-stream-schema-server/pom.xml b/spring-cloud-stream-schema-server/pom.xml index 2ea6df343..08904732a 100644 --- a/spring-cloud-stream-schema-server/pom.xml +++ b/spring-cloud-stream-schema-server/pom.xml @@ -7,7 +7,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 c2ea26fdc..01a1f2f46 100644 --- a/spring-cloud-stream-schema/pom.xml +++ b/spring-cloud-stream-schema/pom.xml @@ -3,7 +3,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 ea694c373..2ebbc63b4 100644 --- a/spring-cloud-stream-test-support-internal/pom.xml +++ b/spring-cloud-stream-test-support-internal/pom.xml @@ -4,7 +4,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 59500bf56..05e45a041 100644 --- a/spring-cloud-stream-test-support/pom.xml +++ b/spring-cloud-stream-test-support/pom.xml @@ -4,7 +4,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 index 97590bfce..78c2298f0 100644 --- a/spring-cloud-stream-tools/pom.xml +++ b/spring-cloud-stream-tools/pom.xml @@ -1,11 +1,11 @@ 4.0.0 - 2.1.1.BUILD-SNAPSHOT + 2.2.0.BUILD-SNAPSHOT org.springframework.cloud spring-cloud-build - 2.1.0.BUILD-SNAPSHOT + 2.1.2.BUILD-SNAPSHOT diff --git a/spring-cloud-stream/pom.xml b/spring-cloud-stream/pom.xml index 39aadefc8..36030bef5 100644 --- a/spring-cloud-stream/pom.xml +++ b/spring-cloud-stream/pom.xml @@ -10,7 +10,7 @@ org.springframework.cloud spring-cloud-stream-parent - 2.1.1.BUILD-SNAPSHOT + 2.2.0.BUILD-SNAPSHOT From 6cc83564042cfc646d7e6050631410988fe209a3 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Fri, 18 Jan 2019 15:54:45 +0100 Subject: [PATCH 02/16] GH-1559 Deprecated reactive support Removed reactive module documentation Added initial documentation for reactive functions Resolves #1559 --- docs/src/main/asciidoc/preface.adoc | 34 +-- .../main/asciidoc/spring-cloud-stream.adoc | 228 ++++-------------- 2 files changed, 56 insertions(+), 206 deletions(-) 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 028280020..b42bf42a4 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,162 +1054,6 @@ 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 - -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] ----- -@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()); - } -} ----- - -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 @@ -1321,8 +1195,8 @@ spring: rabbitmq: host: ---- -NOTE: The `environment` property of the particular binder can also be used for any Spring Boot property, -including this `spring.main.sources` which can be useful for adding additional configurations for the +NOTE: The `environment` property of the particular binder can also be used for any Spring Boot property, +including this `spring.main.sources` which can be useful for adding additional configurations for the particular binders, e.g. overriding auto-configured beans. For example; @@ -2499,20 +2373,20 @@ When autoconfiguration is disabled, the test binder is available on the classpat [[spring_integration_test_binder]] === Spring Integration Test Binder -Current test binder was specifically designed to facilitate _unit testing_ of the actual messaging components and thus bypasses some of the core functionality of the binder API. -While such light-weight approach is sufficient for a lot of cases, it usually requires additional _integration testing_ with real binders (e.g., Rabbit, Kafka etc). +Current test binder was specifically designed to facilitate _unit testing_ of the actual messaging components and thus bypasses some of the core functionality of the binder API. +While such light-weight approach is sufficient for a lot of cases, it usually requires additional _integration testing_ with real binders (e.g., Rabbit, Kafka etc). To begin bridging the gap between _unit_ and _integration_ testing we've developed a new test binder which uses https://spring.io/projects/spring-integration[Spring Integration] framework as an in-JVM Message Broker essentially giving you the best of both worlds - a real binder without the networking. To enable Spring Integration Test Binder all you need is: -- Add required dependencies +- Add required dependencies - Remove the dependency for `spring-cloud-stream-test-support` ***Add required dependencies*** -Below is the example of the required Maven POM entries which could be easily retrofitted into Gradle. +Below is the example of the required Maven POM entries which could be easily retrofitted into Gradle. [source,xml] ---- @@ -2584,7 +2458,7 @@ public class DemoTestBinderApplication { @Test public void sampleTest() { ApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.class, + TestChannelBinderConfiguration.class, DemoTestBinderApplication.class) .web(WebApplicationType.NONE).run(); InputDestination source = context.getBean(InputDestination.class); @@ -2594,20 +2468,20 @@ public void sampleTest() { } ---- -In the above you simply create an ApplicationContext with your configuration (your application) while additionally supplying `TestChannelBinderConfiguration` +In the above you simply create an ApplicationContext with your configuration (your application) while additionally supplying `TestChannelBinderConfiguration` provided by the framework. Then you access `InputDestination` and `OutputDestination` beans to send/receive messages. In the context of this binder `InputDestination` and `OutputDestination` emulate remote destinations such as Rabbit _exchange/queue_ or Kafka _topic_. In the future we plan to simplify the API. -NOTE: In its current state Spring Integration Test Binder only supports the three bindings provided by the framework (Source, Processor, Sink) specifically to promote +NOTE: In its current state Spring Integration Test Binder only supports the three bindings provided by the framework (Source, Processor, Sink) specifically to promote light-weight microservices architectures rather then general purpose messaging applications. - + ==== Spring Integration Test Binder and PollableMessageSource Spring Integration Test Binder also allows you to write tests when working with `PollableMessageSource` (see <> for more details). -The important thing that needs to be understood though is that polling is not event-driven, and that `PollableMessageSource` is a strategy which exposes operation to produce (poll for) a Message (singular). -How often you poll or how many threads you use or where you're polling from (message queue or file system) is entirely up to you; +The important thing that needs to be understood though is that polling is not event-driven, and that `PollableMessageSource` is a strategy which exposes operation to produce (poll for) a Message (singular). +How often you poll or how many threads you use or where you're polling from (message queue or file system) is entirely up to you; In other words it is your responsibility to configure Poller or Threads or the actual source of Message. Luckily Spring has plenty of abstractions to configure exactly that. Let's look at the example: @@ -2624,7 +2498,7 @@ public void samplePollingTest() { System.out.println("Message 2: " + new String(destination.receive().getPayload())); System.out.println("Message 3: " + new String(destination.receive().getPayload())); } - + @EnableBinding(SamplePolledConfiguration.PolledConsumer.class) @Import(TestChannelBinderConfiguration.class) @EnableAutoConfiguration @@ -2641,7 +2515,7 @@ public static class SamplePolledConfiguration { })) { Thread.sleep(2000); } - } + } catch (Exception e) { // handle failure } @@ -2649,7 +2523,7 @@ public static class SamplePolledConfiguration { }); }; } - + public static interface PolledConsumer extends Source { @Input PollableMessageSource pollableSource(); @@ -2657,7 +2531,7 @@ public static class SamplePolledConfiguration { } ---- -The above (very rudimentary) example will produce 3 messages in 2 second intervals sending them to the output destination of `Source` +The above (very rudimentary) example will produce 3 messages in 2 second intervals sending them to the output destination of `Source` which this binder sends to `OutputDestination` where we retrieve them (for any assertions). Currently it prints the following: [source, text] @@ -2666,9 +2540,9 @@ Message 1: POLLED DATA Message 2: POLLED DATA Message 3: POLLED DATA ---- -As you can see the data is the same. That is because this binder defines a default implementation of the actual `MessageSource` - the source -from which the Messages are polled using `poll()` operation. While sufficient for most testing scenarios, there are cases where you may want -to define your own `MessageSource`. To do so simply configure a bean of type `MessageSource` in your test configuration providing your own +As you can see the data is the same. That is because this binder defines a default implementation of the actual `MessageSource` - the source +from which the Messages are polled using `poll()` operation. While sufficient for most testing scenarios, there are cases where you may want +to define your own `MessageSource`. To do so simply configure a bean of type `MessageSource` in your test configuration providing your own implementation of Message sourcing. Here is the example: @@ -2688,7 +2562,7 @@ Message 2: MY OWN DATA D8F3A477-5547-41B4-9434-E69DA7616FEE Message 3: MY OWN DATA 20BF2E64-7FF4-4CB6-A823-4053D30B5C74 ---- -NOTE: DO NOT name this bean `messageSource` as it is going to be in conflict with the bean of the same name (different type) +NOTE: DO NOT name this bean `messageSource` as it is going to be in conflict with the bean of the same name (different type) provided by Spring Boot for unrelated reasons. From cf8da3594f13d4747e01e071169524346fa2134c Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 21 Jan 2019 15:30:27 +0100 Subject: [PATCH 03/16] GH-1578 Fixed error handling in DefaultPollableMessageSource ensured that RequeueCurrentMessageException is treated differently where message does not end up in error channel Resolves #1578 --- spring-cloud-stream-reactive/pom.xml | 2 +- .../binder/DefaultPollableMessageSource.java | 23 ++++++++----------- .../stream/binder/PollableConsumerTests.java | 2 +- 3 files changed, 11 insertions(+), 16 deletions(-) diff --git a/spring-cloud-stream-reactive/pom.xml b/spring-cloud-stream-reactive/pom.xml index afdd00844..6be030a1a 100644 --- a/spring-cloud-stream-reactive/pom.xml +++ b/spring-cloud-stream-reactive/pom.xml @@ -3,7 +3,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/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java index 6db615529..abb2b0a42 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-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. @@ -202,18 +202,7 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life .getAcknowledgmentCallback(message); try { if (this.retryTemplate == null) { - if (this.errorChannel == null) { - this.handle(message, handler); - } - else { - try { - this.handle(message, handler); - } - catch (Exception e) { - this.messagingTemplate - .send(this.errorChannel, this.errorMessageStrategy.buildErrorMessage(e, attributesHolder.get())); - } - } + this.handle(message, handler); } else { this.retryTemplate.execute(context -> { @@ -224,8 +213,14 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life return true; } catch (MessagingException e) { - if (!ackCallback.isAcknowledged() && shouldRequeue(e)) { + if (this.retryTemplate == null && !shouldRequeue(e)) { + this.messagingTemplate + .send(this.errorChannel, this.errorMessageStrategy.buildErrorMessage(e, attributesHolder.get())); + return true; + } + else if (!ackCallback.isAcknowledged() && shouldRequeue(e)) { AckUtils.requeue(ackCallback); + return true; } else { AckUtils.autoNack(ackCallback); 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 a6882bd98..edc1bc0e0 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 @@ -356,7 +356,7 @@ public class PollableConsumerTests { count.incrementAndGet(); throw new RequeueCurrentMessageException("test retry"); })).isTrue(); - fail("Expected exception"); + //fail("Expected exception"); // see GH-1578 } catch (Exception e) { // no op From 723c6877650362654b55633196272351c4f8f1d5 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 16 Jan 2019 13:16:05 -0500 Subject: [PATCH 04/16] GH-1569: Fix binder-specific environment merging Fixes spring-cloud/spring-cloud-stream#1569 When `inheritEnvironment = true` (default), some provided properties (including `spring.main.sources`) are not populated into the target binder's environment, since Spring Boot relies on the presence of the `configurationProperties` properties source which is transferred from the parent context * Remove `configurationProperties` from the `binderEnvironment` before starting binder's application context * Ensure that `spring.main.sources` is applied for the binder's application context in the `BinderFactoryConfigurationTests` * Fix `GreenfieldFunctionEnableBindingTests` for random HTTP port since `8080` is too generic and clashes with build environment * Fix `StreamListenerAnnotatedMethodArgumentsTests` for `Locale.US` for proper assertion against validation message * Mention `spring.profiles.active` configuration property for the binder specific environment --- .../main/asciidoc/spring-cloud-stream.adoc | 10 ++++++ ...ListenerAnnotatedMethodArgumentsTests.java | 14 ++++++-- .../stream/binder/DefaultBinderFactory.java | 9 +++-- .../BinderFactoryConfigurationTests.java | 36 +++++++++++++++++-- .../GreenfieldFunctionEnableBindingTests.java | 28 +++++++++++---- 5 files changed, 82 insertions(+), 15 deletions(-) diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index b42bf42a4..37909b4a7 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -1208,6 +1208,16 @@ environment: sources: com.acme.config.MyCustomBinderConfiguration ---- +To activate some specific profile for the particular binder environment, you should use a `spring.profiles.active` property: + +[source, yaml] +---- +environment: + spring: + profiles: + active: myBinderProfile +---- + === Binding visualization and control Since version 2.0, Spring Cloud Stream supports visualization and control of the Bindings through Actuator endpoints. diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotatedMethodArgumentsTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotatedMethodArgumentsTests.java index 295962c95..afef15c37 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotatedMethodArgumentsTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotatedMethodArgumentsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 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. @@ -18,11 +18,13 @@ package org.springframework.cloud.stream.config; import java.util.ArrayList; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.UUID; import javax.validation.Valid; +import org.junit.BeforeClass; import org.junit.Test; import org.springframework.boot.SpringApplication; @@ -49,12 +51,18 @@ import static org.springframework.cloud.stream.binding.StreamListenerErrorMessag * @author Marius Bogoevici * @author Ilayaperumal Gopinathan * @author Oleg Zhurakousky + * @author Artem Bilan */ public class StreamListenerAnnotatedMethodArgumentsTests { + @BeforeClass + public static void init() { + Locale.setDefault(Locale.US); + } + @Test @SuppressWarnings("unchecked") - public void testAnnotatedArguments() throws Exception { + public void testAnnotatedArguments() { ConfigurableApplicationContext context = SpringApplication.run(TestPojoWithAnnotatedArguments.class, "--server.port=0"); @@ -79,7 +87,7 @@ public class StreamListenerAnnotatedMethodArgumentsTests { } @Test - public void testInputAnnotationAtMethodParameter() throws Exception { + public void testInputAnnotationAtMethodParameter() { try { SpringApplication.run(TestPojoWithInvalidInputAnnotatedArgument.class, "--server.port=0"); fail("Exception expected: " + INVALID_DECLARATIVE_METHOD_PARAMETERS); 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 5336ce6d7..f7262e835 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 @@ -1,5 +1,5 @@ /* - * Copyright 2015-2018 the original author or authors. + * Copyright 2015-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. @@ -52,6 +52,7 @@ import org.springframework.util.StringUtils; * @author Gary Russell * @author Oleg Zhurakousky * @author Soby Chacko + * @author Artem Bilan */ public class DefaultBinderFactory implements BinderFactory, DisposableBean, ApplicationContextAware { @@ -244,10 +245,12 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl if (environment != null && (useApplicationContextAsParent || binderConfiguration.isInheritEnvironment())) { StandardEnvironment binderEnvironment = new StandardEnvironment(); binderEnvironment.merge(environment); + // See ConfigurationPropertySources.ATTACHED_PROPERTY_SOURCE_NAME + binderEnvironment.getPropertySources().remove("configurationProperties"); springApplicationBuilder.environment(binderEnvironment); } - ConfigurableApplicationContext binderProducingContext = springApplicationBuilder - .run(args.toArray(new String[args.size()])); + ConfigurableApplicationContext binderProducingContext = + springApplicationBuilder.run(args.toArray(new String[0])); Binder binder = binderProducingContext.getBean(Binder.class); /* diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java index 20b212878..fa5b41495 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2018 the original author or authors. + * Copyright 2015-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. @@ -33,7 +33,10 @@ import org.springframework.cloud.stream.binder.stub2.StubBinder2ConfigurationA; import org.springframework.cloud.stream.binder.stub2.StubBinder2ConfigurationB; import org.springframework.cloud.stream.config.BinderFactoryConfiguration; import org.springframework.cloud.stream.config.BindingServiceConfiguration; +import org.springframework.context.ApplicationContext; import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; import org.springframework.core.io.ClassPathResource; import org.springframework.core.io.DefaultResourceLoader; @@ -47,6 +50,7 @@ import static org.junit.Assert.fail; * @author Marius Bogoevici * @author Ilayaperumal Gopinathan * @author Soby Chacko + * @author Artem Bilan */ public class BinderFactoryConfigurationTests { @@ -130,6 +134,7 @@ public class BinderFactoryConfigurationTests { ConfigurableApplicationContext context = createBinderTestContext( new String[] { "binder1" }, "binder1.name=foo", "spring.cloud.stream.binders.custom.environment.foo=bar", + "spring.cloud.stream.binders.custom.environment.spring.main.sources=org.springframework.cloud.stream.binder.BinderFactoryConfigurationTests.AdditionalBinderConfiguration", "spring.cloud.stream.binders.custom.type=binder1"); BinderFactory binderFactory = context.getBean(BinderFactory.class); @@ -138,6 +143,12 @@ public class BinderFactoryConfigurationTests { assertThat(binder1).hasFieldOrPropertyWithValue("name", "foo"); assertThat(binderFactory.getBinder(null, MessageChannel.class)).isSameAs(binder1); + + SimpleApplication simpleApplication = context.getBean(SimpleApplication.class); + + assertThat(simpleApplication.binderContext).isNotNull(); + + assertThat(simpleApplication.binderContext.containsBean("fooBean")).isTrue(); } @SuppressWarnings("rawtypes") @@ -214,7 +225,7 @@ public class BinderFactoryConfigurationTests { BinderTypeRegistry binderTypeRegistry = context.getBean(BinderTypeRegistry.class); assertThat(binderTypeRegistry).isNotNull(); assertThat(binderTypeRegistry.getAll().size()).isEqualTo(3); - assertThat(binderTypeRegistry.getAll().keySet().contains("binder1")); + assertThat(binderTypeRegistry.getAll().keySet()).contains("binder1"); assertThat((Class[]) binderTypeRegistry.get("binder1").getConfigurationClasses()) .contains(StubBinder1Configuration.class); @@ -260,5 +271,26 @@ public class BinderFactoryConfigurationTests { @EnableBinding public static class SimpleApplication { + private volatile ApplicationContext binderContext; + + @Bean + public DefaultBinderFactory.Listener testBinderListener() { + return (configurationName, binderContext) -> { + this.binderContext = binderContext; + }; + + } + } + + @Configuration + public static class AdditionalBinderConfiguration { + + @Bean + public String fooBean() { + return "foo"; + } + + } + } 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 d04ca43da..8234370c8 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 @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-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. @@ -61,6 +61,7 @@ import static org.assertj.core.api.Assertions.assertThat; * This test validates proper function binding for applications where EnableBinding is declared. * * @author Oleg Zhurakousky + * @author Artem Bilan */ public class GreenfieldFunctionEnableBindingTests { @@ -110,10 +111,14 @@ public class GreenfieldFunctionEnableBindingTests { @Test public void testHttpEndpoint() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(HttpInboundEndpoint.class)).web( - WebApplicationType.SERVLET).run("--spring.cloud.stream.function.definition=upperCase", "--spring.jmx.enabled=false")) { + TestChannelBinderConfiguration.getCompleteConfiguration(HttpInboundEndpoint.class)) + .web(WebApplicationType.SERVLET) + .run("--spring.cloud.stream.function.definition=upperCase", + "--spring.jmx.enabled=false", + "--server.port=0")) { TestRestTemplate restTemplate = new TestRestTemplate(); - restTemplate.postForLocation("http://localhost:8080", "hello"); + restTemplate.postForLocation( + "http://localhost:" + context.getEnvironment().getProperty("local.server.port"), "hello"); OutputDestination target = context.getBean(OutputDestination.class); String result = new String(target.receive(10000).getPayload()); @@ -139,7 +144,6 @@ public class GreenfieldFunctionEnableBindingTests { Foo result = mapper.readValue(payload, Foo.class); assertThat(result.getBar()).isEqualTo("bar"); - } } @@ -147,24 +151,29 @@ public class GreenfieldFunctionEnableBindingTests { @EnableAutoConfiguration @EnableBinding(Source.class) public static class SourceFromSupplier { + @Bean public Supplier date() { return () -> new Date(12345L); } + } @EnableAutoConfiguration @EnableBinding(Processor.class) public static class ProcessorFromFunction { + @Bean public Function toUpperCase() { - return s -> s.toUpperCase(); + return String::toUpperCase; } + } @EnableAutoConfiguration @EnableBinding(Sink.class) public static class SinkFromConsumer { + @Bean public PollableChannel result() { return new QueueChannel(); @@ -177,6 +186,7 @@ public class GreenfieldFunctionEnableBindingTests { System.out.println(s); }; } + } @EnableAutoConfiguration @@ -188,7 +198,7 @@ public class GreenfieldFunctionEnableBindingTests { @Bean public Function upperCase() { - return s -> s.toUpperCase(); + return String::toUpperCase; } @Bean @@ -200,6 +210,7 @@ public class GreenfieldFunctionEnableBindingTests { .requestChannel(this.source.output()); return httpRequestHandler.get(); } + } @EnableAutoConfiguration @@ -225,9 +236,11 @@ public class GreenfieldFunctionEnableBindingTests { return MessageBuilder.withPayload(foo).setHeader("foo","foo").build(); }; } + } static class Foo { + String bar; public String getBar() { @@ -238,4 +251,5 @@ public class GreenfieldFunctionEnableBindingTests { this.bar = bar; } } + } From f4d1016ca4742575961058dfcc93810bc206709a Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 21 Jan 2019 16:43:10 +0100 Subject: [PATCH 05/16] minor doc polish Resolves #1585 --- docs/src/main/asciidoc/spring-cloud-stream.adoc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 37909b4a7..755255cb6 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -1208,7 +1208,7 @@ environment: sources: com.acme.config.MyCustomBinderConfiguration ---- -To activate some specific profile for the particular binder environment, you should use a `spring.profiles.active` property: +To activate a specific profile for the particular binder environment, you should use a `spring.profiles.active` property: [source, yaml] ---- From 2fa4e1cc75e583755be7dd99b95659dcd0494b73 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 21 Jan 2019 17:31:57 +0100 Subject: [PATCH 06/16] GH-1556 Added isInput() operation to Binding Resolves #1556 --- .../binder/AbstractMessageChannelBinder.java | 15 +++++++++++++++ .../cloud/stream/binder/Binding.java | 14 ++++++++++++-- 2 files changed, 27 insertions(+), 2 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index 7f0bce482..9849c0f86 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -213,6 +213,11 @@ public abstract class AbstractMessageChannelBinder extends Pausable { 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 @@ -108,4 +107,15 @@ public interface Binding extends Pausable { * and a new Binding should be created instead. */ void unbind(); + + /** + * Returns boolean flag representing this binding's type. If 'true' this binding is an 'input' binding + * otherwise it is 'output' (as in binding annotated with either @Input or @Output). + * + * @return 'true' if this binding represents an input binding. + */ + default boolean isInput() { + throw new UnsupportedOperationException("Binding implementation `" + this.getClass().getName() + + "` must implement this operation before it is called"); + } } From 144627c4355b51b81ace7c4254e3a8b68ba21a3d Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 22 Jan 2019 08:51:12 +0100 Subject: [PATCH 07/16] GH-1476, GH-1568 exposed actual binding name Exposed the actual binding name in addition to already exposed destination name of the binding Fixed Actuator endpoint to ensure it uses binding name Resolves #1476 Resolves #1568 --- .../cloud/stream/binder/Binding.java | 15 +++++++++++++-- .../cloud/stream/binder/DefaultBinding.java | 11 +++++++++-- .../cloud/stream/endpoint/BindingsEndpoint.java | 6 +++--- 3 files changed, 25 insertions(+), 7 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java index 7096e5f44..380ca38e4 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java @@ -92,14 +92,25 @@ public interface Binding extends Pausable { } /** - * Returns the name of this binding (i.e., channel name) + * Returns the name of the destination for this binding * - * @return binding name + * @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 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 7d71f4238..8dd989aa0 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. @@ -23,6 +23,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,7 @@ import org.springframework.util.StringUtils; * * @see org.springframework.cloud.stream.annotation.EnableBinding */ -@JsonPropertyOrder({ "name", "group", "pausable", "state"}) +@JsonPropertyOrder({ "name", "bindingName", "group", "pausable", "state"}) @JsonIgnoreProperties("running") public class DefaultBinding implements Binding { @@ -84,6 +85,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/endpoint/BindingsEndpoint.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/endpoint/BindingsEndpoint.java index c22893140..da5607bb8 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/endpoint/BindingsEndpoint.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/endpoint/BindingsEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-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. @@ -116,10 +116,10 @@ public class BindingsEndpoint { return outputBindings; } - private Binding locateBinding(String name) { + private Binding locateBinding(String bindingName) { Stream> bindings = Stream.concat(this.gatherInputBindings().stream(), this.gatherOutputBindings().stream()); return bindings - .filter(binding -> name.equals(binding.getName())) + .filter(binding -> bindingName.equals(binding.getBindingName())) .findFirst() .orElse(null); } From 1624c431a7aeceb712679ccc487ca287678cb1f1 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 22 Jan 2019 09:31:00 +0100 Subject: [PATCH 08/16] GH-1476, GH-1568 minor polishing of docs and binding display order --- docs/src/main/asciidoc/spring-cloud-stream.adoc | 2 +- .../org/springframework/cloud/stream/binder/DefaultBinding.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 755255cb6..bd64838f4 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -1266,7 +1266,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/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 8dd989aa0..7d541e487 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 @@ -40,7 +40,7 @@ import org.springframework.util.StringUtils; * * @see org.springframework.cloud.stream.annotation.EnableBinding */ -@JsonPropertyOrder({ "name", "bindingName", "group", "pausable", "state"}) +@JsonPropertyOrder({"bindingName", "name", "group", "pausable", "state"}) @JsonIgnoreProperties("running") public class DefaultBinding implements Binding { From 5d8d968ead181a13e4422844bda9026c3146db5b Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 22 Jan 2019 16:13:01 +0100 Subject: [PATCH 09/16] GH-1448 Fixed ServerController override Resolves #1448 --- .../schema/server/config/SchemaServerConfiguration.java | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/config/SchemaServerConfiguration.java b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/config/SchemaServerConfiguration.java index 6e82ff191..8532583bd 100644 --- a/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/config/SchemaServerConfiguration.java +++ b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/config/SchemaServerConfiguration.java @@ -33,6 +33,7 @@ import org.springframework.cloud.stream.schema.server.support.AvroSchemaValidato import org.springframework.cloud.stream.schema.server.support.SchemaValidator; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; import org.springframework.data.jpa.repository.config.EnableJpaRepositories; /** @@ -42,6 +43,7 @@ import org.springframework.data.jpa.repository.config.EnableJpaRepositories; @Configuration @EnableJpaRepositories(basePackageClasses = SchemaRepository.class) @EnableConfigurationProperties(SchemaServerProperties.class) +@Import(ServerController.class) public class SchemaServerConfiguration { @Bean @@ -58,12 +60,6 @@ public class SchemaServerConfiguration { }; } - @Bean - public ServerController serverController(SchemaRepository repository, - SchemaServerProperties schemeServerProperties) { - return new ServerController(repository, schemaValidators(), schemeServerProperties); - } - @Bean public Map schemaValidators() { Map validatorMap = new HashMap<>(); From 44f8ae33f228c90bcd1121e8c270708d7ab696ac Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 22 Jan 2019 18:16:07 +0100 Subject: [PATCH 10/16] GH-1535 Added support for Binding auto-startup Resolves #1535 --- .../main/asciidoc/spring-cloud-stream.adoc | 8 ++++ .../binder/AbstractMessageChannelBinder.java | 4 +- .../stream/binder/ConsumerProperties.java | 15 ++++++++ .../stream/binder/ProducerProperties.java | 15 ++++++++ .../stream/binding/BindingServiceTests.java | 37 +++++++++++++++++++ 5 files changed, 77 insertions(+), 2 deletions(-) diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index bd64838f4..1f4d223a1 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -1404,6 +1404,10 @@ The following binding properties are available for input bindings only and must Default values can be set by using the `spring.cloud.stream.default.consumer` prefix (for example, `spring.cloud.stream.default.consumer.headerMode=none`). +autoStartup:: +Signals if this consumer needs to be started automatically ++ +Default: `true`. concurrency:: The concurrency of the inbound consumer. + @@ -1478,6 +1482,10 @@ The following binding properties are available for output bindings only and must Default values can be set by using the prefix `spring.cloud.stream.default.producer` (for example, `spring.cloud.stream.default.producer.partitionKeyExpression=payload.id`). +autoStartup:: +Signals if this consumer needs to be started automatically ++ +Default: `true`. partitionKeyExpression:: A SpEL expression that determines how to partition outbound data. If set, or if `partitionKeyExtractorClass` is set, outbound data on this channel is partitioned. `partitionCount` must be set to a value greater than 1 to be effective. diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index 9849c0f86..ecb33c249 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -190,7 +190,7 @@ public abstract class AbstractMessageChannelBinder { @Override diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java index 609a41742..99a850990 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java @@ -16,10 +16,12 @@ package org.springframework.cloud.stream.binding; +import java.lang.reflect.Field; import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.Iterator; +import java.util.List; import java.util.Map; import java.util.Properties; import java.util.concurrent.CountDownLatch; @@ -37,7 +39,9 @@ import org.mockito.stubbing.Answer; import org.springframework.beans.factory.NoSuchBeanDefinitionException; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.boot.context.properties.source.MapConfigurationPropertySource; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.annotation.Input; @@ -54,22 +58,31 @@ import org.springframework.cloud.stream.binder.DefaultBinderTypeRegistry; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceConfiguration; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; +import org.springframework.cloud.stream.messaging.Processor; +import org.springframework.cloud.stream.messaging.Sink; import org.springframework.cloud.stream.reflection.GenericsUtils; import org.springframework.cloud.stream.utils.MockBinderConfiguration; +import org.springframework.context.ApplicationContext; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Import; +import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.core.DestinationResolutionException; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.util.ReflectionUtils; import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.fail; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; @@ -611,6 +624,30 @@ public class BindingServiceTests { scheduler.destroy(); } + @SuppressWarnings("unchecked") + @Test + public void testBindingAutostartup() throws Exception { + ApplicationContext context = new SpringApplicationBuilder(FooConfiguration.class) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", "--spring.cloud.stream.bindings.input.consumer.auto-startup=false"); + BindingService bindingService = context.getBean(BindingService.class); + + Field cbField = ReflectionUtils.findField(BindingService.class, "consumerBindings"); + cbField.setAccessible(true); + Map cbMap = (Map) cbField.get(bindingService); + Binding inputBinding = ((List>)cbMap.get("input")).get(0); + assertFalse(inputBinding.isRunning()); + } + + @EnableBinding(Sink.class) + @Import(TestChannelBinderConfiguration.class) + @EnableAutoConfiguration + public static class FooConfiguration { + @ServiceActivator(inputChannel=Processor.INPUT) + public void echo(Message value) throws Exception { + } + } + private DefaultBinderFactory createMockBinderFactory() { BinderTypeRegistry binderTypeRegistry = createMockBinderTypeRegistry(); return new DefaultBinderFactory( From 5a9c6a616236f5101c2e0c15b750ee11ab62039b Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 22 Jan 2019 20:27:50 +0100 Subject: [PATCH 11/16] GH-1240 update AbstractMessageChannelBinder to use Spring 5 bean registration Resolves #1240 --- .../binder/AbstractMessageChannelBinder.java | 39 +++++-------------- 1 file changed, 10 insertions(+), 29 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index ecb33c249..e465617fb 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -30,7 +30,6 @@ import org.reactivestreams.Publisher; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.support.DefaultSingletonBeanRegistry; import org.springframework.cloud.stream.config.ListenerContainerCustomizer; import org.springframework.cloud.stream.function.IntegrationFlowFunctionSupport; @@ -43,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; @@ -521,9 +521,8 @@ public abstract class AbstractMessageChannelBinder errorChannel); } MessageChannel defaultErrorChannel = null; if (getApplicationContext().containsBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)) { @@ -547,8 +545,7 @@ public abstract class AbstractMessageChannelBinder errorBridge); } return errorChannel; } @@ -582,9 +579,8 @@ public abstract class AbstractMessageChannelBinder errorChannel); } ErrorMessageSendingRecoverer recoverer; if (errorMessageStrategy == null) { @@ -606,8 +601,7 @@ public abstract class AbstractMessageChannelBinder recoverer); MessageHandler handler; if (polled) { handler = getPolledConsumerErrorMessageHandler(destination, group, consumerProperties); @@ -627,8 +621,8 @@ public abstract class AbstractMessageChannelBinder errorHandler); errorChannel.subscribe(handler); } else { @@ -644,8 +638,7 @@ public abstract class AbstractMessageChannelBinder errorBridge); } else { logger.warn("The provided errorChannel '" + errorChannelName + "' is an instance of DirectChannel, " @@ -812,18 +805,6 @@ public abstract class AbstractMessageChannelBinder Date: Thu, 31 Jan 2019 08:08:06 +0100 Subject: [PATCH 12/16] GH-1594 Ensured Function conversion honors outbound CT Ensured that Message conversion that happenes within FunctionInvoker honors the outbound binding content-type in the event MessageConverter itself did not sent one. Restructured some code/API around function invocation Added tests Resolves #1594 --- .../binder/AbstractMessageChannelBinder.java | 30 ---- .../binding/MessageConverterConfigurer.java | 10 +- .../stream/function/FunctionInvoker.java | 41 +++-- .../IntegrationFlowFunctionSupport.java | 10 +- .../function/StreamFunctionProperties.java | 36 +++-- .../stream/function/FunctionInvokerTests.java | 146 ++++++++++++++++-- 6 files changed, 197 insertions(+), 76 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index e465617fb..08d02146b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -16,7 +16,6 @@ package org.springframework.cloud.stream.binder; -import java.lang.reflect.Method; import java.util.LinkedHashMap; import java.util.Map; import java.util.function.Consumer; @@ -65,7 +64,6 @@ import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.retry.RecoveryCallback; import org.springframework.util.Assert; -import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; /** @@ -835,7 +833,6 @@ public abstract class AbstractMessageChannelBinder 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 default content type + //to MessageConverters instead of interceptors + if (message.getPayload() instanceof byte[] && message.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE)) { + return message; + } + + // ===== 1.3 backward compatibility code part-1 === String oct = message.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE) ? message.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString() : null; String ct = message.getPayload() instanceof String 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 c7fdc9945..df1444d2a 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 @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-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,7 +16,9 @@ package org.springframework.cloud.stream.function; +import java.lang.reflect.Field; import java.time.Duration; +import java.util.Map; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import java.util.function.Function; @@ -30,6 +32,7 @@ import org.springframework.cloud.function.context.FunctionType; import org.springframework.cloud.function.context.catalog.FunctionInspector; import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; @@ -38,6 +41,8 @@ import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.CompositeMessageConverter; import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; +import org.springframework.util.MimeType; +import org.springframework.util.ReflectionUtils; /** * @@ -54,6 +59,13 @@ class FunctionInvoker implements Function>, Flux inputClass; private final Class outputClass; @@ -70,6 +82,10 @@ class FunctionInvoker implements Function>, Flux implements Function>, Flux implements Function>, Flux implements Function>, Flux) ReflectionUtils.getField(MESSAGE_HEADERS_FIELD, returnMessage.getHeaders())) + .put(MessageHeaders.CONTENT_TYPE, MimeType.valueOf(bindingServiceProperties.getBindingProperties("output").getContentType())); } Assert.notNull(returnMessage, "Failed to convert result value '" + value + "' to message."); } return returnMessage; } + @SuppressWarnings("unchecked") private Message wrapOutputToMessage(T value, Message originalMessage) { Message returnMessage = (Message) MessageBuilder.withPayload(value).copyHeaders(originalMessage.getHeaders()) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java index 29705f4f0..d9df46d49 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-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. @@ -58,8 +58,6 @@ public class IntegrationFlowFunctionSupport { private final StreamFunctionProperties functionProperties; - private final BindingServiceProperties bindingServiceProperties; - @Autowired private MessageChannel errorChannel; @@ -81,7 +79,7 @@ public class IntegrationFlowFunctionSupport { this.functionInspector = functionInspector; this.messageConverterFactory = messageConverterFactory; this.functionProperties = functionProperties; - this.bindingServiceProperties = bindingServiceProperties; + this.functionProperties.setBindingServiceProperties(bindingServiceProperties); } /** @@ -158,14 +156,14 @@ public class IntegrationFlowFunctionSupport { if (inputChannel instanceof IntegrationObjectSupport) { String inputBindingName = ((IntegrationObjectSupport)inputChannel).getComponentName(); if (StringUtils.hasText(inputBindingName)) { - this.functionProperties.setConsumerProperties(this.bindingServiceProperties.getConsumerProperties(inputBindingName)); + this.functionProperties.setInputDestinationName(inputBindingName); } } if (outputChannel instanceof IntegrationObjectSupport) { String outputBindingName = ((IntegrationObjectSupport)outputChannel).getComponentName(); if (StringUtils.hasText(outputBindingName)) { - this.functionProperties.setProducerProperties(this.bindingServiceProperties.getProducerProperties(outputBindingName)); + this.functionProperties.setOutputDestinationName(outputBindingName); } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java index 96ebfab05..dfb9f4df8 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-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. @@ -17,8 +17,8 @@ package org.springframework.cloud.stream.function; import org.springframework.boot.context.properties.ConfigurationProperties; -import org.springframework.cloud.stream.binder.ConsumerProperties; -import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.cloud.stream.config.BindingServiceProperties; +import org.springframework.cloud.stream.messaging.Processor; /** * @@ -36,9 +36,11 @@ public class StreamFunctionProperties { */ private String definition; - private ConsumerProperties consumerProperties; + private BindingServiceProperties bindingServiceProperties; - private ProducerProperties producerProperties; + private String inputDestinationName = Processor.INPUT; + + private String outputDestinationName = Processor.OUTPUT; public String getDefinition() { return this.definition; @@ -48,19 +50,27 @@ public class StreamFunctionProperties { this.definition = definition; } - protected ConsumerProperties getConsumerProperties() { - return consumerProperties; + BindingServiceProperties getBindingServiceProperties() { + return bindingServiceProperties; } - void setConsumerProperties(ConsumerProperties consumerProperties) { - this.consumerProperties = consumerProperties; + void setBindingServiceProperties(BindingServiceProperties bindingServiceProperties) { + this.bindingServiceProperties = bindingServiceProperties; } - protected ProducerProperties getProducerProperties() { - return producerProperties; + String getInputDestinationName() { + return inputDestinationName; } - void setProducerProperties(ProducerProperties producerProperties) { - this.producerProperties = producerProperties; + void setInputDestinationName(String inputDestinationName) { + this.inputDestinationName = inputDestinationName; + } + + String getOutputDestinationName() { + return outputDestinationName; + } + + void setOutputDestinationName(String outputDestinationName) { + this.outputDestinationName = outputDestinationName; } } 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 51d16a350..b5f7bd906 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 @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-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. @@ -30,16 +30,22 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.catalog.FunctionInspector; -import org.springframework.cloud.stream.binder.ConsumerProperties; -import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamMessageConverter; +import org.springframework.cloud.stream.binder.test.InputDestination; +import org.springframework.cloud.stream.binder.test.OutputDestination; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; +import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.cloud.stream.function.pojo.Baz; import org.springframework.cloud.stream.function.pojo.ErrorBaz; +import org.springframework.cloud.stream.messaging.Processor; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.converter.MessageConverter; import org.springframework.messaging.support.GenericMessage; import org.springframework.util.ReflectionUtils; @@ -54,6 +60,112 @@ import static org.junit.Assert.assertEquals; */ public class FunctionInvokerTests { + @Test + public void testFunctionHonorsOutboundBindingContentType() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(ConverterDoesNotProduceCTConfiguration.class)).web( + WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=func", + "--spring.cloud.stream.bindings.output.contentType=text/plain")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + Message inputMessage = MessageBuilder.withPayload("{\"name\":\"bob\"}".getBytes()) + .setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertEquals("text/plain", outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()); + + } + } + + @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; + } + }; + } + } + + @Test + public void testFunctionHonorsConverterSetContentType() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(ConverterInjectingCTConfiguration.class)).web( + WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=func", + "--spring.cloud.stream.bindings.output.contentType=text/plain")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + Message inputMessage = MessageBuilder.withPayload("{\"name\":\"bob\"}".getBytes()) + .setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertEquals("ping/pong", outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()); + + } + } + + @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; + } + }; + } + } + @Test public void testSameMessageTypesAreNotConverted() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( @@ -176,12 +288,14 @@ public class FunctionInvokerTests { private StreamFunctionProperties createStreamFunctionProperties() { StreamFunctionProperties functionProperties = new StreamFunctionProperties(); - ConsumerProperties consumerProperties = new ConsumerProperties(); - consumerProperties.setMaxAttempts(3); + functionProperties.setInputDestinationName("input"); + functionProperties.setOutputDestinationName("output"); + BindingServiceProperties bindingServiceProperties = new BindingServiceProperties(); + bindingServiceProperties.getConsumerProperties("input").setMaxAttempts(3); try { - Field f = ReflectionUtils.findField(StreamFunctionProperties.class, "consumerProperties"); + Field f = ReflectionUtils.findField(StreamFunctionProperties.class, "bindingServiceProperties"); f.setAccessible(true); - f.set(functionProperties, consumerProperties); + f.set(functionProperties, bindingServiceProperties); return functionProperties; } catch (Exception e) { throw new IllegalStateException(e); @@ -190,17 +304,15 @@ public class FunctionInvokerTests { private StreamFunctionProperties createStreamFunctionPropertiesWithNativeEncoding() { StreamFunctionProperties functionProperties = new StreamFunctionProperties(); - ConsumerProperties consumerProperties = new ConsumerProperties(); - consumerProperties.setMaxAttempts(3); - ProducerProperties producerProperties = new ProducerProperties(); - producerProperties.setUseNativeEncoding(true); + functionProperties.setInputDestinationName("input"); + functionProperties.setOutputDestinationName("output"); + BindingServiceProperties bindingServiceProperties = new BindingServiceProperties(); + bindingServiceProperties.getConsumerProperties("input").setMaxAttempts(3); + bindingServiceProperties.getProducerProperties("output").setUseNativeEncoding(true); try { - Field c = ReflectionUtils.findField(StreamFunctionProperties.class, "consumerProperties"); - Field p = ReflectionUtils.findField(StreamFunctionProperties.class, "producerProperties"); - c.setAccessible(true); - c.set(functionProperties, consumerProperties); - p.setAccessible(true); - p.set(functionProperties, producerProperties); + Field bspField = ReflectionUtils.findField(StreamFunctionProperties.class, "bindingServiceProperties"); + bspField.setAccessible(true); + bspField.set(functionProperties, bindingServiceProperties); return functionProperties; } catch (Exception e) { throw new IllegalStateException(e); From 4de98320b206f1fdeec48adb3e5c5720e07ffc24 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 31 Jan 2019 08:17:10 +0100 Subject: [PATCH 13/16] GH-1595 Inherited @Bean from @StreamMessageConverter Resolves #1595 --- .../cloud/stream/annotation/StreamMessageConverter.java | 2 ++ .../cloud/stream/function/FunctionInvokerTests.java | 2 -- 2 files changed, 2 insertions(+), 2 deletions(-) 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 dd31a18ca..773b5b8f2 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 @@ -35,6 +36,7 @@ import org.springframework.beans.factory.annotation.Qualifier; @Retention(RetentionPolicy.RUNTIME) @Documented @Qualifier +@Bean public @interface StreamMessageConverter { } 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 b5f7bd906..af7b59b82 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 @@ -92,7 +92,6 @@ public class FunctionInvokerTests { } @StreamMessageConverter - @Bean public MessageConverter customConverter() { return new MessageConverter() { @@ -145,7 +144,6 @@ public class FunctionInvokerTests { } @StreamMessageConverter - @Bean public MessageConverter customConverter() { return new MessageConverter() { From e9d22cb352f21984313daaf175476e35ad945488 Mon Sep 17 00:00:00 2001 From: nicolas Date: Mon, 21 Jan 2019 01:47:23 -0500 Subject: [PATCH 14/16] RetryTemplate per binding --- .../cloud/stream/binder/AbstractBinder.java | 4 ++++ .../cloud/stream/binder/ConsumerProperties.java | 17 +++++++++++++++++ 2 files changed, 21 insertions(+) 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 60daa46ec..2478c5a3d 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 @@ -180,6 +180,10 @@ public abstract class AbstractBinder Date: Mon, 4 Feb 2019 06:51:19 +0100 Subject: [PATCH 15/16] GH-1511 Polishing and added tests - Polished RetryTemplate changes from the previous commit - Rebased with 2.2.x branch - Added tests - Updated author tags with original committer's name Resolves #1511 Resolves #1588 --- .../main/asciidoc/spring-cloud-stream.adoc | 9 ++ .../cloud/stream/binder/AbstractBinder.java | 22 ++-- .../stream/binder/ConsumerProperties.java | 19 ++- .../stream/config/RetryTemplateTests.java | 119 ++++++++++++++++++ 4 files changed, 150 insertions(+), 19 deletions(-) create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/RetryTemplateTests.java diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 1f4d223a1..d8f0d047f 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -1054,6 +1054,15 @@ 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`. +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. + +[source] +---- +spring.cloud.stream.bindings..consumer.retry-template-name= +---- + + [[spring-cloud-stream-overview-binders]] == Binders 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 2478c5a3d..0a15ecffb 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; @@ -49,6 +51,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 { @@ -65,9 +68,9 @@ public abstract class AbstractBinder consumerBindingRetryTemplates; /** * For binder implementations that support a prefix, apply the prefix to the name. @@ -179,12 +182,8 @@ public abstract class AbstractBinder 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(); + } + } +} From 5b39e57ac9086d19e53851eb4558fe0a83ad3357 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 4 Feb 2019 10:44:23 +0100 Subject: [PATCH 16/16] GH-1240 Additional fixes related to 5a9c6a616236f5101c2e0c15b750ee11ab62039b --- .../binder/AbstractMessageChannelBinder.java | 26 +++++++++---------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index 08d02146b..5755323cf 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -581,10 +581,8 @@ public abstract class AbstractMessageChannelBinder