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

GH-1916 Fix partition support for reactive functions

Browse Source 
Fixed output partition support for reactive functions. For imperative functions it was already addressed with https://github.com/spring-cloud/spring-cloud-stream/issues/1893
Added tests

Resolves #1916
This commit is contained in:
Oleg Zhurakousky
2020-02-24 16:39:24 +01:00
parent 1361e2b092
commit b5e9444d0c
2 changed files with 87 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
View File

@@ -84,6 +84,7 @@ import org.springframework.core.env.Environment;
import org.springframework.core.type.MethodMetadata;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.AbstractSubscribableChannel;
import org.springframework.integration.channel.MessageChannelReactiveUtils;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
@@ -333,6 +334,8 @@ public class FunctionConfiguration {
private static class FunctionToDestinationBinder implements InitializingBean, ApplicationContextAware {
protected final Log logger = LogFactory.getLog(getClass());
private GenericApplicationContext applicationContext;
private BindableProxyFactory[] bindableProxyFactories;
@@ -393,18 +396,33 @@ public class FunctionConfiguration {
SubscribableChannel inputChannel = this.applicationContext.getBean(inputBindingName, SubscribableChannel.class);
return MessageChannelReactiveUtils.toPublisher(inputChannel);
}).toArray(Publisher[]::new);
Object resultPublishers = function.apply(inputPublishers.length == 1 ? inputPublishers[0] : Tuples.fromArray(inputPublishers));
ProducerProperties producerProperties = this.serviceProperties.getBindings().get(outputBindingNames.iterator().next()).getProducer();
PartitionAwareFunction functionToInvoke = new PartitionAwareFunction(function, this.applicationContext, producerProperties);
Object resultPublishers = functionToInvoke.apply(inputPublishers.length == 1 ? inputPublishers[0] : Tuples.fromArray(inputPublishers));
if (resultPublishers instanceof Iterable) {
Iterator<String> outputBindingIter = outputBindingNames.iterator();
((Iterable) resultPublishers).forEach(publisher -> {
MessageChannel outputChannel = this.applicationContext.getBean(outputBindingIter.next(), MessageChannel.class);
Flux.from((Publisher) publisher).doOnNext(message -> outputChannel.send((Message) message)).subscribe();
Flux.from((Publisher) publisher)
.onErrorContinue((ex, pay) -> {
logger.error("Failed to process the following content which will be dropped: " + pay, (Throwable) ex);
})
.doOnNext(message -> outputChannel.send((Message) message)).subscribe();
});
}
else {
outputBindingNames.stream().forEach(outputBindingName -> {
MessageChannel outputChannel = this.applicationContext.getBean(outputBindingName, MessageChannel.class);
Flux.from((Publisher) resultPublishers).doOnNext(message -> outputChannel.send((Message) message)).subscribe();
Flux.from((Publisher) resultPublishers)
.onErrorContinue((ex, pay) -> {
logger.error("Failed to process the following content which will be dropped: " + pay, (Throwable) ex);
})
.doOnNext(message -> {
outputChannel.send((Message) message);
}).subscribe();
});
}
}
@@ -412,8 +430,8 @@ public class FunctionConfiguration {
String outputDestinationName = this.determineOutputDestinationName(0, bindableProxyFactory, functionType);
this.adjustFunctionForNativeEncodingIfNecessary(outputDestinationName, function, 0);
if (this.functionProperties.isComposeFrom()) {
SubscribableChannel outputChannel = this.applicationContext.getBean(outputDestinationName, SubscribableChannel.class);
// logger.info("Composing at the head of 'output' channel");
AbstractSubscribableChannel outputChannel = this.applicationContext.getBean(outputDestinationName, AbstractSubscribableChannel.class);
logger.info("Composing at the head of output destination: " + outputChannel.getBeanName());
String outputChannelName = ((AbstractMessageChannel) outputChannel).getBeanName();
ServiceActivatingHandler handler = createFunctionHandler(function, null, outputChannelName);
@@ -490,8 +508,10 @@ public class FunctionConfiguration {
}
private boolean isReactiveOrMultipleInputOutput(BindableProxyFactory bindableProxyFactory, Type functionType) {
boolean reactiveInputsOutputs = FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, 0)) ||
FunctionTypeUtils.isReactive(FunctionTypeUtils.getOutputType(functionType, 0));
return isMultipleInputOutput(bindableProxyFactory)
|| (FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, 0))
|| (reactiveInputsOutputs
&& StringUtils.hasText(this.determineOutputDestinationName(0, bindableProxyFactory, functionType)));
}
@@ -575,7 +595,7 @@ public class FunctionConfiguration {
this.function = new PartitionAwareFunction((FunctionInvocationWrapper) function, this.applicationContext, producerProperties);
Type type = ((FunctionInvocationWrapper) function).getFunctionType();
if (FunctionTypeUtils.isReactive(FunctionTypeUtils.getOutputType(type, 0))) {
throw new IllegalStateException("Functions with mixed semantics (imperative input vs. reactive output) ar not supported at the moment");
//throw new IllegalStateException("Functions with mixed semantics (imperative input vs. reactive output) ar not supported at the moment");
}
this.consumerProperties = consumerProperties;
this.producerProperties = producerProperties;

60
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
View File

@@ -158,6 +158,26 @@ public class ImplicitFunctionBindingTests {
}
}
@Test
public void testHeaderRetentionWithComposition() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(NoEnableBindingConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false", "--spring.cloud.function.definition=func|addHeaders")) {
InputDestination inputDestination = context.getBean(InputDestination.class);
OutputDestination outputDestination = context.getBean(OutputDestination.class);
Message<byte[]> inputMessage = MessageBuilder.withPayload("Hello".getBytes()).build();
inputDestination.send(inputMessage);
Message<byte[]> outputMessage = outputDestination.receive();
assertThat(outputMessage.getPayload()).isEqualTo("Hello".getBytes());
}
}
@Test
public void testReactiveFunctionWithState() {
@@ -577,6 +597,29 @@ public class ImplicitFunctionBindingTests {
}
}
@Test
public void partitionOnOutputPayloadTestReactive() {
System.clearProperty("spring.cloud.function.definition");
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
.getCompleteConfiguration(PojoFunctionConfiguration.class))
.web(WebApplicationType.NONE).run("--spring.cloud.function.definition=funcReactive",
"--spring.cloud.stream.bindings.funcReactive-out-0.producer.partitionKeyExpression=payload.id",
"--spring.cloud.stream.bindings.funcReactive-out-0.producer.partitionCount=5",
"--spring.jmx.enabled=false")) {
InputDestination inputDestination = context.getBean(InputDestination.class);
OutputDestination outputDestination = context.getBean(OutputDestination.class);
Message<byte[]> inputMessage = MessageBuilder.withPayload("Jim Lahey".getBytes()).build();
inputDestination.send(inputMessage, "funcReactive-in-0");
assertThat(outputDestination.receive(100, "funcReactive-out-0").getHeaders().get("scst_partition")).isEqualTo(3);
assertThat(outputDestination.receive(100)).isNull();
}
}
@Test
public void partitionOnOutputPayloadWithSupplierTest() {
System.clearProperty("spring.cloud.function.definition");
@@ -645,6 +688,13 @@ public class ImplicitFunctionBindingTests {
};
}
@Bean
public Function<Flux<String>, Flux<Message<String>>> addHeaders() {
return flux -> flux.map(value -> {
return MessageBuilder.withPayload(value).setHeader("foo", "bar").build();
});
}
@Bean
public Function<Flux<String>, Flux<String>> aggregate() {
return inbound -> inbound.log().window(Duration.ofSeconds(1))
@@ -854,6 +904,16 @@ public class ImplicitFunctionBindingTests {
return person;
};
}
@Bean
public Function<Flux<String>, Flux<Person>> funcReactive() {
return flux -> flux.map(value -> {
Person person = new Person();
person.setName(value);
person.setId(3);
return person;
});
}
}
public static class Person {