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

Change naming convention for multiple in/out

Browse Source
This commit is contained in:
Oleg Zhurakousky
2019-09-30 10:56:25 -04:00
parent 60d59ca24e
commit ba49a81474
3 changed files with 173 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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 - `<functionName> + .in. + <index>`
* output - `<functionName> + .out. + <index>`
* input - `<functionName> + -in- + <index>`
* output - `<functionName> + -out- + <index>`
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 <<Configuration Options>> 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]

2
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java
View File

@@ -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;

162
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/MultipleInputOutputFunctionTests.java
View File

@@ -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<byte[]> stringInputMessage = MessageBuilder.withPayload("ricky".getBytes()).build();
Message<byte[]> integerInputMessage = MessageBuilder.withPayload("bobby".getBytes()).build();
inputDestination.send(stringInputMessage, 0);
inputDestination.send(integerInputMessage, 1);
Message<byte[]> 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<byte[]> stringInputMessage = MessageBuilder.withPayload("ricky".getBytes()).build();
Message<byte[]> integerInputMessage = MessageBuilder.withPayload("bobby".getBytes()).build();
inputDestination.send(stringInputMessage, 0);
inputDestination.send(integerInputMessage, 1);
Message<byte[]> 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<Tuple2<Flux<Person>, Flux<Employee>>, Flux<String>> multiInputSingleOutput() {
return tuple -> {
Flux<String> stringStream = tuple.getT1().map(p -> p.getName().toUpperCase());
Flux<String> intStream = tuple.getT2().map(p -> p.getName().toUpperCase());
return Flux.merge(stringStream, intStream);
};
}
@Bean
public Function<Tuple2<Flux<Person>, Flux<Employee>>, Flux<Person>> 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;
}
}
}