From c56b4d1ea96a370b896886334ac72e734542e264 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 16 Sep 2019 11:51:21 +0200 Subject: [PATCH] Add initial support for functions wih multiple inputs/outputs as well as multiple functions - Currently there is no multi-out support for Supplier and I am not really sure what the use case would be so holding off. - The logic in FunctionConfiguration effectively split in bootstrapping simple functions (e.g., Function) vs. multi-in/out (e.g., Function, Flux>, Flux>). This is temporary given that multi-in/out is still considered WIP. Once it becomes stable we can merge the two two for consistency. - Added multiple input/output support for TestBinder - Updated documentation Resolves #1746 Resolves #1745 Resolves #1314 --- README.adoc | 41 +- docs/src/main/asciidoc/preface.adoc | 41 +- .../main/asciidoc/spring-cloud-stream.adoc | 1243 +++++++---------- .../BindableFunctionProxyFactory.java | 95 +- .../function/FunctionConfiguration.java | 459 +++--- .../binder/test/AbstractDestination.java | 15 +- .../stream/binder/test/InputDestination.java | 6 +- .../stream/binder/test/OutputDestination.java | 21 +- .../MultipleInputOutputFunctionTests.java | 292 ++++ .../SourceToFunctionsSupportTests.java | 1 - 10 files changed, 1169 insertions(+), 1045 deletions(-) create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/MultipleInputOutputFunctionTests.java diff --git a/README.adoc b/README.adoc index 967946c6a..968e5e27a 100644 --- a/README.adoc +++ b/README.adoc @@ -105,16 +105,17 @@ Modify the `com.example.loggingconsumer.LoggingConsumerApplication` class to loo [source, java] ---- @SpringBootApplication -@EnableBinding(Sink.class) public class LoggingConsumerApplication { public static void main(String[] args) { SpringApplication.run(LoggingConsumerApplication.class, args); } - @StreamListener(Sink.INPUT) - public void handle(Person person) { - System.out.println("Received: " + person); + @Bean + public Consumer log() { + return person -> { + System.out.println("Received: " + person); + }; } public static class Person { @@ -134,10 +135,10 @@ public class LoggingConsumerApplication { As you can see from the preceding listing: -* We have enabled `Sink` binding (input-no-output) by using `@EnableBinding(Sink.class)`. -Doing so signals to the framework to initiate binding to the messaging middleware, where it automatically creates the destination (that is, queue, topic, and others) that are bound to the `Sink.INPUT` channel. -* We have added a `handler` method to receive incoming messages of type `Person`. -Doing so lets you see one of the core features of the framework: It tries to automatically convert incoming message payloads to type `Person`. +* We are using functional programming model (see <>) to define a single message handler as `Consumer`. +* We are relying on framework conventions to bind such handler to the input destination binding exposed by the binder. + +Doing so also lets you see one of the core features of the framework: It tries to automatically convert incoming message payloads to type `Person`. You now have a fully functional Spring Cloud Stream application that does listens for messages. From here, for simplicity, we assume you selected RabbitMQ in <>. @@ -172,35 +173,21 @@ 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.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] - - -The following sections outline the most notable ones: - -* <> -* <> +== What's New in 3.0? +TBD [[spring-cloud-stream-preface-new-features]] === New Features and Components +TBD [[spring-cloud-stream-preface-notable-enhancements]] === Notable Enhancements - +TBD [[spring-cloud-stream-preface-notable-deprecations]] === Notable Deprecations - -As of version 2.2, the following items have been deprecated: - -- 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 -- Introduction of `@StreamRetryTemplate` qualifier. While configuring custom instance of the `RetryTemplate` and to avoid conflicts you must qualify the instance of such `RetryTemplate` with this qualifier. See <> for more details. +TBD = Appendices [appendix] diff --git a/docs/src/main/asciidoc/preface.adoc b/docs/src/main/asciidoc/preface.adoc index 51b2574b3..16bee295f 100644 --- a/docs/src/main/asciidoc/preface.adoc +++ b/docs/src/main/asciidoc/preface.adoc @@ -85,16 +85,17 @@ Modify the `com.example.loggingconsumer.LoggingConsumerApplication` class to loo [source, java] ---- @SpringBootApplication -@EnableBinding(Sink.class) public class LoggingConsumerApplication { public static void main(String[] args) { SpringApplication.run(LoggingConsumerApplication.class, args); } - @StreamListener(Sink.INPUT) - public void handle(Person person) { - System.out.println("Received: " + person); + @Bean + public Consumer log() { + return person -> { + System.out.println("Received: " + person); + }; } public static class Person { @@ -114,10 +115,10 @@ public class LoggingConsumerApplication { As you can see from the preceding listing: -* We have enabled `Sink` binding (input-no-output) by using `@EnableBinding(Sink.class)`. -Doing so signals to the framework to initiate binding to the messaging middleware, where it automatically creates the destination (that is, queue, topic, and others) that are bound to the `Sink.INPUT` channel. -* We have added a `handler` method to receive incoming messages of type `Person`. -Doing so lets you see one of the core features of the framework: It tries to automatically convert incoming message payloads to type `Person`. +* We are using functional programming model (see <>) to define a single message handler as `Consumer`. +* We are relying on framework conventions to bind such handler to the input destination binding exposed by the binder. + +Doing so also lets you see one of the core features of the framework: It tries to automatically convert incoming message payloads to type `Person`. You now have a fully functional Spring Cloud Stream application that does listens for messages. From here, for simplicity, we assume you selected RabbitMQ in <>. @@ -152,32 +153,18 @@ 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.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] - - -The following sections outline the most notable ones: - -* <> -* <> +== What's New in 3.0? +TBD [[spring-cloud-stream-preface-new-features]] === New Features and Components +TBD [[spring-cloud-stream-preface-notable-enhancements]] === Notable Enhancements - +TBD [[spring-cloud-stream-preface-notable-deprecations]] === Notable Deprecations - -As of version 2.2, the following items have been deprecated: - -- 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 -- Introduction of `@StreamRetryTemplate` qualifier. While configuring custom instance of the `RetryTemplate` and to avoid conflicts you must qualify the instance of such `RetryTemplate` with this qualifier. See <> for more details. +TBD diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 712833abc..4386a7da5 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -32,65 +32,46 @@ Spring Cloud Stream is a framework for building message-driven microservice appl Spring Cloud Stream builds upon Spring Boot to create standalone, production-grade Spring applications and uses Spring Integration to provide connectivity to message brokers. It provides opinionated configuration of middleware from several vendors, introducing the concepts of persistent publish-subscribe semantics, consumer groups, and partitions. -You can add the `@EnableBinding` annotation to your application to get immediate connectivity to a message broker, and you can add `@StreamListener` to a method to cause it to receive events for stream processing. -The following example shows a sink application that receives external messages: +By simply adding spring-cloud-stream dependencies to the classpath of your application you'll get immediate connectivity +to a message broker exposed via provided spring-cloud-stream binder (more on hat later), and you can implement your functional +requirement that will be executed based on the incoming message using simple `java.util.function.Function` [source,java] ---- @SpringBootApplication -@EnableBinding(Sink.class) -public class VoteRecordingSinkApplication { +public class SampleApplication { - public static void main(String[] args) { - SpringApplication.run(VoteRecordingSinkApplication.class, args); - } + public static void main(String[] args) { + SpringApplication.run(SampleApplication.class, args); + } - @StreamListener(Sink.INPUT) - public void processVote(Vote vote) { - votingService.recordVote(vote); - } + @Bean + public Function uppercase() { + return value -> { + System.out.println("Received: " + value); + return value.toUpperCase() + }; + } } ---- -The `@EnableBinding` annotation takes one or more interfaces as parameters (in this case, the parameter is a single `Sink` interface). -An interface declares input and output channels. -Spring Cloud Stream provides the `Source`, `Sink`, and `Processor` interfaces. You can also define your own interfaces. - -The following listing shows the definition of the `Sink` interface: - [source,java] ---- -public interface Sink { - String INPUT = "input"; +@Test +public void testRoutingViaExplicitEnablingAndDefinitionHeader() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + SampleApplication.class)).run()) { - @Input(Sink.INPUT) - SubscribableChannel input(); -} ----- + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); -The `@Input` annotation identifies an input channel, through which received messages enter the application. -The `@Output` annotation identifies an output channel, through which published messages leave the application. -The `@Input` and `@Output` annotations can take a channel name as a parameter. -If a name is not provided, the name of the annotated method is used. + Message inputMessage = new GenericMessage<>("Hello".getBytes()); + inputDestination.send(inputMessage); -Spring Cloud Stream creates an implementation of the interface for you. -You can use this in the application by autowiring it, as shown in the following example (from a test case): - -[source,java] ----- -@RunWith(SpringJUnit4ClassRunner.class) -@SpringApplicationConfiguration(classes = VoteRecordingSinkApplication.class) -@WebAppConfiguration -@DirtiesContext -public class StreamApplicationTests { - - @Autowired - private Sink sink; - - @Test - public void contextLoads() { - assertThat(this.sink.input()).isNotNull(); - } + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage.getPayload()).isEqualTo("HELLO".getBytes()); + } } ---- @@ -249,37 +230,43 @@ _Destination Binding_ does require special attention. The next section discusses As stated earlier, _Destination Bindings_ provide a bridge between the external messaging system and application-provided _Producers_ and _Consumers_. -Applying the @EnableBinding annotation to one of the application’s configuration classes defines a destination binding. -The `@EnableBinding` annotation itself is meta-annotated with `@Configuration` and triggers the configuration of the Spring Cloud Stream infrastructure. - -The following example shows a fully configured and functioning Spring Cloud Stream application that receives the payload of the message from the `INPUT` -destination as a `String` type (see <> section), logs it to the console and sends it to the `OUTPUT` destination after converting it to upper case. +The following example shows a fully configured and functioning Spring Cloud Stream application that receives the payload of the message +as a `String` type (see <> section), logs it to the console and sends it down stream after converting it to upper case. [source, java] ---- @SpringBootApplication -@EnableBinding(Processor.class) -public class MyApplication { +public class SampleApplication { public static void main(String[] args) { - SpringApplication.run(MyApplication.class, args); + SpringApplication.run(SampleApplication.class, args); } - @StreamListener(Processor.INPUT) - @SendTo(Processor.OUTPUT) - public String handle(String value) { - System.out.println("Received: " + value); - return value.toUpperCase(); + @Bean + public Function uppercase() { + return value -> { + System.out.println("Received: " + value); + return value.toUpperCase() + }; } } ---- +Unlike previous versions of spring-cloud-stream which relied on `@EnableBinding` and `@StreamListener` annotations, +the above example looks no different then any vanilla spring-boot application. It defines a single bean of type `Function` +and that it is. So, how does it became spring-cloud-stream application? +It became spring-cloud-stream application simply based on the presence of spring-cloud-stream and binder dependencies +and auto-configuration classes on the classpath, which by default look for beans of type `Supplier`, `Function` or `Consumer` +to bind to destinations exposed by the provided binder following certain naming conventions and +rules to avoid extra configuration. +More details are in the <> section, but to finish making sense of the above sample; +Assuming that spring-cloud-stream and binder dependencies are on the classpath, a single bean of type `Function` defined +in the above configuration' is treated as message handler and is bound to `"input"` and `"output"` _binding +destinations_ the identical way as you would explicitly do with `@StreamListener` in the previous versions of spring-cloud-stream. -As you can see the `@EnableBinding` annotation can take one or more interface classes as parameters. The parameters are referred to as _bindings_, -and they contain methods representing _bindable components_. -These components are typically message channels (see https://docs.spring.io/spring-boot/docs/current/reference/html/boot-features-messaging.html[Spring Messaging]) -for channel-based binders (such as Rabbit, Kafka, and others). However other types of bindings can -provide support for the native features of the corresponding technology. For example Kafka Streams binder (formerly known as KStream) allows native bindings directly to Kafka Streams -(see http://cloud.spring.io/spring-cloud-static/spring-cloud-stream-binder-kafka/{spring-cloud-stream-version}/[Kafka Binder] for more details). + +In previous versions of spring-cloud-stream _binding destinations_, mentioned in previous paragraph, derived from the `@EnableBinding` +annotation which typically would take one or more interface classes as parameters. The parameters are referred to +as _bindings_, and they contain methods representing _bindable components_. Spring Cloud Stream already provides _binding_ interfaces for typical message exchange contracts, which include: @@ -314,38 +301,22 @@ public interface Source { public interface Processor extends Source, Sink {} ---- -While the preceding example satisfies the majority of cases, you can also define your own contracts by defining your own bindings interfaces and use `@Input` and `@Output` -annotations to identify the actual _bindable components_. - -For example: - +And you can define your own interfaces as well [source, java] ---- -public interface Barista { +public interface MyBinding { - @Input - SubscribableChannel orders(); + String FOO = "foo"; - @Output - MessageChannel hotDrinks(); - - @Output - MessageChannel coldDrinks(); + @Output(MyBinding.FOO) + MessageChannel foo(); } ---- -Using the interface shown in the preceding example as a parameter to `@EnableBinding` triggers the creation of the three bound channels named `orders`, `hotDrinks`, and `coldDrinks`, -respectively. +NOTE: The reason why `@EnableBinding` and binding interfaces are not required with functional programming model is because +they could be derived from the type of functional interface itself. For example, _Processor = Function_, _Source = Supplierc_ +and so on. -You can provide as many binding interfaces as you need, as arguments to the `@EnableBinding` annotation, as shown in the following example: - -[source, java] ----- -@EnableBinding(value = { Orders.class, Payment.class }) ----- - -In Spring Cloud Stream, the bindable `MessageChannel` components are the Spring Messaging `MessageChannel` (for outbound) and its extension, `SubscribableChannel`, -(for inbound). *Pollable Destination Binding* @@ -367,67 +338,382 @@ public interface PolledBarista { In this case, an implementation of `PollableMessageSource` is bound to the `orders` “channel”. See <> for more details. -*Customizing Channel Names* - -By using the `@Input` and `@Output` annotations, you can specify a customized channel name for the channel, as shown in the following example: - -[source, java] ----- -public interface Barista { - @Input("inboundOrders") - SubscribableChannel orders(); -} ----- - -In the preceding example, the created bound channel is named `inboundOrders`. - -Normally, you need not access individual channels or bindings directly (other then configuring them via `@EnableBinding` annotation). However there may be -times, such as testing or other corner cases, when you do. - -Aside from generating channels for each binding and registering them as Spring beans, for each bound interface, Spring Cloud Stream generates a bean that implements the interface. -That means you can have access to the interfaces representing the bindings or individual channels by auto-wiring either in your application, as shown in the following two examples: - -_Autowire Binding interface_ - -[source, java] ----- -@Autowired -private Source source - -public void sayHello(String name) { - source.output().send(MessageBuilder.withPayload(name).build()); -} ----- - -_Autowire individual channel_ - -[source, java] ----- -@Autowired -private MessageChannel output; - -public void sayHello(String name) { - output.send(MessageBuilder.withPayload(name).build()); -} ----- - -You can also use standard Spring's `@Qualifier` annotation for cases when channel names are customized or in multiple-channel scenarios that require specifically named channels. - -The following example shows how to use the @Qualifier annotation in this way: - -[source, java] ----- -@Autowired -@Qualifier("myChannel") -private MessageChannel output; ----- [[spring-cloud-stream-overview-producing-consuming-messages]] === Producing and Consuming Messages -You can write a Spring Cloud Stream application by using either Spring Integration annotations or Spring Cloud Stream native annotation. +You can write a Spring Cloud Stream application by simply writing functions and exposing them as `@Bean`s. +You can also use Spring Integration annotations based configuration or +Spring Cloud Stream annotation based configuration, although starting with spring-cloud-stream 3.x +we recommend using functional implementations. -==== Spring Integration Support +[[spring_cloud_function]] +==== Spring Cloud Function support + +===== Overview + +Since Spring Cloud Stream v2.1, another alternative for defining _stream handlers_ and _sources_ is to use build-in +support for https://cloud.spring.io/spring-cloud-function/[Spring Cloud Function] where they can be expressed as beans of + type `java.util.function.[Supplier/Function/Consumer]`. + +To specify which functional bean to bind to the external destination(s) exposed by the bindings, you must provide `spring.cloud.stream.function.definition` or native to spring-cloud-function `spring.cloud.function.definition` property. + +Here is the example of the application exposing message handler as `java.util.function.Function` effectively supporting +_pass-thru_ semantics by acting as consumer and producer of data. +[source,java] +---- +@SpringBootApplication +public class MyFunctionBootApp { + + public static void main(String[] args) { + SpringApplication.run(MyFunctionBootApp.class); + } + + @Bean + public Function toUpperCase() { + return s -> s.toUpperCase(); + } +} +---- +In the above you we simply define a bean of type `java.util.function.Function` called _toUpperCase_ and identify it as a bean to be used as message handler +whose 'input' and 'output' must be bound to the external destinations exposed by the provided destination binder. + +Below are the examples of simple functional applications to support other semantics: + +Here is the example of a _source_ semantics exposed as `java.util.function.Supplier` +[source,java] +---- +@SpringBootApplication +public static class SourceFromSupplier { + + @Bean + public Supplier date() { + return () -> new Date(12345L); + } +} +---- + +Here is the example of a _sink semantics_ exposed as `java.util.function.Consumer` +[source,java] +---- +@SpringBootApplication +public static class SinkFromConsumer { + + @Bean + public Consumer sink() { + return System.out::println; + } +} +---- + +NOTE: We are using `--spring.cloud.function.definition` property to explicitly declare which function bean +we want to be bound to binding destinations. For cases when you only have single such bean it is not required +but for all other cases it is. + +===== Content-based routing with functions +Routing with functions can be achieved by relying on `RoutingFunction` available in Spring Cloud Function 3.0. All you need to do is enable it via +`--spring.cloud.stream.function.routing.enabled=true` application property. Once enabled `RoutingFunction` will be bound to input destination +receiving all the messages and route them to other functions based on the provided instruction. + +Instruction could be provided with individual messages as well as application properties. + +Here are couple of samples: + +***Using message headers*** +[source,java] +---- +@SpringBootApplication +public class SampleApplication { + + public static void main(String[] args) { + SpringApplication.run(SampleApplication.class, + "--spring.cloud.stream.function.routing.enabled=true"); + } + + @Bean + public Consumer even() { + return value -> { + System.out.println("EVEN: " + value); + }; + } + + @Bean + public Consumer odd() { + return value -> { + System.out.println("ODD: " + value); + }; + } +} +---- +By default `RoutingFunction` will look for `spring.cloud.function.definition` header and if it is found its value will be treated as routing instruction. +So in the above case the value of such header should be either `odd` or `even` (the name of the function beans) to route request to available functions. + +You can also use SpEL for more dynamic scenarios via `spring.cloud.function.routing-expression` header. +For example, +setting `spring.cloud.function.routing-expression` header to value `T(java.lang.System).currentTimeMillis() % 2 == 0 ? 'even' : 'odd'` will end up semi-randomly routing request to either `odd` or `even` functions. +Also, for SpEL, the _root object_ of the evaluation context is `Message` so you can do evaluation on individual headers (or message) as well `....routing-expression=headers['type']` + +***Using application properties*** + +The `spring.cloud.function.routing-expression` and/or `spring.cloud.function.definition` +can be passed as application properties (e.g., `spring.cloud.function.routing-expression=headers['type']`. + +Passing instructions via application properties is especially important for reactive functions since given that fact that reactive +function is only invoked once to pass the Publisher, so access to the individual items is limited. + +===== 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] +---- +@SpringBootApplication +public static class SinkFromConsumer { + + @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. +As an example let's add the following function bean to the application defined above +[source,java] +---- +@Bean +public Function wrapInQuotes() { + return s -> "\"" + s + "\""; +} +---- +and modify the `spring.cloud.function.definition` property to reflect your intention to compose a new function from both ‘toUpperCase’ and ‘wrapInQuotes’. +To do that Spring Cloud Function allows you to use `|` (pipe) symbol. So to finish our example our property will now look like this: + +[source,java] +---- +--spring.cloud.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.function.definition=reactiveUpperCase|wrapInQuotes +---- + +===== Functions with multiple input and output arguments + +Starting with version 3.0 spring-cloud-stream provides support for functions that +have multiple inputs and/or multiple outputs (return values). What does this actually mean and +what type of use cases it is targeting? + +* _Big Data: Imagine the source of data you're dealing with is highly un-organized and contains various types of data elements +(e.g., orders, transactions etc) and you effectively need to sort it out._ +* _Data aggregation: Another use case may require you to merge data elements from 2+ incoming _streams_. + +The above describes just a few use cases where you may need to use a single function to accept and/or produce +multiple _streams_ of data. And that is the type of use cases we are targeting here. + +Also, note a slightly different emphasis on the concept of _sreams_ here. The assumption is that such functions are only valuable +if they are given access to the actual streams of data (not the individual elements). So for that we are relying on +abstractions provided by https://projectreactor.io/[Project Reactor] (i.e., `Flux` and `Mono`) which is already available on the +classpath as part of the dependencies brought in by spring-cloud-functions. + +Another important aspect is representation of multiple input and outputs. While java provides +variety of different abstractions to represent _multiple of something_ those abstractions +are _a) unbounded_, _b) lack arity_ and _c) lack type information_ which are all important in this context. +As an example, let's look at `Collection` or an array which only allows us to +describe _multiple_ of a single type or up-cast everything to an `Object`, affecting transparent type conversion feature of +spring-cloud-stream and so on. + +So to accommodate all these requirements the initial support is relying on he signature which utilizes another abstraction +provided by _Project Reactor_ - Tuples. However, we are working on allowing a more flexible signatures. + +IMPORTANT: While simple function binding destinations are usually named `"input"` and `"output"` (see the next section for exception to that rule), +and for the most parts are hidden from the typical user's concerns, we can not rely on the same naming convention here. +So, this is where understanding of the naming convention for binding destinations is important. + +*Binding naming convention:* + +* input - ` + _in_ + ` +* output - ` + _out_ + ` + +Let's look at the few samples: + +[source,java] +---- +@SpringBootApplication +public class SampleApplication { + + @Bean + public Function, Flux>, Flux> gather() { + return tuple -> { + Flux stringStream = tuple.getT1(); + Flux intStream = tuple.getT2().map(i -> String.valueOf(i)); + return Flux.merge(stringStream, intStream); + }; + } +} +---- + +The above example demonstrates function which takes two inputs (first of type `String` and second of type `Integer`) +and produces a single output of type `String`. + +So, for the above example the two input bindings will be `gather_in_0` and `gather_in_1` and for consistency the +output binding also follows the same convention and is named `gather_out_0`. + + +Knowing that will allow you to set binding specific properties the same way you did with `@StreamListener`. +For example, the following will override content-type for `gather_in_0` binding: + +---- +--spring.cloud.stream.bindings.gather_in_0.content-type=text/plain +---- + + +[source,java] +---- +@SpringBootApplication +public class SampleApplication { + + @Bean + public static Function, Tuple2, Flux>> scatter() { + return flux -> { + Flux connectedFlux = flux.publish().autoConnect(2); + UnicastProcessor even = UnicastProcessor.create(); + UnicastProcessor odd = UnicastProcessor.create(); + Flux evenFlux = connectedFlux.filter(number -> number % 2 == 0).doOnNext(number -> even.onNext("EVEN: " + number)); + Flux oddFlux = connectedFlux.filter(number -> number % 2 != 0).doOnNext(number -> odd.onNext("ODD: " + number)); + + return Tuples.of(Flux.from(even).doOnSubscribe(x -> evenFlux.subscribe()), Flux.from(odd).doOnSubscribe(x -> oddFlux.subscribe())); + }; + } +} +---- + +The above example is somewhat of a the opposite from the previous sample and demonstrates function which +takes single input of type `Integer` and produces two outputs (both of type `String`). + +So, for the above example the input binding is `gather_in_0` and the +output bindings are `gather_out_0` and `gather_out_1`. + +And you test it with the following code: +[source,java] +---- +@Test +public void testSingleInputMultiOutput() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + SampleApplication.class)) + .run("--spring.cloud.function.definition=scatter")) { + context.getBean(InputDestination.class); + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + for (int i = 0; i < 10; i++) { + inputDestination.send(MessageBuilder.withPayload(String.valueOf(i).getBytes()).build()); + } + + int counter = 0; + for (int i = 0; i < 5; i++) { + Message even = outputDestination.receive(0, 0); + assertThat(even.getPayload()).isEqualTo(("EVEN: " + String.valueOf(counter++)).getBytes()); + Message odd = outputDestination.receive(0, 1); + assertThat(odd.getPayload()).isEqualTo(("ODD: " + String.valueOf(counter++)).getBytes()); + } + } +} +---- + +===== Multiple functions in a single application + +There may also be a need for grouping several message handlers in a single application. You would do so by +defining several functions. + +[source,java] +---- +@SpringBootApplication +public class SampleApplication { + + @Bean + public Function uppercase() { + return value -> value.toUpperCase(); + } + + @Bean + public Function reverse() { + return value -> new StringBuilder(value).reverse().toString(); + } +} +---- + +In the above example we have configuration which defines two functions `uppercase` and `reverse`. +So first, as mentioned before, we need to notice that there is a a conflict (more then one function) and therefore +we need to resolve it by providing `spring.cloud.function.definition` property pointing to the actual function +we want to bind. Except here we will use `;` delimiter to point to both functions (see test case below). + +As with functions with multiple inputs/outputs we can no longer rely on the naming convention for +destination bindings used by functions with single inputs/outputs. So we follow the same convention as +for functions with multiple inputs/outputs: + +* input - ` + _in_ + ` +* output - ` + _out_ + ` + +This means that the above configuration will result in the following destination bindings: +`uppercase_in_0`, `uppercase_out_0`, `reverse_in_0` and `reverse_out_0`. + +And you test it with the following code: +[source,java] +---- +@Test +public void testMultipleFunctions() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + ReactiveFunctionConfiguration.class)) + .run("--spring.cloud.function.definition=uppercase;reverse")) { + context.getBean(InputDestination.class); + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + Message inputMessage = MessageBuilder.withPayload("Hello".getBytes()).build(); + inputDestination.send(inputMessage, 0); + inputDestination.send(inputMessage, 1); + + Message outputMessage = outputDestination.receive(0, 0); + assertThat(outputMessage.getPayload()).isEqualTo("HELLO".getBytes()); + + outputMessage = outputDestination.receive(0, 1); + assertThat(outputMessage.getPayload()).isEqualTo("olleH".getBytes()); + } +} +---- + +===== Batch Consumers + +When using a `MessageChannelBinder` that supports batch listeners, and the feature is enabled for the consumer binding, you can set `spring.cloud.stream.function.definition` to `true` to enable the entire batch of messages to be passed to the function in a `List`. + +[source, java] +---- +@Bean +public Function, Person> findFirstPerson() { + return persons -> persons.get(0); +} +---- + +==== Annotation-based support +As mentioned earlier you can also use Spring Integration annotations based configuration or +Spring Cloud Stream annotation based configuration. + +===== Spring Integration Support Spring Cloud Stream is built on the concepts and patterns defined by http://www.enterpriseintegrationpatterns.com/[Enterprise Integration Patterns] and relies in its internal implementation on an already established and popular implementation of Enterprise Integration Patterns within the Spring portfolio of projects: @@ -468,7 +754,7 @@ Each method annotated with `@StreamListener` receives its own copy of a message, However, if you consume from the same binding by using one of the Spring Integration annotation (such as `@Aggregator`, `@Transformer`, or `@ServiceActivator`), those consume in a competing model. No individual consumer group is created for each subscription. -==== Using @StreamListener Annotation +===== Using @StreamListener Annotation Complementary to its Spring Integration support, Spring Cloud Stream provides its own `@StreamListener` annotation, modeled after other Spring Messaging annotations (`@MessageMapping`, `@JmsListener`, `@RabbitListener`, and others) and provides conviniences, such as content-based routing and others. @@ -526,7 +812,7 @@ NOTE: Spring Cloud Stream does NOT provide a default `org.springframework.valida conflicts with validators provided by other frameworks that may be part of your application (e.g., MVC), therefore you may need to provide your own validator by configuring a bean of type `org.springframework.validation.Validator`. -==== Using @StreamListener for Content-based routing +===== Using @StreamListener for Content-based routing Spring Cloud Stream supports dispatching messages to multiple handler methods annotated with `@StreamListener` based on conditions. @@ -598,197 +884,6 @@ NOTE: At the moment, dispatching through `@StreamListener` conditions is support support. -[[spring_cloud_function]] -==== Spring Cloud Function support - -===== Overview - -Since Spring Cloud Stream v2.1, another alternative for defining _stream handlers_ and _sources_ is to use build-in -support for https://cloud.spring.io/spring-cloud-function/[Spring Cloud Function] where they can be expressed as beans of - type `java.util.function.[Supplier/Function/Consumer]`. - -To specify which functional bean to bind to the external destination(s) exposed by the bindings, you must provide `spring.cloud.stream.function.definition` or native to spring-cloud-function `spring.cloud.function.definition` property. - -Here is the example of the Processor application exposing message handler as `java.util.function.Function` -[source,java] ----- -@SpringBootApplication -@EnableBinding(Processor.class) -public class MyFunctionBootApp { - - public static void main(String[] args) { - SpringApplication.run(MyFunctionBootApp.class, "--spring.cloud.function.definition=toUpperCase"); - } - - @Bean - public Function toUpperCase() { - return s -> s.toUpperCase(); - } -} ----- -In the above you we simply define a bean of type `java.util.function.Function` called _toUpperCase_ and identify it as a bean to be used as message handler -whose 'input' and 'output' must be bound to the external destinations exposed by the Processor binding. - -Below are the examples of simple functional applications to support Source, Processor and Sink. - -Here is the example of a Source application defined as `java.util.function.Supplier` -[source,java] ----- -@SpringBootApplication -@EnableBinding(Source.class) -public static class SourceFromSupplier { - public static void main(String[] args) { - SpringApplication.run(SourceFromSupplier.class, "--spring.cloud.stream.function.definition=date"); - } - @Bean - public Supplier date() { - return () -> new Date(12345L); - } -} ----- - -Here is the example of a Processor application defined as `java.util.function.Function` -[source,java] ----- -@SpringBootApplication -public static class ProcessorFromFunction { - public static void main(String[] args) { - SpringApplication.run(ProcessorFromFunction.class, "--spring.cloud.stream.function.definition=toUpperCase"); - } - @Bean - public Function toUpperCase() { - return s -> s.toUpperCase(); - } -} ----- - -Here is the example of a Sink application defined as `java.util.function.Consumer` -[source,java] ----- -@EnableAutoConfiguration -public static class SinkFromConsumer { - public static void main(String[] args) { - SpringApplication.run(SinkFromConsumer.class, "--spring.cloud.stream.function.definition=sink"); - } - @Bean - public Consumer sink() { - return System.out::println; - } -} ----- -===== Content-based routing with functions -Routing with functions can achieved by relying on `RoutingFunction` available in Spring Cloud Function 3.0. All you need to do is enable it via -`--spring.cloud.stream.function.routing.enabled=true` application property. Once enabled `RoutingFunction` will be bound to input destination -receiving all the messages and route them to other functions based on the provided instruction. - -Instruction could be provided with individual messages as well as application properties. - -Here are couple of samples: - -***Using message headers*** -[source,java] ----- -@SpringBootApplication -public class SampleApplication { - - public static void main(String[] args) { - SpringApplication.run(SampleApplication.class, - "--spring.cloud.stream.function.routing.enabled=true"); - } - - @Bean - public Consumer even() { - return value -> { - System.out.println("EVEN: " + value); - }; - } - - @Bean - public Consumer odd() { - return value -> { - System.out.println("ODD: " + value); - }; - } -} ----- -By default `RoutingFunction` will look for `spring.cloud.function.definition` header and if it is found its value will be treated as routing instruction. -So in the above case the value of such header should be either `odd` or `even` (the name of the function beans) to route request to available functions. - -You can also use SpEL for more dynamic scenarios via `spring.cloud.function.routing-expression` header. -For example, -setting `spring.cloud.function.routing-expression` header to value `T(java.lang.System).currentTimeMillis() % 2 == 0 ? 'even' : 'odd'` will end up semi-randomly routing request to either `odd` or `even` functions. -Also, for SpEL, the _root object_ of the evaluation context is `Message` so you can do evaluation on individual headers (or message) as well `....routing-expression=headers['type']` - -***Using application properties*** - -The `spring.cloud.function.routing-expression` and/or `spring.cloud.function.definition` -can be passed as application properties (e.g., `spring.cloud.function.routing-expression=headers['type']`. - -Passing instructions via application properties is especially important for reactive functions since given that fact that reactive -function is only invoked once to pass the Publisher, so access to the individual items is limited. - -===== 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 -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. -As an example let's add the following function bean to the application defined above -[source,java] ----- -@Bean -public Function wrapInQuotes() { - return s -> "\"" + s + "\""; -} ----- -and modify the `spring.cloud.stream.function.definition` property to reflect your intention to compose a new function from both ‘toUpperCase’ and ‘wrapInQuotes’. -To do that Spring Cloud Function allows you to use `|` (pipe) symbol. So to finish our example our property will now look like this: - -[source,java] ----- ---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 ----- - -===== Batch Consumers - -When using a `MessageChannelBinder` that supports batch listeners, and the feature is enabled for the consumer binding, you can set `spring.cloud.stream.function.definition` to `true` to enable the entire batch of messages to be passed to the function in a `List`. - -==== -[source, java] ----- -@Bean -public Function, Person> findFirstPerson() { - return persons -> persons.get(0); -} ----- -==== [[spring-cloud-streams-overview-using-polled-consumers]] ==== Using Polled Consumers @@ -1790,12 +1885,11 @@ To better understand the mechanics and the necessity behind content-type negotia [source, java] ---- -@StreamListener(Processor.INPUT) -@SendTo(Processor.OUTPUT) -public String handle(Person person) {..} + +public Function personFunction {..} ---- -NOTE: For simplicity, we assume that this is the only handler in the application (we assume there is no internal pipeline). +NOTE: For simplicity, we assume that this is the only handler function in the application (we assume there is no internal pipeline). The handler shown in the preceding example expects a `Person` object as an argument and produces a `String` type as an output. In order for the framework to succeed in passing the incoming `Message` as an argument to this handler, it has to somehow transform the payload of the `Message` type from the wire format to a `Person` type. @@ -1887,12 +1981,9 @@ As mentioned earlier, the framework already provides a stack of `MessageConverte The following list describes the provided `MessageConverters`, in order of precedence (the first `MessageConverter` that works is used): . `ApplicationJsonMessageMarshallingConverter`: Variation of the `org.springframework.messaging.converter.MappingJackson2MessageConverter`. Supports conversion of the payload of the `Message` to/from POJO for cases when `contentType` is `application/json` (DEFAULT). -. `TupleJsonMessageConverter`: *DEPRECATED* Supports conversion of the payload of the `Message` to/from `org.springframework.tuple.Tuple`. . `ByteArrayMessageConverter`: Supports conversion of the payload of the `Message` from `byte[]` to `byte[]` for cases when `contentType` is `application/octet-stream`. It is essentially a pass through and exists primarily for backward compatibility. . `ObjectStringMessageConverter`: Supports conversion of any type to a `String` when `contentType` is `text/plain`. It invokes Object’s `toString()` method or, if the payload is `byte[]`, a new `String(byte[])`. -. `JavaSerializationMessageConverter`: *DEPRECATED* Supports conversion based on java serialization when `contentType` is `application/x-java-serialized-object`. -. `KryoMessageConverter`: *DEPRECATED* Supports conversion based on Kryo serialization when `contentType` is `application/x-java-object`. . `JsonUnmarshallingConverter`: Similar to the `ApplicationJsonMessageMarshallingConverter`. It supports conversion of any type when `contentType` is `application/x-java-object`. It expects the actual type information to be embedded in the `contentType` as an attribute (for example, `application/x-java-object;type=foo.bar.Cat`). @@ -1904,8 +1995,8 @@ does not know how to convert. If that is the case, you can add custom `MessageCo === User-defined Message Converters Spring Cloud Stream exposes a mechanism to define and register additional `MessageConverters`. -To use it, implement `org.springframework.messaging.converter.MessageConverter`, configure it as a `@Bean`, and annotate it with `@StreamMessageConverter`. -It is then apended to the existing stack of `MessageConverter`s. +To use it, implement `org.springframework.messaging.converter.MessageConverter`, configure it as a `@Bean`. +It is then appended to the existing stack of `MessageConverter`s. NOTE: It is important to understand that custom `MessageConverter` implementations are added to the head of the existing stack. Consequently, custom `MessageConverter` implementations take precedence over the existing ones, which lets you override as well as add to the existing converters. @@ -1914,14 +2005,12 @@ The following example shows how to create a message converter bean to support a [source,java] ---- -@EnableBinding(Sink.class) @SpringBootApplication public static class SinkApplication { ... @Bean - @StreamMessageConverter public MessageConverter customMessageConverter() { return new MyCustomMessageConverter(); } @@ -1949,346 +2038,7 @@ public class MyCustomMessageConverter extends AbstractMessageConverter { Spring Cloud Stream also provides support for Avro-based converters and schema evolution. See `<>` for details. -[[schema-evolution]] -== Schema Evolution Support - -Spring Cloud Stream provides support for schema evolution so that the data can be evolved over time and still work with older or newer producers and consumers and vice versa. -Most serialization models, especially the ones that aim for portability across different platforms and languages, rely on a schema that describes how the data is serialized in the binary payload. -In order to serialize the data and then to interpret it, both the sending and receiving sides must have access to a schema that describes the binary format. -In certain cases, the schema can be inferred from the payload type on serialization or from the target type on deserialization. -However, many applications benefit from having access to an explicit schema that describes the binary data format. -A schema registry lets you store schema information in a textual format (typically JSON) and makes that information accessible to various applications that need it to receive and send data in binary format. -A schema is referenceable as a tuple consisting of: - -* A subject that is the logical name of the schema -* The schema version -* The schema format, which describes the binary format of the data - -This following sections goes through the details of various components involved in schema evolution process. - -=== Schema Registry Client - -The client-side abstraction for interacting with schema registry servers is the `SchemaRegistryClient` interface, which has the following structure: - -[source,java] ----- -public interface SchemaRegistryClient { - - SchemaRegistrationResponse register(String subject, String format, String schema); - - String fetch(SchemaReference schemaReference); - - String fetch(Integer id); - -} ----- - -Spring Cloud Stream provides out-of-the-box implementations for interacting with its own schema server and for interacting with the Confluent Schema Registry. - -A client for the Spring Cloud Stream schema registry can be configured by using the `@EnableSchemaRegistryClient`, as follows: - -[source,java] ----- - @EnableBinding(Sink.class) - @SpringBootApplication - @EnableSchemaRegistryClient - public static class AvroSinkApplication { - ... - } ----- - -NOTE: The default converter is optimized to cache not only the schemas from the remote server but also the `parse()` and `toString()` methods, which are quite expensive. -Because of this, it uses a `DefaultSchemaRegistryClient` that does not cache responses. -If you intend to change the default behavior, you can use the client directly on your code and override it to the desired outcome. -To do so, you have to add the property `spring.cloud.stream.schemaRegistryClient.cached=true` to your application properties. - -==== Schema Registry Client Properties - -The Schema Registry Client supports the following properties: - -`spring.cloud.stream.schemaRegistryClient.endpoint`:: The location of the schema-server. -When setting this, use a full URL, including protocol (`http` or `https`) , port, and context path. -+ -Default:: `http://localhost:8990/` -`spring.cloud.stream.schemaRegistryClient.cached`:: Whether the client should cache schema server responses. -Normally set to `false`, as the caching happens in the message converter. -Clients using the schema registry client should set this to `true`. -+ -Default:: `false` - -=== Avro Schema Registry Client Message Converters - -For applications that have a SchemaRegistryClient bean registered with the application context, Spring Cloud Stream auto configures an Apache Avro message converter for schema management. -This eases schema evolution, as applications that receive messages can get easy access to a writer schema that can be reconciled with their own reader schema. - -For outbound messages, if the content type of the channel is set to `application/*+avro`, the `MessageConverter` is activated, as shown in the following example: - -[source,properties] ----- -spring.cloud.stream.bindings.output.contentType=application/*+avro ----- - -During the outbound conversion, the message converter tries to infer the schema of each outbound messages (based on its type) and register it to a subject (based on the payload type) by using the `SchemaRegistryClient`. -If an identical schema is already found, then a reference to it is retrieved. -If not, the schema is registered, and a new version number is provided. -The message is sent with a `contentType` header by using the following scheme: `application/[prefix].[subject].v[version]+avro`, where `prefix` is configurable and `subject` is deduced from the payload type. - -For example, a message of the type `User` might be sent as a binary payload with a content type of `application/vnd.user.v2+avro`, where `user` is the subject and `2` is the version number. - -When receiving messages, the converter infers the schema reference from the header of the incoming message and tries to retrieve it. The schema is used as the writer schema in the deserialization process. - -==== Avro Schema Registry Message Converter Properties - -If you have enabled Avro based schema registry client by setting `spring.cloud.stream.bindings.output.contentType=application/*+avro`, you can customize the behavior of the registration by setting the following properties. - -spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled:: Enable if you want the converter to use reflection to infer a Schema from a POJO. -+ -Default: `false` -+ -spring.cloud.stream.schema.avro.readerSchema:: Avro compares schema versions by looking at a writer schema (origin payload) and a reader schema (your application payload). See the https://avro.apache.org/docs/1.7.6/spec.html[Avro documentation] for more information. If set, this overrides any lookups at the schema server and uses the local schema as the reader schema. -Default: `null` -+ -spring.cloud.stream.schema.avro.schemaLocations:: Registers any `.avsc` files listed in this property with the Schema Server. -+ -Default: `empty` -+ -spring.cloud.stream.schema.avro.prefix:: The prefix to be used on the Content-Type header. -+ -Default: `vnd` -spring.cloud.stream.schema.avro.subjectNamingStrategy:: Determines the subject name used to register the Avro schema in the schema registry. Two implementations are available, `org.springframework.cloud.stream.schema.avro.DefaultSubjectNamingStrategy`, -where the subject is the schema name, and `org.springframework.cloud.stream.schema.avro.QualifiedSubjectNamingStrategy`, which returns a fully qualified subject using the Avro schema namespace and name. Custom strategies can be created by implementing `org.springframework.cloud.stream.schema.avro.SubjectNamingStrategy`. -+ -Default: `org.springframework.cloud.stream.schema.avro.DefaultSubjectNamingStrategy` - -=== Apache Avro Message Converters - -Spring Cloud Stream provides support for schema-based message converters through its `spring-cloud-stream-schema` module. -Currently, the only serialization format supported out of the box for schema-based message converters is Apache Avro, with more formats to be added in future versions. - -The `spring-cloud-stream-schema` module contains two types of message converters that can be used for Apache Avro serialization: - -* Converters that use the class information of the serialized or deserialized objects or a schema with a location known at startup. -* Converters that use a schema registry. They locate the schemas at runtime and dynamically register new schemas as domain objects evolve. - -=== Converters with Schema Support - -The `AvroSchemaMessageConverter` supports serializing and deserializing messages either by using a predefined schema or by using the schema information available in the class (either reflectively or contained in the `SpecificRecord`). -If you provide a custom converter, then the default AvroSchemaMessageConverter bean is not created. The following example shows a custom converter: - -To use custom converters, you can simply add it to the application context, optionally specifying one or more `MimeTypes` with which to associate it. -The default `MimeType` is `application/avro`. - -If the target type of the conversion is a `GenericRecord`, a schema must be set. - -The following example shows how to configure a converter in a sink application by registering the Apache Avro `MessageConverter` without a predefined schema. -In this example, note that the mime type value is `avro/bytes`, not the default `application/avro`. - -[source,java] ----- -@EnableBinding(Sink.class) -@SpringBootApplication -public static class SinkApplication { - - ... - - @Bean - public MessageConverter userMessageConverter() { - return new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes")); - } -} ----- - -Conversely, the following application registers a converter with a predefined schema (found on the classpath): - -[source,java] ----- -@EnableBinding(Sink.class) -@SpringBootApplication -public static class SinkApplication { - - ... - - @Bean - public MessageConverter userMessageConverter() { - AvroSchemaMessageConverter converter = new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes")); - converter.setSchemaLocation(new ClassPathResource("schemas/User.avro")); - return converter; - } -} ----- - -=== Schema Registry Server - -Spring Cloud Stream provides a schema registry server implementation. -To use it, you can add the `spring-cloud-stream-schema-server` artifact to your project and use the `@EnableSchemaRegistryServer` annotation, which adds the schema registry server REST controller to your application. -This annotation is intended to be used with Spring Boot web applications, and the listening port of the server is controlled by the `server.port` property. -The `spring.cloud.stream.schema.server.path` property can be used to control the root path of the schema server (especially when it is embedded in other applications). -The `spring.cloud.stream.schema.server.allowSchemaDeletion` boolean property enables the deletion of a schema. By default, this is disabled. - -The schema registry server uses a relational database to store the schemas. -By default, it uses an embedded database. -You can customize the schema storage by using the http://docs.spring.io/spring-boot/docs/current-SNAPSHOT/reference/htmlsingle/#boot-features-sql[Spring Boot SQL database and JDBC configuration options]. - -The following example shows a Spring Boot application that enables the schema registry: - -[source,java] ----- -@SpringBootApplication -@EnableSchemaRegistryServer -public class SchemaRegistryServerApplication { - public static void main(String[] args) { - SpringApplication.run(SchemaRegistryServerApplication.class, args); - } -} ----- - -==== Schema Registry Server API - -The Schema Registry Server API consists of the following operations: - -* `POST /` -- see `<>` -* 'GET /{subject}/{format}/{version}' -- see `<>` -* `GET /{subject}/{format}` -- see `<>` -* `GET /schemas/{id}` -- see `<>` -* `DELETE /{subject}/{format}/{version}` -- see `<>` -* `DELETE /schemas/{id}` -- see `<>` -* `DELETE /{subject}` -- see `<>` - -[[spring-cloud-stream-overview-registering-new-schema]] -===== Registering a New Schema - -To register a new schema, send a `POST` request to the `/` endpoint. - -The `/` accepts a JSON payload with the following fields: - -* `subject`: The schema subject -* `format`: The schema format -* `definition`: The schema definition - -Its response is a schema object in JSON, with the following fields: - -* `id`: The schema ID -* `subject`: The schema subject -* `format`: The schema format -* `version`: The schema version -* `definition`: The schema definition - -[[spring-cloud-stream-overview-retrieve-schema-subject-format-version]] -===== Retrieving an Existing Schema by Subject, Format, and Version - -To retrieve an existing schema by subject, format, and version, send `GET` request to the `/{subject}/{format}/{version}` endpoint. - -Its response is a schema object in JSON, with the following fields: - -* `id`: The schema ID -* `subject`: The schema subject -* `format`: The schema format -* `version`: The schema version -* `definition`: The schema definition - -[[spring-cloud-stream-overview-retrieve-schema-subject-format]] -===== Retrieving an Existing Schema by Subject and Format - -To retrieve an existing schema by subject and format, send a `GET` request to the `/subject/format` endpoint. - -Its response is a list of schemas with each schema object in JSON, with the following fields: - -* `id`: The schema ID -* `subject`: The schema subject -* `format`: The schema format -* `version`: The schema version -* `definition`: The schema definition - -[[spring-cloud-stream-overview-retrieve-schema-id]] -===== Retrieving an Existing Schema by ID - -To retrieve a schema by its ID, send a `GET` request to the `/schemas/{id}` endpoint. - -Its response is a schema object in JSON, with the following fields: - -* `id`: The schema ID -* `subject`: The schema subject -* `format`: The schema format -* `version`: The schema version -* `definition`: The schema definition - -[[spring-cloud-stream-overview-deleting-schema-subject-format-version]] -===== Deleting a Schema by Subject, Format, and Version - -To delete a schema identified by its subject, format, and version, send a `DELETE` request to the `/{subject}/{format}/{version}` endpoint. - -[[spring-cloud-stream-overview-deleting-schema-id]] -===== Deleting a Schema by ID - -To delete a schema by its ID, send a `DELETE` request to the `/schemas/{id}` endpoint. - -[[spring-cloud-stream-overview-deleting-schema-subject]] -===== Deleting a Schema by Subject -`DELETE /{subject}` - -Delete existing schemas by their subject. - -NOTE: This note applies to users of Spring Cloud Stream 1.1.0.RELEASE only. -Spring Cloud Stream 1.1.0.RELEASE used the table name, `schema`, for storing `Schema` objects. `Schema` is a keyword in a number of database implementations. -To avoid any conflicts in the future, starting with 1.1.1.RELEASE, we have opted for the name `SCHEMA_REPOSITORY` for the storage table. -Any Spring Cloud Stream 1.1.0.RELEASE users who upgrade should migrate their existing schemas to the new table before upgrading. - -==== Using Confluent's Schema Registry - -The default configuration creates a `DefaultSchemaRegistryClient` bean. -If you want to use the Confluent schema registry, you need to create a bean of type `ConfluentSchemaRegistryClient`, which supersedes the one configured by default by the framework. The following example shows how to create such a bean: - -[source,java] ----- -@Bean -public SchemaRegistryClient schemaRegistryClient(@Value("${spring.cloud.stream.schemaRegistryClient.endpoint}") String endpoint){ - ConfluentSchemaRegistryClient client = new ConfluentSchemaRegistryClient(); - client.setEndpoint(endpoint); - return client; -} ----- -NOTE: The ConfluentSchemaRegistryClient is tested against Confluent platform version 4.0.0. - -=== Schema Registration and Resolution - -To better understand how Spring Cloud Stream registers and resolves new schemas and its use of Avro schema comparison features, we provide two separate subsections: - -* `<>` -* `<>` - -[[spring-cloud-stream-overview-schema-registration-process]] -==== Schema Registration Process (Serialization) - -The first part of the registration process is extracting a schema from the payload that is being sent over a channel. -Avro types such as `SpecificRecord` or `GenericRecord` already contain a schema, which can be retrieved immediately from the instance. -In the case of POJOs, a schema is inferred if the `spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled` property is set to `true` (the default). - -.Schema Writer Resolution Process -image::{github-raw}/docs/src/main/asciidoc/images/schema_resolution.png[width=800,scaledwidth="75%",align="center"] - -Ones a schema is obtained, the converter loads its metadata (version) from the remote server. -First, it queries a local cache. If no result is found, it submits the data to the server, which replies with versioning information. -The converter always caches the results to avoid the overhead of querying the Schema Server for every new message that needs to be serialized. - -.Schema Registration Process -image::{github-raw}/docs/src/main/asciidoc/images/registration.png[width=800,scaledwidth="75%",align="center"] - -With the schema version information, the converter sets the `contentType` header of the message to carry the version information -- for example: `application/vnd.user.v1+avro`. - -[[spring-cloud-stream-overview-schema-resolution-process]] -==== Schema Resolution Process (Deserialization) - -When reading messages that contain version information (that is, a `contentType` header with a scheme like the one described under `<>`, the converter queries the Schema server to fetch the writer schema of the message. -Once it has found the correct schema of the incoming message, it retrieves the reader schema and, by using Avro's schema resolution support, reads it into the reader definition (setting defaults and any missing properties). - -.Schema Reading Resolution Process -image::{github-raw}/docs/src/main/asciidoc/images/schema_reading.png[width=800,scaledwidth="75%",align="center"] - -NOTE: You should understand the difference between a writer schema (the application that wrote the message) and a reader schema (the receiving application). -We suggest taking a moment to read https://avro.apache.org/docs/1.7.6/spec.html[the Avro terminology] and understand the process. -Spring Cloud Stream always fetches the writer schema to determine how to read a message. -If you want to get Avro's schema evolution support working, you need to make sure that a `readerSchema` was properly set for your application. - +[ == Inter-Application Communication Spring Cloud Stream enables communication between applications. Inter-application communication is a complex issue spanning several concerns, as described in the following topics: @@ -2416,98 +2166,13 @@ While a scenario in which using multiple instances for partitioned data processi == Testing Spring Cloud Stream provides support for testing your microservice applications without connecting to a messaging system. -You can do that by using the `TestSupportBinder` provided by the `spring-cloud-stream-test-support` library, which can be added as a test dependency to the application, as shown in the following example: -[source,xml] ----- - - org.springframework.cloud - spring-cloud-stream-test-support - test - ----- - -NOTE: The `TestSupportBinder` uses the Spring Boot autoconfiguration mechanism to supersede the other binders found on the classpath. -Therefore, when adding a binder as a dependency, you must make sure that the `test` scope is being used. - -The `TestSupportBinder` lets you interact with the bound channels and inspect any messages sent and received by the application. - -For outbound message channels, the `TestSupportBinder` registers a single subscriber and retains the messages emitted by the application in a `MessageCollector`. -They can be retrieved during tests and have assertions made against them. - -You can also send messages to inbound message channels so that the consumer application can consume the messages. -The following example shows how to test both input and output channels on a processor: - -[source,java] ----- -@RunWith(SpringRunner.class) -@SpringBootTest(webEnvironment= SpringBootTest.WebEnvironment.RANDOM_PORT) -public class ExampleTest { - - @Autowired - private Processor processor; - - @Autowired - private MessageCollector messageCollector; - - @Test - @SuppressWarnings("unchecked") - public void testWiring() { - Message message = new GenericMessage<>("hello"); - processor.input().send(message); - Message received = (Message) messageCollector.forChannel(processor.output()).poll(); - assertThat(received.getPayload(), equalTo("hello world")); - } - - - @SpringBootApplication - @EnableBinding(Processor.class) - public static class MyProcessor { - - @Autowired - private Processor channels; - - @Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT) - public String transform(String in) { - return in + " world"; - } - } -} ----- - -In the preceding example, we create an application that has an input channel and an output channel, both bound through the `Processor` interface. -The bound interface is injected into the test so that we can have access to both channels. -We send a message on the input channel, and we use the `MessageCollector` provided by Spring Cloud Stream's test support to capture that the message has been sent to the output channel as a result. -Once we have received the message, we can validate that the component functions correctly. - - -=== Disabling the Test Binder Autoconfiguration - -The intent behind the test binder superseding all the other binders on the classpath is to make it easy to test your applications without making changes to your production dependencies. -In some cases (for example, integration tests) it is useful to use the actual production binders instead, and that requires disabling the test binder autoconfiguration. -To do so, you can exclude the `org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration` class by using one of the Spring Boot autoconfiguration exclusion mechanisms, as shown in the following example: - -[source,java] ----- - @SpringBootApplication(exclude = TestSupportBinderAutoConfiguration.class) - @EnableBinding(Processor.class) - public static class MyProcessor { - - @Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT) - public String transform(String in) { - return in + " world"; - } - } ----- - -When autoconfiguration is disabled, the test binder is available on the classpath, and its `defaultCandidate` property is set to `false` so that it does not interfere with the regular user configuration. It can be referenced under the name, `test`, as shown in the following example: - -`spring.cloud.stream.defaultBinder=test` [[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). +The old test binder defined in `spring-cloud-stream-test-support` module 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). So we are effectively deprecating it. 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. @@ -2568,6 +2233,31 @@ To avoid conflicts with the existing test binder you must eremove the following Now you can test your microservice as a simple unit test +[source,java] +---- +@SpringBootApplication +public class DemoTestBinderApplication { + public Function echo() { + return value -> value; + } +} + +. . . + +@Test +public void sampleTest() { + ApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.class, + DemoTestBinderApplication.class) + .web(WebApplicationType.NONE).run("--spring.cloud.function.definition=echo"); + InputDestination source = context.getBean(InputDestination.class); + OutputDestination target = context.getBean(OutputDestination.class); + source.send(new GenericMessage("hello".getBytes())); + System.out.println("Result: " + new String(target.receive().getPayload())); +} +---- + +or with legacy annotation-based configuration [source,java] ---- @@ -2601,6 +2291,7 @@ public void sampleTest() { } ---- + 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_. diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java index 000d68e26..600f0d4a9 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java @@ -22,31 +22,41 @@ import org.springframework.cloud.stream.binding.BoundTargetHolder; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.SubscribableChannel; import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; /** * {@link FactoryBean} for creating inputs/outputs destinations to be bound to * function arguments. It is an extension to {@link BindableProxyFactory} which * operates on Bindable interfaces (e.g., Source, Processor, Sink) which internally - * define inputs and output channels. Unlike BindableProxyFactory, this class simply - * operates based on the count of provided inputs and outputs and the names of inputs and outputs - * are based on convention. + * define inputs and output channels. Unlike BindableProxyFactory, this class + * operates based on the count of provided inputs and outputs deriving the binding + * (channel) names based on convention - {@code `_ + + _`} + *
+ * For example, `myFunction_in_0` - is the binding for the first input argument of the + * function with the name `myFunction`. * * @author Oleg Zhurakousky * - * - * * @since 3.0 */ -public class BindableFunctionProxyFactory extends BindableProxyFactory { +class BindableFunctionProxyFactory extends BindableProxyFactory { private final int inputCount; private final int outputCount; - public BindableFunctionProxyFactory(int inputCount, int outputCount) { + private final String functionDefinition; + + private final boolean nameBasedOnFunctionName; + + private boolean multiple; + + BindableFunctionProxyFactory(String functionDefinition, int inputCount, int outputCount, boolean nameBasedOnFunctionName) { super(null); this.inputCount = inputCount; this.outputCount = outputCount; + this.functionDefinition = functionDefinition; + this.nameBasedOnFunctionName = nameBasedOnFunctionName; } @@ -55,25 +65,73 @@ public class BindableFunctionProxyFactory extends BindableProxyFactory { Assert.notEmpty(BindableFunctionProxyFactory.this.bindingTargetFactories, "'bindingTargetFactories' cannot be empty"); + this.multiple = this.inputCount > 1 || this.outputCount > 1; + if (this.inputCount > 0) { - if (this.inputCount == 1) { - this.createInput("input"); + if (multiple || nameBasedOnFunctionName) { + for (int i = 0; i < inputCount; i++) { + this.createInput(this.buildInputNameForIndex(i)); + } } else { - throw new UnsupportedOperationException("Multiple inputs are not currently supported"); + this.createInput("input"); } } if (this.outputCount > 0) { - if (this.outputCount == 1) { - this.createOutput("output"); + if (multiple || nameBasedOnFunctionName) { + for (int i = 0; i < outputCount; i++) { + this.createOutput(this.buildOutputNameForIndex(i)); + } } else { - throw new UnsupportedOperationException("Multiple outputs are not currently supported"); + this.createOutput("output"); } } } + @Override + public Class getObjectType() { + return this.type; + } + + @Override + public boolean isSingleton() { + return true; + } + + protected boolean isNameBasedOnFunctionName() { + return this.nameBasedOnFunctionName; + } + + protected String getFunctionDefinition() { + return this.functionDefinition; + } + + protected String getInputName(int index) { + return CollectionUtils.isEmpty(this.getInputs()) + ? null + : this.getInputs().toArray(new String[0])[index]; + } + + protected String getOutputName(int index) { + return CollectionUtils.isEmpty(this.getOutputs()) + ? null + : this.getOutputs().toArray(new String[0])[index]; + } + + protected boolean isMultiple() { + return this.multiple; + } + + private String buildInputNameForIndex(int index) { + return this.functionDefinition + "_in_" + index; + } + + private String buildOutputNameForIndex(int index) { + return this.functionDefinition + "_out_" + index; + } + private void createInput(String name) { BindableFunctionProxyFactory.this.inputHolders.put(name, new BoundTargetHolder(getBindingTargetFactory(SubscribableChannel.class) @@ -86,15 +144,4 @@ public class BindableFunctionProxyFactory extends BindableProxyFactory { .createOutput(name), true)); } - - @Override - public Class getObjectType() { - return this.type; - } - - @Override - public boolean isSingleton() { - return true; - } - } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index 89a81be11..7f30eab53 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -19,6 +19,8 @@ package org.springframework.cloud.stream.function; import java.lang.reflect.Method; import java.lang.reflect.Type; import java.time.Duration; +import java.util.Iterator; +import java.util.Set; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import java.util.function.Function; @@ -31,6 +33,7 @@ import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.MonoSink; +import reactor.util.function.Tuples; import org.springframework.beans.BeansException; import org.springframework.beans.factory.InitializingBean; @@ -70,6 +73,7 @@ import org.springframework.core.annotation.AnnotationUtils; import org.springframework.core.env.Environment; import org.springframework.core.type.MethodMetadata; import org.springframework.integration.channel.MessageChannelReactiveUtils; +import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlowBuilder; import org.springframework.integration.dsl.IntegrationFlows; @@ -78,11 +82,9 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.SubscribableChannel; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; -import org.springframework.util.MimeTypeUtils; import org.springframework.util.ObjectUtils; import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; @@ -99,83 +101,87 @@ import org.springframework.util.StringUtils; @AutoConfigureBefore(BindingServiceConfiguration.class) public class FunctionConfiguration { - /* - * Creates an effective representation of Bindable interfaces by maintaining the count of inputs and - * outputs based on the provided function, thus preserving the contract and the infrastructure code used - * by current EnableBinding/StreamListener combination. - * It is then used buy `functionInitializer` or 'supplierInitializer` where functions are actually bound to channels. - * - * Also, see the BindableFunctionProxyFactory - */ @Bean - public InitializingBean functionBindingHolder(Environment environment, FunctionCatalog functionCatalog, + public InitializingBean functionBindingRegistrar(Environment environment, FunctionCatalog functionCatalog, StreamFunctionProperties streamFunctionProperties, BinderTypeRegistry binderTypeRegistry) { - return new FunctionBindingHolder(binderTypeRegistry, functionCatalog, streamFunctionProperties); + return new FunctionBindingRegistrar(binderTypeRegistry, functionCatalog, streamFunctionProperties); } @Bean public InitializingBean functionInitializer(FunctionCatalog functionCatalog, FunctionInspector functionInspector, - StreamFunctionProperties functionProperties, @Nullable BindableProxyFactory[] bpfs, BindingServiceProperties serviceProperties, - ConfigurableApplicationContext applicationContext, FunctionBindingHolder bindingHolder) { + StreamFunctionProperties functionProperties, @Nullable BindableProxyFactory[] bindableProxyFactories, + BindingServiceProperties serviceProperties, ConfigurableApplicationContext applicationContext, FunctionBindingRegistrar bindingHolder) { - if (bpfs == null || bpfs.length > 1) { - return null; // basically we're not dealing with multiple EnableBinding which is how multiple BindableProxyFactory are created - } - BindableProxyFactory bindableProxyFactory = bpfs[0]; - - return bindingHolder.getInputCount() > 0 // basically not a Supplier - || !ObjectUtils.isEmpty(applicationContext.getBeanNamesForAnnotation(EnableBinding.class)) // implies existing binding to which we are going to 'compose to' - ? new FunctionChannelBindingInitializer(functionCatalog, functionInspector, functionProperties, bindableProxyFactory, serviceProperties) - : null; - } - - @Bean - public IntegrationFlow supplierInitializer(FunctionCatalog functionCatalog, FunctionInspector functionInspector, - StreamFunctionProperties functionProperties, GenericApplicationContext context, FunctionBindingHolder bindingHolder) { - if (bindingHolder.getInputCount() > 0) { + if (bindableProxyFactories == null) { return null; } - FunctionInvocationWrapper functionWrapper = functionCatalog.lookup(functionProperties.getDefinition()); - IntegrationFlow integrationFlow = null; - if (ObjectUtils.isEmpty(context.getBeanNamesForAnnotation(EnableBinding.class)) && functionWrapper != null && functionWrapper.isSupplier()) { - AtomicReference> triggerRef = new AtomicReference<>(); - Publisher beginPublishingTrigger = Mono.create(emmiter -> { - triggerRef.set(emmiter); - }); - context.addApplicationListener(event -> { - if (event instanceof BindingCreatedEvent) { - if (triggerRef.get() != null) { - triggerRef.get().success(); - } - } - }); + return new FunctionChannelBindingInitializer(functionCatalog, functionInspector, functionProperties, bindableProxyFactories, serviceProperties); + } - RootBeanDefinition bd = (RootBeanDefinition) context.getBeanDefinition(functionProperties.getParsedDefinition()[0]); - Method factoryMethod = bd.getResolvedFactoryMethod(); - if (factoryMethod == null) { - Object source = bd.getSource(); - if (source instanceof MethodMetadata) { - Class factory = ClassUtils.resolveClassName(((MethodMetadata) source).getDeclaringClassName(), null); - Class[] params = FunctionContextUtils.getParamTypesFromBeanDefinitionFactory(factory, bd); - factoryMethod = ReflectionUtils.findMethod(factory, ((MethodMetadata) source).getMethodName(), params); - } - } - Assert.notNull(factoryMethod, "Failed to introspect factory method since it was not discovered for function '" - + functionProperties.getDefinition() + "'"); - PollableSupplier pollable = factoryMethod.getReturnType().isAssignableFrom(Supplier.class) - ? AnnotationUtils.findAnnotation(factoryMethod, PollableSupplier.class) - : null; - - if (!functionProperties.isComposeFrom() && !functionProperties.isComposeTo()) { - integrationFlow = this.integrationFlowFromProvidedSupplier(functionWrapper, functionInspector, beginPublishingTrigger, pollable) - .channel("output").get(); - } + /* + * Binding initializer responsible only for Suppliers only + */ + @Bean + IntegrationFlow supplierInitializer(FunctionCatalog functionCatalog, FunctionInspector functionInspector, + StreamFunctionProperties functionProperties, GenericApplicationContext context) { + if (!ObjectUtils.isEmpty(context.getBeanNamesForAnnotation(EnableBinding.class))) { + return null; } + IntegrationFlow integrationFlow = null; + String[] functionDefinitions = StringUtils.hasText(functionProperties.getDefinition()) + ? functionProperties.getDefinition().split(";") + : new String[] {}; + + for (String functionDefinition : functionDefinitions) { + FunctionInvocationWrapper functionWrapper = functionCatalog.lookup(functionDefinition); + if (functionWrapper != null && functionWrapper.isSupplier()) { + Publisher beginPublishingTrigger = this.setupBindingTrigger(context); + + RootBeanDefinition bd = (RootBeanDefinition) context.getBeanDefinition(functionProperties.getParsedDefinition()[0]); + Method factoryMethod = bd.getResolvedFactoryMethod(); + if (factoryMethod == null) { + Object source = bd.getSource(); + if (source instanceof MethodMetadata) { + Class factory = ClassUtils.resolveClassName(((MethodMetadata) source).getDeclaringClassName(), null); + Class[] params = FunctionContextUtils.getParamTypesFromBeanDefinitionFactory(factory, bd); + factoryMethod = ReflectionUtils.findMethod(factory, ((MethodMetadata) source).getMethodName(), params); + } + } + Assert.notNull(factoryMethod, "Failed to introspect factory method since it was not discovered for function '" + + functionProperties.getDefinition() + "'"); + PollableSupplier pollable = factoryMethod.getReturnType().isAssignableFrom(Supplier.class) + ? AnnotationUtils.findAnnotation(factoryMethod, PollableSupplier.class) + : null; + + if (!functionProperties.isComposeFrom() && !functionProperties.isComposeTo()) { + integrationFlow = this.integrationFlowFromProvidedSupplier(functionWrapper, functionInspector, beginPublishingTrigger, pollable) + .channel("output").get(); + } + } + } return integrationFlow; } + /* + * Creates a publishing trigger to ensure Supplier does not begin publishing until binding is created + */ + private Publisher setupBindingTrigger(GenericApplicationContext context) { + AtomicReference> triggerRef = new AtomicReference<>(); + Publisher beginPublishingTrigger = Mono.create(emmiter -> { + triggerRef.set(emmiter); + }); + context.addApplicationListener(event -> { + if (event instanceof BindingCreatedEvent) { + if (triggerRef.get() != null) { + triggerRef.get().success(); + } + } + }); + return beginPublishingTrigger; + } + @SuppressWarnings({ "rawtypes", "unchecked" }) private IntegrationFlowBuilder integrationFlowFromProvidedSupplier(Supplier supplier, FunctionInspector inspector, Publisher beginPublishingTrigger, PollableSupplier pollable) { @@ -206,13 +212,13 @@ public class FunctionConfiguration { @SuppressWarnings("unchecked") private Message wrapToMessageIfNecessary(T value) { - return value instanceof Message ? (Message) value : MessageBuilder.withPayload(value).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON).build(); + return value instanceof Message + ? (Message) value + : MessageBuilder.withPayload(value).build(); } - /** - * - * @author Oleg Zhurakousky - * @since 3.0 + /* + * Binding initializer responsible only for Functions and Consumers. */ private static class FunctionChannelBindingInitializer implements InitializingBean, ApplicationContextAware { @@ -224,7 +230,7 @@ public class FunctionConfiguration { private final StreamFunctionProperties functionProperties; - private final BindableProxyFactory bindableProxyFactory; + private final BindableProxyFactory[] bindableProxyFactories; private final BindingServiceProperties serviceProperties; @@ -232,35 +238,38 @@ public class FunctionConfiguration { FunctionChannelBindingInitializer(FunctionCatalog functionCatalog, FunctionInspector functionInspector, - StreamFunctionProperties functionProperties, BindableProxyFactory bindableProxyFactory, BindingServiceProperties serviceProperties) { + StreamFunctionProperties functionProperties, BindableProxyFactory[] bindableProxyFactories, BindingServiceProperties serviceProperties) { this.functionCatalog = functionCatalog; this.functionInspector = functionInspector; this.functionProperties = functionProperties; - this.bindableProxyFactory = bindableProxyFactory; + this.bindableProxyFactories = bindableProxyFactories; this.serviceProperties = serviceProperties; } @Override public void afterPropertiesSet() throws Exception { - MessageChannel messageChannel = null; - String channelName = Sink.INPUT; - if (context.containsBean(channelName)) { - Object bean = context.getBean(channelName); - if (bean instanceof MessageChannel) { - messageChannel = context.getBean(channelName, MessageChannel.class); + Stream.of(this.bindableProxyFactories).forEach(bindableProxyFactory -> { + String functionDefinition = getFunctionDefinition(bindableProxyFactory); + FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition); + if (function != null && !function.isSupplier()) { + if (isMultipleInputOutput(bindableProxyFactory)) { + this.bindMultipleArgumentsFunction(bindableProxyFactory, functionDefinition); + } + else { + SubscribableChannel messageChannel = this.determineChannelToSubscribeTo(bindableProxyFactory); + if (messageChannel != null && function != null) { + this.bindOrComposeSimpleFunctions(((IntegrationObjectSupport) messageChannel).getComponentName(), + (SubscribableChannel) messageChannel, bindableProxyFactory, functionDefinition); + } + } } - } - if (messageChannel == null && context.containsBean(Source.OUTPUT)) { - channelName = "output"; - Object bean = context.getBean(channelName); - if (bean instanceof MessageChannel) { - messageChannel = context.getBean(channelName, SubscribableChannel.class); - } - } + }); + } - if (messageChannel != null && functionCatalog.lookup(functionProperties.getDefinition()) != null) { - this.doPostProcess(channelName, (SubscribableChannel) messageChannel); - } + private String getFunctionDefinition(BindableProxyFactory bindableProxyFactory) { + return bindableProxyFactory instanceof BindableFunctionProxyFactory + ? ((BindableFunctionProxyFactory) bindableProxyFactory).getFunctionDefinition() + : functionProperties.getDefinition(); } @Override @@ -268,67 +277,134 @@ public class FunctionConfiguration { this.context = (GenericApplicationContext) applicationContext; } - private void doPostProcess(String channelName, SubscribableChannel messageChannel) { - //TODO there is something about moving channel interceptors in AMCB (not sure if it is still required) - if (functionProperties.isComposeTo() && messageChannel instanceof SubscribableChannel && Sink.INPUT.equals(channelName)) { - throw new UnsupportedOperationException("Composing at tail is not currently supported"); - } - else if (functionProperties.isComposeFrom() && Source.OUTPUT.equals(channelName)) { - Assert.notNull(this.bindableProxyFactory, "Can not compose function into the existing app since `bindableProxyFactory` is null."); - logger.info("Composing at the head of 'output' channel"); - BindingProperties properties = this.serviceProperties.getBindings().get(Source.OUTPUT); - FunctionInvocationWrapper function = functionCatalog.lookup(functionProperties.getDefinition(), properties.getContentType()); - ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function)); - handler.setBeanFactory(context); - handler.afterPropertiesSet(); - - DirectWithAttributesChannel newOutputChannel = new DirectWithAttributesChannel(); - newOutputChannel.setAttribute("type", "output"); - newOutputChannel.setComponentName("output.extended"); - this.context.registerBean("output.extended", MessageChannel.class, () -> newOutputChannel); - this.bindableProxyFactory.replaceOutputChannel(channelName, "output.extended", newOutputChannel); - - handler.setOutputChannelName("output.extended"); - SubscribableChannel subscribeChannel = (SubscribableChannel) messageChannel; - subscribeChannel.subscribe(handler); + private SubscribableChannel determineChannelToSubscribeTo(BindableProxyFactory bindableProxyFactory) { + SubscribableChannel messageChannel = null; + if (bindableProxyFactory instanceof BindableFunctionProxyFactory) { + String channelName = ((BindableFunctionProxyFactory) bindableProxyFactory).getInputName(0); + messageChannel = context.getBean(channelName, SubscribableChannel.class); } else { - if (Sink.INPUT.equals(channelName)) { - BindingProperties properties = this.serviceProperties.getBindings().get(Sink.INPUT); - FunctionInvocationWrapper function = functionCatalog.lookup(functionProperties.getDefinition(), properties.getContentType()); - this.postProcessForStandAloneFunction(function, messageChannel); + // could be "input" or "output" if subscribing to existing Source + if (context.containsBean(Sink.INPUT)) { + messageChannel = context.getBean(Sink.INPUT, SubscribableChannel.class); } + else if (context.containsBean(Source.OUTPUT)) { + messageChannel = context.getBean(Source.OUTPUT, SubscribableChannel.class); + } + } + return messageChannel; + } + + @SuppressWarnings({ "unchecked", "rawtypes" }) + private void bindMultipleArgumentsFunction(BindableProxyFactory bindableProxyFactory, String functionDefinition) { + Assert.isTrue(!functionProperties.isComposeTo() && !functionProperties.isComposeFrom(), + "Composing to/from existing Sinks and Sources are not supported for functions with multiple arguments."); + + BindableFunctionProxyFactory functionProxyFactory = (BindableFunctionProxyFactory) bindableProxyFactory; + Set inputBindingNames = functionProxyFactory.getInputs(); + Set outputBindingNames = functionProxyFactory.getOutputs(); + + String[] outputContentTypes = outputBindingNames.stream() + .map(bindingName -> this.serviceProperties.getBindings().get(bindingName).getContentType()) + .toArray(String[]::new); + + FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition, outputContentTypes); + + if (isMultipleInputOutput(bindableProxyFactory)) { + this.assertSupportedSignatures(function.getFunctionType()); + } + + Publisher[] inputPublishers = inputBindingNames.stream().map(inputBindingName -> { + SubscribableChannel inputChannel = context.getBean(inputBindingName, SubscribableChannel.class); + return this.enhancePublisher(MessageChannelReactiveUtils.toPublisher(inputChannel), inputBindingName); + }).toArray(Publisher[]::new); + + + Object resultPublishers = function.apply(inputPublishers.length == 1 ? inputPublishers[0] : Tuples.fromArray(inputPublishers)); + if (resultPublishers instanceof Iterable) { + Iterator outputBindingIter = outputBindingNames.iterator(); + ((Iterable) resultPublishers).forEach(publisher -> { + MessageChannel outputChannel = context.getBean(outputBindingIter.next(), MessageChannel.class); + Flux.from((Publisher) publisher).doOnNext(message -> outputChannel.send((Message) message)).subscribe(); + }); + } + else { + outputBindingNames.stream().forEach(outputBindingName -> { + MessageChannel outputChannel = context.getBean(outputBindingName, MessageChannel.class); + Flux.from((Publisher) resultPublishers).doOnNext(message -> outputChannel.send((Message) message)).subscribe(); + }); } } - private void postProcessForStandAloneFunction(FunctionInvocationWrapper function, MessageChannel inputChannel) { - Type functionType = FunctionTypeUtils.getFunctionType(function, this.functionInspector); - if (FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, 0))) { - MessageChannel outputChannel = context.getBean(Source.OUTPUT, MessageChannel.class); - SubscribableChannel subscribeChannel = (SubscribableChannel) inputChannel; - Publisher publisher = this.enhancePublisher(MessageChannelReactiveUtils.toPublisher(subscribeChannel)); - this.subscribeToInput(function, publisher, outputChannel::send); + // + private void bindOrComposeSimpleFunctions(String channelName, SubscribableChannel messageChannel, + BindableProxyFactory bindableProxyFactory, String functionDefinition) { + //TODO there is something about moving channel interceptors in AMCB (not sure if it is still required) + String channelType = (String) ((DirectWithAttributesChannel) messageChannel).getAttribute("type"); + if (Source.OUTPUT.equals(channelType) && functionProperties.isComposeFrom()) { + logger.info("Composing at the head of 'output' channel"); + BindingProperties properties = this.serviceProperties.getBindings().get(channelName); + FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition, properties.getContentType()); + this.composeSimpleFunctionToExistingFlow(function, messageChannel, channelName, bindableProxyFactory); } else { - ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function)); - handler.setBeanFactory(context); - handler.afterPropertiesSet(); - if (!FunctionTypeUtils.isConsumer(functionType)) { - handler.setOutputChannelName(Source.OUTPUT); - } - SubscribableChannel subscribeChannel = (SubscribableChannel) inputChannel; - subscribeChannel.subscribe(handler); + BindingProperties properties = this.serviceProperties.getBindings().get(channelName); + FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition, properties.getContentType()); + this.bindSimpleFunctions(function, messageChannel, bindableProxyFactory); } } + private void composeSimpleFunctionToExistingFlow(FunctionInvocationWrapper function, SubscribableChannel messageChannel, + String channelName, BindableProxyFactory bindableProxyFactory) { + ServiceActivatingHandler handler = createFunctionHandler(function); + + DirectWithAttributesChannel newOutputChannel = new DirectWithAttributesChannel(); + newOutputChannel.setAttribute("type", "output"); + newOutputChannel.setComponentName("output.extended"); + this.context.registerBean("output.extended", MessageChannel.class, () -> newOutputChannel); + bindableProxyFactory.replaceOutputChannel(channelName, "output.extended", newOutputChannel); + + handler.setOutputChannelName("output.extended"); + messageChannel.subscribe(handler); + } + + private void bindSimpleFunctions(FunctionInvocationWrapper function, SubscribableChannel inputChannel, BindableProxyFactory bindableProxyFactory) { + Type functionType = FunctionTypeUtils.getFunctionType(function, this.functionInspector); + String outputChannelName = bindableProxyFactory instanceof BindableFunctionProxyFactory + ? ((BindableFunctionProxyFactory) bindableProxyFactory).getOutputName(0) + : Source.OUTPUT; + + if (FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, 0))) { + MessageChannel outputChannel = context.getBean(outputChannelName, MessageChannel.class); + SubscribableChannel subscribeChannel = (SubscribableChannel) inputChannel; + Publisher publisher = this.enhancePublisher(MessageChannelReactiveUtils.toPublisher(subscribeChannel), + ((DirectWithAttributesChannel) inputChannel).getBeanName()); + this.subscribeToInput(function, publisher, message -> outputChannel.send((Message) message)); + } + else { + ServiceActivatingHandler handler = createFunctionHandler(function); + if (!FunctionTypeUtils.isConsumer(functionType)) { + handler.setOutputChannelName(outputChannelName); + } + inputChannel.subscribe(handler); + } + } + + private ServiceActivatingHandler createFunctionHandler(FunctionInvocationWrapper function) { + ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function)); + handler.setBeanFactory(context); + handler.afterPropertiesSet(); + return handler; + } + /* * Enhance publisher to add error handling, retries etc. */ @SuppressWarnings({ "unchecked", "rawtypes" }) - private Publisher enhancePublisher(Publisher publisher) { + private Publisher enhancePublisher(Publisher publisher, String bindingName) { Flux flux = Flux.from(publisher) .concatMap(message -> { - ConsumerProperties consumerProperties = this.serviceProperties.getBindings().get(Sink.INPUT).getConsumer(); + ConsumerProperties consumerProperties = this.serviceProperties.getBindings().get(bindingName).getConsumer(); return Flux.just(message) .doOnError(e -> { e.printStackTrace(); @@ -349,27 +425,60 @@ public class FunctionConfiguration { @SuppressWarnings({ "unchecked", "rawtypes" }) - private void subscribeToInput(Function function, - Publisher publisher, Consumer> outputProcessor) { - - Function>, Flux>> functionInvoker = function; + private void subscribeToInput(Function function, Publisher publisher, Consumer outputProcessor) { + Function, Flux> functionInvoker = function; Flux inputPublisher = Flux.from(publisher); - subscribeToOutput(outputProcessor, - functionInvoker.apply((Flux>) inputPublisher)).subscribe(); + subscribeToOutput(outputProcessor, functionInvoker.apply((Flux) inputPublisher)).subscribe(); } - private Mono subscribeToOutput(Consumer> outputProcessor, - Publisher> outputPublisher) { - - Flux> output = outputProcessor == null ? Flux.from(outputPublisher) - : Flux.from(outputPublisher).doOnNext(outputProcessor); + @SuppressWarnings("rawtypes") + private Mono subscribeToOutput(Consumer outputProcessor, Flux resultPublisher) { + Flux output = outputProcessor == null + ? resultPublisher + : resultPublisher.doOnNext(outputProcessor); return output.then(); } + + private void assertSupportedSignatures(Type functionType) { + Assert.isTrue(!FunctionTypeUtils.isConsumer(functionType), + "Function '" + functionProperties.getDefinition() + "' is a Consumer which is not supported " + + "for multi-in/out reactive streams. Only Functions are supported"); + Assert.isTrue(!FunctionTypeUtils.isSupplier(functionType), + "Function '" + functionProperties.getDefinition() + "' is a Supplier which is not supported " + + "for multi-in/out reactive streams. Only Functions are supported"); + Assert.isTrue(!FunctionTypeUtils.isInputArray(functionType) && !FunctionTypeUtils.isOutputArray(functionType), + "Function '" + functionProperties.getDefinition() + "' has the following signature: [" + + functionType + "]. Your input and/or outout lacks arity and therefore we " + + "can not determine how many input/output destinations are required in the context of " + + "function input/output binding."); + + int inputCount = FunctionTypeUtils.getInputCount(functionType); + for (int i = 0; i < inputCount; i++) { + Assert.isTrue(FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, i)), + "Function '" + functionProperties.getDefinition() + "' has the following signature: [" + + functionType + "]. Non-reactive functions with multiple " + + "inputs/outputs are not supported in the context of Spring Cloud Stream."); + } + int outputCount = FunctionTypeUtils.getOutputCount(functionType); + for (int i = 0; i < outputCount; i++) { + Assert.isTrue(FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, i)), + "Function '" + functionProperties.getDefinition() + "' has the following signature: [" + + functionType + "]. Non-reactive functions with multiple " + + "inputs/outputs are not supported in the context of Spring Cloud Stream."); + } + + } + + private boolean isMultipleInputOutput(BindableProxyFactory bindableProxyFactory) { + return bindableProxyFactory instanceof BindableFunctionProxyFactory + && ((BindableFunctionProxyFactory) bindableProxyFactory).isMultiple(); + } } + /** * - * Ensure that SI does not attempt any conversion and sends a raw Message. - * + * It's signatures ensures that within the context of s-c-stream Spring Integration does + * not attempt any conversion and sends a raw Message. */ @SuppressWarnings("rawtypes") private static class FunctionWrapper implements Function, Object> { @@ -389,11 +498,11 @@ public class FunctionConfiguration { } } - /* - * This class will effectively create a different representation of Bindable interfaces (e.g., Source, Processor...). - * It's main goal is to determine the count of inputs and outputs based on the provided function. + /** + * Creates and registers instances of BindableFunctionProxyFactory for each user defined function + * thus triggering destination bindings between function arguments and destinations. */ - private static class FunctionBindingHolder implements InitializingBean, ApplicationContextAware, EnvironmentAware { + private static class FunctionBindingRegistrar implements InitializingBean, ApplicationContextAware, EnvironmentAware { private final BinderTypeRegistry binderTypeRegistry; @@ -409,7 +518,7 @@ public class FunctionConfiguration { private int outputCount; - FunctionBindingHolder(BinderTypeRegistry binderTypeRegistry, FunctionCatalog functionCatalog, StreamFunctionProperties streamFunctionProperties) { + FunctionBindingRegistrar(BinderTypeRegistry binderTypeRegistry, FunctionCatalog functionCatalog, StreamFunctionProperties streamFunctionProperties) { this.binderTypeRegistry = binderTypeRegistry; this.functionCatalog = functionCatalog; this.streamFunctionProperties = streamFunctionProperties; @@ -426,37 +535,35 @@ public class FunctionConfiguration { && ObjectUtils.isEmpty(applicationContext.getBeanNamesForAnnotation(EnableBinding.class)) && this.determineFunctionName(functionCatalog, environment)) { BeanDefinitionRegistry registry = (BeanDefinitionRegistry) applicationContext.getBeanFactory(); - RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(BindableFunctionProxyFactory.class); - FunctionInvocationWrapper function = functionCatalog.lookup(streamFunctionProperties.getDefinition()); - if (function != null) { - if (function.isSupplier()) { - this.inputCount = 0; - this.outputCount = 1; + String[] functionDefinitions = streamFunctionProperties.getDefinition().split(";"); + boolean nameBasedOnFunctionName = functionDefinitions.length > 1; + for (String functionDefinition : functionDefinitions) { + RootBeanDefinition functionBindableProxyDefinition = new RootBeanDefinition(BindableFunctionProxyFactory.class); + FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition); + if (function != null) { + Type functionType = function.getFunctionType(); + if (function.isSupplier()) { + this.inputCount = 0; + this.outputCount = FunctionTypeUtils.getOutputCount(functionType); + } + else if (function.isConsumer()) { + this.inputCount = FunctionTypeUtils.getInputCount(functionType); + this.outputCount = 0; + } + else { + this.inputCount = FunctionTypeUtils.getInputCount(functionType); + this.outputCount = FunctionTypeUtils.getOutputCount(functionType); + } + functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(functionDefinition); + functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.inputCount); + functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.outputCount); + functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(nameBasedOnFunctionName); + registry.registerBeanDefinition(functionDefinition + "_binding", functionBindableProxyDefinition); } - else if (function.isConsumer()) { - this.inputCount = 1; - this.outputCount = 0; - } - else { - this.inputCount = 1; - this.outputCount = 1; - } - rootBeanDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.inputCount); - rootBeanDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.outputCount); - registry.registerBeanDefinition(streamFunctionProperties.getDefinition() + "_binding", - rootBeanDefinition); } } } - int getInputCount() { - return this.inputCount; - } - - int getOutputCount() { - return this.outputCount; - } - private boolean determineFunctionName(FunctionCatalog catalog, Environment environment) { String definition = streamFunctionProperties.getDefinition(); if (!StringUtils.hasText(definition)) { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/AbstractDestination.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/AbstractDestination.java index d9688d0c9..0ba9f0c60 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/AbstractDestination.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/AbstractDestination.java @@ -16,6 +16,9 @@ package org.springframework.cloud.stream.binder.test; +import java.util.ArrayList; +import java.util.List; + import org.springframework.messaging.SubscribableChannel; /** @@ -24,18 +27,18 @@ import org.springframework.messaging.SubscribableChannel; */ abstract class AbstractDestination { - private SubscribableChannel channel; + private final List channels = new ArrayList<>(); - SubscribableChannel getChannel() { - return this.channel; + SubscribableChannel getChannel(int index) { + return this.channels.get(index); } void setChannel(SubscribableChannel channel) { - this.channel = channel; - this.afterChannelIsSet(); + this.channels.add(channel); + this.afterChannelIsSet(this.channels.size() - 1); } - void afterChannelIsSet() { + void afterChannelIsSet(int channelIndex) { // noop } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/InputDestination.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/InputDestination.java index 53086951e..f8ecf1fd6 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/InputDestination.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/InputDestination.java @@ -34,7 +34,11 @@ public class InputDestination extends AbstractDestination { * @param message message to send */ public void send(Message message) { - this.getChannel().send(message); + this.getChannel(0).send(message); + } + + public void send(Message message, int inputIndex) { + this.getChannel(inputIndex).send(message); } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/OutputDestination.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/OutputDestination.java index a51bbfa3e..753982858 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/OutputDestination.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/OutputDestination.java @@ -16,6 +16,8 @@ package org.springframework.cloud.stream.binder.test; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedTransferQueue; import java.util.concurrent.TimeUnit; @@ -32,7 +34,7 @@ import org.springframework.messaging.Message; */ public class OutputDestination extends AbstractDestination { - private BlockingQueue> messages; + private final List>> messageQueues = new ArrayList<>(); /** * Allows to access {@link Message}s received by this {@link OutputDestination}. @@ -40,9 +42,9 @@ public class OutputDestination extends AbstractDestination { * @return received message */ @SuppressWarnings("unchecked") - public Message receive(long timeout) { + public Message receive(long timeout, int channelIndex) { try { - return (Message) this.messages.poll(timeout, TimeUnit.MILLISECONDS); + return (Message) this.messageQueues.get(channelIndex).poll(timeout, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -55,13 +57,18 @@ public class OutputDestination extends AbstractDestination { * @return received message */ public Message receive() { - return this.receive(0); + return this.receive(0, 0); + } + + public Message receive(long timeout) { + return this.receive(timeout, 0); } @Override - void afterChannelIsSet() { - this.messages = new LinkedTransferQueue<>(); - this.getChannel().subscribe(message -> this.messages.offer(message)); + void afterChannelIsSet(int channelIndex) { + BlockingQueue> messageQueue = new LinkedTransferQueue<>(); + this.messageQueues.add(messageQueue); + this.getChannel(channelIndex).subscribe(message -> this.messageQueues.get(channelIndex).offer(message)); } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/MultipleInputOutputFunctionTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/MultipleInputOutputFunctionTests.java new file mode 100644 index 000000000..0481c2b7d --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/MultipleInputOutputFunctionTests.java @@ -0,0 +1,292 @@ +/* + * Copyright 2019-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.function; + +import java.util.function.Consumer; +import java.util.function.Function; + +import org.junit.Test; +import reactor.core.publisher.Flux; +import reactor.core.publisher.UnicastProcessor; +import reactor.util.function.Tuple2; +import reactor.util.function.Tuples; + +import org.springframework.beans.factory.BeanCreationException; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +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.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; + +import static org.assertj.core.api.Assertions.assertThat; + + +/** + * + * @author Oleg Zhurakousky + * + */ +public class MultipleInputOutputFunctionTests { + + @Test(expected = BeanCreationException.class) + public void testFailureWithNonReactiveFunction() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + ReactiveFunctionConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.function.definition=multipleInputNonReactive")) { + context.getBean(InputDestination.class); + } + } + + @Test(expected = BeanCreationException.class) + public void testFailureWithReactiveArrayOutput() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + ReactiveFunctionConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.function.definition=multiReactiveInputReactiveArrayOutput")) { + context.getBean(InputDestination.class); + } + } + + @Test(expected = BeanCreationException.class) + public void testFailureWithReactiveArrayOutputNonGeneric() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + ReactiveFunctionConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.function.definition=multiReactiveInputReactiveArrayOutputNoGeneric")) { + context.getBean(InputDestination.class); + } + } + + @Test(expected = BeanCreationException.class) + public void testFailureWithReactiveArrayInput() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + ReactiveFunctionConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.function.definition=genericReactiveArrayInput")) { + context.getBean(InputDestination.class); + } + } + + @Test(expected = BeanCreationException.class) + public void testFailureWithReactiveArrayInputNonGeneric() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + ReactiveFunctionConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.function.definition=nonGenericReactiveArrayInput")) { + context.getBean(InputDestination.class); + } + } + + @Test(expected = BeanCreationException.class) + public void testFailureWithConsumer() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + ReactiveFunctionConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.function.definition=multiInputConsumer")) { + context.getBean(InputDestination.class); + } + } + + @Test + public void testMultiInputSingleOutput() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + ReactiveFunctionConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.function.definition=multiInputSingleOutput")) { + context.getBean(InputDestination.class); + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + Message stringInputMessage = MessageBuilder.withPayload("one".getBytes()).build(); + Message integerInputMessage = MessageBuilder.withPayload("1".getBytes()).build(); + inputDestination.send(stringInputMessage, 0); + inputDestination.send(integerInputMessage, 1); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage.getPayload()).isEqualTo("one".getBytes()); + outputMessage = outputDestination.receive(); + assertThat(outputMessage.getPayload()).isEqualTo("1".getBytes()); + } + } + + @Test + public void testSingleInputMultiOutput() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + ReactiveFunctionConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.function.definition=singleInputMultipleOutputs")) { + context.getBean(InputDestination.class); + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + for (int i = 0; i < 10; i++) { + inputDestination.send(MessageBuilder.withPayload(String.valueOf(i).getBytes()).build()); + } + + int counter = 0; + for (int i = 0; i < 5; i++) { + Message even = outputDestination.receive(0, 0); + assertThat(even.getPayload()).isEqualTo(("EVEN: " + String.valueOf(counter++)).getBytes()); + Message odd = outputDestination.receive(0, 1); + assertThat(odd.getPayload()).isEqualTo(("ODD: " + String.valueOf(counter++)).getBytes()); + } + } + } + + @Test + public void testMultipleFunctions() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + ReactiveFunctionConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.function.definition=uppercase;reverse")) { + context.getBean(InputDestination.class); + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + Message inputMessage = MessageBuilder.withPayload("Hello".getBytes()).build(); + inputDestination.send(inputMessage, 0); + inputDestination.send(inputMessage, 1); + + Message outputMessage = outputDestination.receive(0, 0); + assertThat(outputMessage.getPayload()).isEqualTo("HELLO".getBytes()); + + outputMessage = outputDestination.receive(0, 1); + assertThat(outputMessage.getPayload()).isEqualTo("olleH".getBytes()); + } + } + + @Test + public void testMultipleFunctionsWithComposition() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + ReactiveFunctionConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.function.definition=uppercase|reverse;reverse|uppercase")) { + context.getBean(InputDestination.class); + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + Message inputMessage = MessageBuilder.withPayload("Hello".getBytes()).build(); + inputDestination.send(inputMessage, 0); + inputDestination.send(inputMessage, 1); + + Message outputMessage = outputDestination.receive(0, 0); + assertThat(outputMessage.getPayload()).isEqualTo("OLLEH".getBytes()); + + outputMessage = outputDestination.receive(0, 1); + assertThat(outputMessage.getPayload()).isEqualTo("OLLEH".getBytes()); + } + } + + @EnableAutoConfiguration + public static class ReactiveFunctionConfiguration { + + @Bean + public Function uppercase() { + return value -> value.toUpperCase(); + } + + @Bean + public Function reverse() { + return value -> new StringBuilder(value).reverse().toString(); + } + + @Bean + public Function, String> multipleInputNonReactive() { // not supported + return tuple -> null; + } + + @Bean + public Function, Flux>, Flux[]> multiReactiveInputReactiveArrayOutput() { // not supported + return tuple -> null; + } + + @Bean + public Consumer, Flux>> multiInputConsumer() { // not supported + return tuple -> System.out.println(); + } + + @SuppressWarnings("rawtypes") + @Bean + public Function, Flux>, Flux[]> multiReactiveInputReactiveArrayOutputNoGeneric() { // not supported + return tuple -> null; + } + + @Bean + public Function[], Tuple2, Flux>> genericReactiveArrayInput() { // not supported + return tuple -> null; + } + + @SuppressWarnings("rawtypes") + @Bean + public Function, Flux>> nonGenericReactiveArrayInput() { // not supported + return tuple -> null; + } + + @Bean + public Function, Flux>, Flux> multiInputSingleOutput() { + return tuple -> { + Flux stringStream = tuple.getT1(); + Flux intStream = tuple.getT2().map(i -> String.valueOf(i)); + return Flux.merge(stringStream, intStream); + }; + } + + @Bean + @SuppressWarnings({ "unchecked", "rawtypes" }) + public static Function, Tuple2, Flux>> singleInputMultipleOutputs() { + return flux -> { + Flux connectedFlux = flux.publish().autoConnect(2); + UnicastProcessor even = UnicastProcessor.create(); + UnicastProcessor odd = UnicastProcessor.create(); + Flux evenFlux = connectedFlux.filter(number -> number % 2 == 0).doOnNext(number -> even.onNext("EVEN: " + number)); + Flux oddFlux = connectedFlux.filter(number -> number % 2 != 0).doOnNext(number -> odd.onNext("ODD: " + number)); + + return Tuples.of(Flux.from(even).doOnSubscribe(x -> evenFlux.subscribe()), Flux.from(odd).doOnSubscribe(x -> oddFlux.subscribe())); + }; + } + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java index 811069243..dd1276647 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java @@ -234,7 +234,6 @@ public class SourceToFunctionsSupportTests { assertThat(new String(target.receive(2000).getPayload())).isEqualTo("6"); assertThat(context.getBean("supplierInitializer")).isNotEqualTo(null); - assertThat(context.getBean("functionInitializer")).isEqualTo(null); } }