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

GH-1559 Deprecated reactive support

Browse Source 
Removed reactive module documentation
Added initial documentation for reactive functions

Resolves #1559
This commit is contained in:
Oleg Zhurakousky
2019-01-18 15:54:45 +01:00
parent 0dad98e1e6
commit 6cc8356404
2 changed files with 56 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

@@ -659,7 +659,27 @@ public static class SinkFromConsumer {
}
}
----
===== Reactive Functions support
Since _Spring Cloud Function_ is build on top of https://projectreactor.io/[Project Reactor] there isn't much you need to do
to benefit from reactive programming model while implementing `Supplier`, `Function` or `Consumer`.
For example:
[source,java]
----
@EnableAutoConfiguration
@EnableBinding(Processor.class)
public static class SinkFromConsumer {
public static void main(String[] args) {
SpringApplication.run(SinkFromConsumer.class, "--spring.cloud.stream.function.definition=reactiveUpperCase");
}
@Bean
public Function<Flux<String>, Flux<String>> reactiveUpperCase() {
return flux -> flux.map(val -> val.toUpperCase());
}
}
----
===== Functional Composition
Using this programming model you can also benefit from functional composition where you can dynamically compose complex handlers from a set of simple functions.
@@ -676,7 +696,17 @@ To do that Spring Cloud Function allows you to use `|` (pipe) symbol. So to fini
[source,java]
----
—spring.cloud.stream.function.definition=toUpperCase|wrapInQuotes
--spring.cloud.stream.function.definition=toUpperCase|wrapInQuotes
----
NOTE: One of the great benefits of functional composition support provided by _Spring Cloud Function_ is
the fact that you can compose _reactive_ and _imperative_ functions.
For example, the above composition could be defined as such (if both functions present):
[source,java]
----
--spring.cloud.stream.function.definition=reactiveUpperCase|wrapInQuotes
----
@@ -1024,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<String> receive(@Input(Processor.INPUT) Flux<String> 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<String> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(s -> s.toUpperCase()));
}
}
----
===== Reactive Sources
Spring Cloud Stream reactive support also provides the ability for creating reactive sources through the `@StreamEmitter` annotation.
By using the `@StreamEmitter` annotation, a regular source may be converted to a reactive one.
`@StreamEmitter` is a method level annotation that marks a method to be an emitter to outputs declared with `@EnableBinding`.
You cannot use the `@Input` annotation along with `@StreamEmitter`, as the methods marked with this annotation are not listening for any input. Rather, methods marked with `@StreamEmitter` generate output.
Following the same programming model used in `@StreamListener`, `@StreamEmitter` also allows flexible ways of using the `@Output` annotation, depending on whether the method has any arguments, a return type, and other considerations.
The remainder of this section contains examples of using the `@StreamEmitter` annotation in various styles.
The following example emits the `Hello, World` message every millisecond and publishes to a Reactor `Flux`:
[source, java]
----
@EnableBinding(Source.class)
@EnableAutoConfiguration
public static class HelloWorldEmitter {
@StreamEmitter
@Output(Source.OUTPUT)
public Flux<String> emit() {
return Flux.intervalMillis(1)
.map(l -> "Hello World");
}
}
----
In the preceding example, the resulting messages in the `Flux` are sent to the output channel of the `Source`.
The next example is another flavor of an `@StreamEmmitter` that sends a Reactor `Flux`.
Instead of returning a `Flux`, the following method uses a `FluxSender` to programmatically send a `Flux` from a source:
[source, java]
----
@EnableBinding(Source.class)
@EnableAutoConfiguration
public static class HelloWorldEmitter {
@StreamEmitter
@Output(Source.OUTPUT)
public void emit(FluxSender output) {
output.send(Flux.intervalMillis(1)
.map(l -> "Hello World"));
}
}
----
The next example is exactly same as the above snippet in functionality and style.
However, instead of using an explicit `@Output` annotation on the method, it uses the annotation on the method parameter.
[source, java]
----
@EnableBinding(Source.class)
@EnableAutoConfiguration
public static class HelloWorldEmitter {
@StreamEmitter
public void emit(@Output(Source.OUTPUT) FluxSender output) {
output.send(Flux.intervalMillis(1)
.map(l -> "Hello World"));
}
}
----
The last example in this section is yet another flavor of writing reacting sources by using the Reactive Streams Publisher API and taking advantage of the support for it in https://github.com/spring-projects/spring-integration-java-dsl/wiki/Spring-Integration-Java-DSL-Reference[Spring Integration Java DSL].
The `Publisher` in the following example still uses Reactor `Flux` under the hood, but, from an application perspective, that is transparent to the user and only needs Reactive Streams and Java DSL for Spring Integration:
[source, java]
----
@EnableBinding(Source.class)
@EnableAutoConfiguration
public static class HelloWorldEmitter {
@StreamEmitter
@Output(Source.OUTPUT)
@Bean
public Publisher<Message<String>> emit() {
return IntegrationFlows.from(() ->
new GenericMessage<>("Hello World"),
e -> e.poller(p -> p.fixedDelay(1)))
.toReactivePublisher();
}
}
----
[[spring-cloud-stream-overview-binders]]
== Binders
@@ -1321,8 +1195,8 @@ spring:
rabbitmq:
host: <host2>
----
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 <<Using Polled Consumers>> 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.