diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index db03cc014..f8e9a7594 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -317,18 +317,18 @@ public class SampleApplication { We certainly can't use `input` and `output` as names given that we actually have multiple inputs and outputs. For those cases the following naming convention applies: -* input - ` + .in. + ` -* output - ` + .out. + ` +* input - ` + -in- + ` +* output - ` + -out- + ` So if for example you would want to map the input of 'uppercase()' function to a remote destination (e.g., topic, queue etc) called "my-topic" you would do so with the following property: ---- -spring.cloud.stream.bindings.uppercase.in.0.destination=my-topic +spring.cloud.stream.bindings.uppercase-in-0.destination=my-topic ---- And if you want to change the content-type of the output of the 'lowercase()' function you would do so with the following property. ---- -spring.cloud.stream.bindings.lowercase.out.0.content-type=text/plain +spring.cloud.stream.bindings.lowercase-out-0.content-type=text/plain ---- For more on properties and other configuration options please see <> section. @@ -635,15 +635,15 @@ public class SampleApplication { 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`. +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: +For example, the following will override content-type for `gather-in-0` binding: ---- ---spring.cloud.stream.bindings.gather.in.0.content-type=text/plain +--spring.cloud.stream.bindings.gather-in-0.content-type=text/plain ---- @@ -670,8 +670,8 @@ public class SampleApplication { 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`. +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] 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 dbc8e6ce1..967a10795 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 @@ -41,7 +41,7 @@ import org.springframework.util.CollectionUtils; */ class BindableFunctionProxyFactory extends BindableProxyFactory { - static final String delimiter = "."; + static final String delimiter = "-"; private final int inputCount; 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 index 0481c2b7d..5b1e782aa 100644 --- 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 @@ -16,6 +16,8 @@ package org.springframework.cloud.stream.function; +import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.util.function.Consumer; import java.util.function.Function; @@ -35,7 +37,12 @@ import org.springframework.cloud.stream.binder.test.TestChannelBinderConfigurati import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.integration.support.MessageBuilder; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.converter.AbstractMessageConverter; +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.util.MimeType; import static org.assertj.core.api.Assertions.assertThat; @@ -221,6 +228,61 @@ public class MultipleInputOutputFunctionTests { } } + @Test + public void testMultiInputSingleOutputWithCustomContentType() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + ContentTypeConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.function.definition=multiInputSingleOutput", + "--spring.cloud.stream.bindings.multiInputSingleOutput-in-0.content-type=string/person", + "--spring.cloud.stream.bindings.multiInputSingleOutput-in-1.content-type=string/employee")) { + context.getBean(InputDestination.class); + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + Message stringInputMessage = MessageBuilder.withPayload("ricky".getBytes()).build(); + Message integerInputMessage = MessageBuilder.withPayload("bobby".getBytes()).build(); + inputDestination.send(stringInputMessage, 0); + inputDestination.send(integerInputMessage, 1); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage.getPayload()).isEqualTo("RICKY".getBytes()); + outputMessage = outputDestination.receive(); + assertThat(outputMessage.getPayload()).isEqualTo("BOBBY".getBytes()); + } + } + + @Test + public void testMultiInputSingleOutputWithCustomContentType2() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + ContentTypeConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.function.definition=multiInputSingleOutput2", + "--spring.cloud.stream.bindings.multiInputSingleOutput2-in-0.content-type=string/person", + "--spring.cloud.stream.bindings.multiInputSingleOutput2-in-1.content-type=string/employee", + "--spring.cloud.stream.bindings.multiInputSingleOutput2-out-0.content-type=string/person")) { + context.getBean(InputDestination.class); + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + Message stringInputMessage = MessageBuilder.withPayload("ricky".getBytes()).build(); + Message integerInputMessage = MessageBuilder.withPayload("bobby".getBytes()).build(); + inputDestination.send(stringInputMessage, 0); + inputDestination.send(integerInputMessage, 1); + + Message outputMessage = outputDestination.receive(2500); + assertThat(outputMessage.getPayload()).isEqualTo("rickybobby".getBytes()); +// outputMessage = outputDestination.receive(); +// assertThat(outputMessage.getPayload()).isEqualTo("BOBBY".getBytes()); + } + } + @EnableAutoConfiguration public static class ReactiveFunctionConfiguration { @@ -289,4 +351,104 @@ public class MultipleInputOutputFunctionTests { }; } } + + @EnableAutoConfiguration + public static class ContentTypeConfiguration { + + @Bean + public Function, Flux>, Flux> multiInputSingleOutput() { + return tuple -> { + Flux stringStream = tuple.getT1().map(p -> p.getName().toUpperCase()); + Flux intStream = tuple.getT2().map(p -> p.getName().toUpperCase()); + return Flux.merge(stringStream, intStream); + }; + } + + @Bean + public Function, Flux>, Flux> multiInputSingleOutput2() { + return tuple -> { + return Flux.merge(tuple.getT1(), tuple.getT2()).buffer(Duration.ofMillis(1000)).map(list -> { + String personName = ((Person) list.get(0)).getName(); + String employeeName = ((Employee) list.get(1)).getName(); + return new Person(personName + employeeName); + }); + }; + } + + @Bean + public MessageConverter stringToPersonConverter() { + return new AbstractMessageConverter(MimeType.valueOf("string/person")) { + + @Override + protected boolean supports(Class clazz) { + return Person.class.isAssignableFrom(clazz); + } + + @Nullable + protected Object convertFromInternal( + Message message, Class targetClass, @Nullable Object conversionHint) { + String name = new String(((byte[]) message.getPayload()), StandardCharsets.UTF_8); + Person person = new Person(name); + return person; + } + + @Nullable + protected Object convertToInternal( + Object payload, @Nullable MessageHeaders headers, @Nullable Object conversionHint) { + + return ((Person) payload).getName().getBytes(StandardCharsets.UTF_8); + } + }; + } + + @Bean + public MessageConverter stringToEmployeeConverter() { + return new AbstractMessageConverter(MimeType.valueOf("string/employee")) { + + @Override + protected boolean supports(Class clazz) { + return Employee.class.isAssignableFrom(clazz); + } + + @Nullable + protected Object convertFromInternal( + Message message, Class targetClass, @Nullable Object conversionHint) { + String name = new String(((byte[]) message.getPayload()), StandardCharsets.UTF_8); + Employee person = new Employee(name); + return person; + } + + @Nullable + protected Object convertToInternal( + Object payload, @Nullable MessageHeaders headers, @Nullable Object conversionHint) { + + return ((Employee) payload).getName().getBytes(StandardCharsets.UTF_8); + } + }; + } + } + + private static class Person { + private final String name; + + Person(String name) { + this.name = name; + } + + public String getName() { + return name; + } + } + + private static class Employee { + private final String name; + + Employee(String name) { + this.name = name; + } + + public String getName() { + return name; + } + } }