GH-1912 Added support for non-reactive-source messaging bridge
Currently the use of reactive EmitterProcessor is the only way to connect foreign sources such as HTTP with spring-cloud-stream The newly added StreamBridgeUtils class provides a simpler mechanism to brideg non-stream application with stream application without requiring reactor API while also honoring the same type conversion and partitioning contract used by functions Added docs Resolves #1912 Address PR comments Resolves #1918
This commit is contained in:
Oleg Zhurakousky
@@ -614,10 +614,55 @@ For example `--spring.cloud.stream.poller.fixed-delay=2000` sets the poller inte
|
||||
|
||||
===== Foreign event-driven sources
|
||||
|
||||
There are cases where the actual source of data may be coming from outside system that is not a binder. For example, the
|
||||
source of the data may be a classic web endpoint. How do we bridge such source with the functional Supplier?
|
||||
There are cases where the actual source of data may be coming from the external (foreign) system that is not a binder. For example, the
|
||||
source of the data may be a classic REST endpoint. How do we bridge such source with the functional mechanism used by spring-cloud-stream?
|
||||
|
||||
Let's look at a simple example:
|
||||
Spring Cloud Stream provides two mechanisms, so let's look at them in more details
|
||||
|
||||
Here, for both samples we'll use a standard MVC endpoint method called `delegateToSupplier` bound to the root web context,
|
||||
delegating incoming requests to stream via two different mechanisms -
|
||||
imperative (via StreamBridgeUtils) and reactive (via EmitterProcessor).
|
||||
|
||||
====== Using StreamBridgeUtils
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@SpringBootApplication
|
||||
@Controller
|
||||
public class WebSourceApplication {
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(WebSourceApplication.class, "--spring.cloud.stream.source=toStream");
|
||||
}
|
||||
|
||||
@Autowired
|
||||
private StreamBridgeUtils streamBridge;
|
||||
|
||||
@RequestMapping
|
||||
@ResponseStatus(HttpStatus.ACCEPTED)
|
||||
public void delegateToSupplier(@RequestBody String body) {
|
||||
System.out.println("Sending " + body);
|
||||
streamBridge.send("toStream-out-0", body);
|
||||
}
|
||||
}
|
||||
----
|
||||
|
||||
Here we autowire a `StreamBridgeUtils` bean which allows us to send data to an output binding effectively
|
||||
bridging non-stream application with spring-cloud-stream. Note that preceding example does not have any
|
||||
source functions defined (e.g., Supplier bean) leaving the framework with no trigger to create source bindings, which would be typical for cases where
|
||||
configuration contains function beans.
|
||||
So to trigger the creation of source binding we use `spring.cloud.stream.source` property where you can declare the name of your sources.
|
||||
The provided name will be used as a trigger to create a source binding.
|
||||
So in the preceding example the name of the output binding will be `toStream-out-0` which is consistent with the binding naming
|
||||
convention used by functions (see <<Binding and Binding names>>). You can use `;` to signify multiple sources
|
||||
(e.g., `--spring.cloud.stream.source=foo;bar`)
|
||||
|
||||
Also, note that `streamBridge.send(..)` method takes an `Object` for data. This means you can send POJO or `Message` to it and it
|
||||
will go through the same routine when sending output as if it was from any Function or Supplier providing the same level
|
||||
of consistency as with functions. This means the output type conversion, partitioning etc are honored as if it was from the output produced by functions.
|
||||
|
||||
|
||||
====== Using reactor API
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@@ -644,11 +689,11 @@ public class WebSourceApplication {
|
||||
}
|
||||
----
|
||||
|
||||
Here you see a standard MVC endpoint method called `delegateToSupplier` bound to the root web context and a `Supplier` bean. Note how `Supplier` bean returns `Flux` of `Strings`.
|
||||
Yes we are benefiting from the reactive support (see <<Reactive Functions support>> for more details). Specifically
|
||||
https://projectreactor.io/docs/core/release/api/reactor/core/publisher/EmitterProcessor.html[EmitterProcessor] which effectively serves as bridge
|
||||
between the web and spring-cloud-stream. As you can see in the endpoint method, all we do is call `onNext(..)` operation of the `EmitterProcessor`
|
||||
with the value of the HTTP request's body. From that point on the Supplier rules described earlier apply.
|
||||
Here we declare a `Supplier` bean which returns `Flux` of `Strings`.
|
||||
This example uses https://projectreactor.io/docs/core/release/api/reactor/core/publisher/EmitterProcessor.html[EmitterProcessor]
|
||||
from the reactor API (see <<Reactive Functions support>> for more details) to effectively provide a
|
||||
bridge between the actual event source (rest endpoint in this case) and spring-cloud-stream. All you need to do
|
||||
is define a `Supplier<Flux<your-type>>` and return the `EmitterProcessor` while feeding the incoming data via `EmitterProcessor#onNext(data)` operation.
|
||||
|
||||
You can now send message to spring-cloud-stream source as
|
||||
|
||||
|
||||
@@ -143,7 +143,9 @@ public class MessageConverterConfigurer
|
||||
String contentType = bindingProperties.getContentType();
|
||||
ProducerProperties producerProperties = bindingProperties.getProducer();
|
||||
boolean partitioned = !inbound && producerProperties != null && producerProperties.isPartitioned();
|
||||
boolean functional = streamFunctionProperties != null && StringUtils.hasText(streamFunctionProperties.getDefinition());
|
||||
boolean functional = streamFunctionProperties != null
|
||||
&& (StringUtils.hasText(streamFunctionProperties.getDefinition()) || StringUtils.hasText(bindingServiceProperties.getSource()));
|
||||
|
||||
if (partitioned) {
|
||||
if (inbound || !functional) {
|
||||
messageChannel.addInterceptor(new PartitioningInterceptor(bindingProperties));
|
||||
|
||||
@@ -60,6 +60,13 @@ public class BindingServiceProperties
|
||||
|
||||
private static final int DEFAULT_BINDING_RETRY_INTERVAL = 30;
|
||||
|
||||
/**
|
||||
* A colon delimited string representing the names of the sources based on which source bindings will be created.
|
||||
* This is primarily to support cases where source binding may be required without providing a corresponding Supplier.
|
||||
* (e.g., for cases where the actual source of data is outside of scope of spring-cloud-stream - HTTP -> Stream)
|
||||
*/
|
||||
private String source;
|
||||
|
||||
/**
|
||||
* The instance id of the application: a number from 0 to instanceCount-1. Used for
|
||||
* partitioning and with Kafka. NOTE: Could also be managed per individual binding
|
||||
@@ -288,6 +295,14 @@ public class BindingServiceProperties
|
||||
this.bindingRetryInterval = bindingRetryInterval;
|
||||
}
|
||||
|
||||
public String getSource() {
|
||||
return source;
|
||||
}
|
||||
|
||||
public void setSource(String source) {
|
||||
this.source = source;
|
||||
}
|
||||
|
||||
public void updateProducerProperties(String bindingName,
|
||||
ProducerProperties producerProperties) {
|
||||
if (this.bindings.containsKey(bindingName)) {
|
||||
@@ -318,5 +333,4 @@ public class BindingServiceProperties
|
||||
Bindable.ofInstance(bindingPropertiesTarget));
|
||||
this.bindings.put(binding, bindingPropertiesTarget);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -48,6 +48,7 @@ import org.springframework.beans.factory.support.RootBeanDefinition;
|
||||
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
|
||||
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.cloud.function.context.FunctionCatalog;
|
||||
import org.springframework.cloud.function.context.FunctionProperties;
|
||||
@@ -60,10 +61,8 @@ import org.springframework.cloud.function.context.config.ContextFunctionCatalogA
|
||||
import org.springframework.cloud.function.context.config.FunctionContextUtils;
|
||||
import org.springframework.cloud.function.context.config.RoutingFunction;
|
||||
import org.springframework.cloud.stream.annotation.EnableBinding;
|
||||
import org.springframework.cloud.stream.binder.BinderHeaders;
|
||||
import org.springframework.cloud.stream.binder.BindingCreatedEvent;
|
||||
import org.springframework.cloud.stream.binder.ConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.PartitionHandler;
|
||||
import org.springframework.cloud.stream.binder.ProducerProperties;
|
||||
import org.springframework.cloud.stream.binding.BindableProxyFactory;
|
||||
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
|
||||
@@ -84,14 +83,12 @@ import org.springframework.context.support.GenericApplicationContext;
|
||||
import org.springframework.core.annotation.AnnotationUtils;
|
||||
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;
|
||||
import org.springframework.integration.dsl.IntegrationFlows;
|
||||
import org.springframework.integration.expression.ExpressionUtils;
|
||||
import org.springframework.integration.handler.ServiceActivatingHandler;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.lang.Nullable;
|
||||
@@ -122,6 +119,15 @@ import org.springframework.util.StringUtils;
|
||||
@ConditionalOnBean(FunctionRegistry.class)
|
||||
public class FunctionConfiguration {
|
||||
|
||||
private final static String SOURCE_PROPERY = "spring.cloud.stream.source";
|
||||
|
||||
@Bean
|
||||
@ConditionalOnProperty(SOURCE_PROPERY)
|
||||
public StreamBridge streamBridgeUtils(FunctionCatalog functionCatalog, FunctionRegistry functionRegistry,
|
||||
BindingServiceProperties bindingServiceProperties, ConfigurableApplicationContext applicationContext) {
|
||||
return new StreamBridge(functionCatalog, functionRegistry, bindingServiceProperties, applicationContext);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public InitializingBean functionBindingRegistrar(Environment environment, FunctionCatalog functionCatalog,
|
||||
StreamFunctionProperties streamFunctionProperties) {
|
||||
@@ -141,7 +147,6 @@ public class FunctionConfiguration {
|
||||
? new FunctionToDestinationBinder(functionCatalog, functionProperties,
|
||||
serviceProperties, dynamicDestinationResolver)
|
||||
: null;
|
||||
|
||||
}
|
||||
|
||||
/*
|
||||
@@ -184,7 +189,7 @@ public class FunctionConfiguration {
|
||||
PollableBean pollable = extractPollableAnnotation(functionProperties, context, proxyFactory);
|
||||
|
||||
Type functionType = ((FunctionInvocationWrapper) functionWrapper).getFunctionType();
|
||||
IntegrationFlow integrationFlow = integrationFlowFromProvidedSupplier(new PartitionAwareFunction(functionWrapper, context, producerProperties),
|
||||
IntegrationFlow integrationFlow = integrationFlowFromProvidedSupplier(new PartitionAwareFunctionWrapper(functionWrapper, context, producerProperties),
|
||||
beginPublishingTrigger, pollable, context, taskScheduler, functionType)
|
||||
.route(Message.class, message -> {
|
||||
if (message.getHeaders().get("spring.cloud.stream.sendto.destination") != null) {
|
||||
@@ -287,54 +292,6 @@ public class FunctionConfiguration {
|
||||
: MessageBuilder.withPayload(value).build();
|
||||
}
|
||||
|
||||
/**
|
||||
* hHis class is effectively a wrapper which is aware of the stream related partition information
|
||||
* for outgoing messages. It has only one responsibility and that is to modify the result message
|
||||
* with 'scst_partition' header if necessary.
|
||||
*/
|
||||
private static class PartitionAwareFunction implements Supplier<Object>, Function<Object, Object> {
|
||||
private final FunctionInvocationWrapper function;
|
||||
|
||||
@SuppressWarnings("rawtypes")
|
||||
private final Function<Message, Message> outputMessageEnricher;
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
PartitionAwareFunction(FunctionInvocationWrapper function, ConfigurableApplicationContext context, ProducerProperties producerProperties) {
|
||||
this.function = function;
|
||||
if (producerProperties != null && producerProperties.isPartitioned()) {
|
||||
StandardEvaluationContext evaluationContext = ExpressionUtils.createStandardEvaluationContext(context.getBeanFactory());
|
||||
PartitionHandler partitionHandler = new PartitionHandler(evaluationContext, producerProperties, context.getBeanFactory());
|
||||
|
||||
this.outputMessageEnricher = outputMessage -> {
|
||||
int partitionId = partitionHandler.determinePartition(outputMessage);
|
||||
return MessageBuilder
|
||||
.fromMessage(outputMessage)
|
||||
.setHeader(BinderHeaders.PARTITION_HEADER, partitionId).build();
|
||||
};
|
||||
}
|
||||
else {
|
||||
this.outputMessageEnricher = null;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object apply(Object input) {
|
||||
if (this.outputMessageEnricher == null) { // to avoid breaking change
|
||||
return this.function.apply(input);
|
||||
}
|
||||
return this.function.apply(input, this.outputMessageEnricher);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object get() {
|
||||
if (this.outputMessageEnricher == null) { // to avoid breaking change
|
||||
return this.function.get();
|
||||
}
|
||||
return this.function.get(this.outputMessageEnricher);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private static class FunctionToDestinationBinder implements InitializingBean, ApplicationContextAware {
|
||||
|
||||
protected final Log logger = LogFactory.getLog(getClass());
|
||||
@@ -417,7 +374,7 @@ public class FunctionConfiguration {
|
||||
if (!CollectionUtils.isEmpty(outputBindingNames)) {
|
||||
BindingProperties bindingProperties = this.serviceProperties.getBindings().get(outputBindingNames.iterator().next());
|
||||
ProducerProperties producerProperties = bindingProperties == null ? null : bindingProperties.getProducer();
|
||||
functionToInvoke = new PartitionAwareFunction(function, this.applicationContext, producerProperties);
|
||||
functionToInvoke = new PartitionAwareFunctionWrapper(function, this.applicationContext, producerProperties);
|
||||
}
|
||||
|
||||
Object resultPublishers = functionToInvoke.apply(inputPublishers.length == 1 ? inputPublishers[0] : Tuples.fromArray(inputPublishers));
|
||||
@@ -588,7 +545,7 @@ public class FunctionConfiguration {
|
||||
|
||||
isRoutingFunction = ((FunctionInvocationWrapper) function).getTarget() instanceof RoutingFunction;
|
||||
this.applicationContext = applicationContext;
|
||||
this.function = new PartitionAwareFunction((FunctionInvocationWrapper) function, this.applicationContext, producerProperties);
|
||||
this.function = new PartitionAwareFunctionWrapper((FunctionInvocationWrapper) function, this.applicationContext, producerProperties);
|
||||
this.consumerProperties = consumerProperties;
|
||||
this.producerProperties = producerProperties;
|
||||
this.headersField = ReflectionUtils.findField(MessageHeaders.class, "headers");
|
||||
@@ -643,35 +600,53 @@ public class FunctionConfiguration {
|
||||
|
||||
@Override
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
if (ObjectUtils.isEmpty(applicationContext.getBeanNamesForAnnotation(EnableBinding.class))
|
||||
&& this.determineFunctionName(functionCatalog, environment)) {
|
||||
if (ObjectUtils.isEmpty(applicationContext.getBeanNamesForAnnotation(EnableBinding.class))) {
|
||||
this.determineFunctionName(functionCatalog, environment);
|
||||
BeanDefinitionRegistry registry = (BeanDefinitionRegistry) applicationContext.getBeanFactory();
|
||||
String[] functionDefinitions = streamFunctionProperties.getDefinition().split(";");
|
||||
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 = this.getOutputCount(functionType, true);
|
||||
}
|
||||
else if (function.isConsumer()) {
|
||||
this.inputCount = FunctionTypeUtils.getInputCount(functionType);
|
||||
this.outputCount = 0;
|
||||
}
|
||||
else {
|
||||
this.inputCount = FunctionTypeUtils.getInputCount(functionType);
|
||||
this.outputCount = this.getOutputCount(functionType, false);
|
||||
}
|
||||
|
||||
functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(functionDefinition);
|
||||
functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.inputCount);
|
||||
functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.outputCount);
|
||||
functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(streamFunctionProperties);
|
||||
registry.registerBeanDefinition(functionDefinition + "_binding", functionBindableProxyDefinition);
|
||||
if (StringUtils.hasText(streamFunctionProperties.getDefinition())) {
|
||||
String[] functionDefinitions = streamFunctionProperties.getDefinition().split(";");
|
||||
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 = this.getOutputCount(functionType, true);
|
||||
}
|
||||
else if (function.isConsumer()) {
|
||||
this.inputCount = FunctionTypeUtils.getInputCount(functionType);
|
||||
this.outputCount = 0;
|
||||
}
|
||||
else {
|
||||
this.inputCount = FunctionTypeUtils.getInputCount(functionType);
|
||||
this.outputCount = this.getOutputCount(functionType, false);
|
||||
}
|
||||
|
||||
functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(functionDefinition);
|
||||
functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.inputCount);
|
||||
functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.outputCount);
|
||||
functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.streamFunctionProperties);
|
||||
registry.registerBeanDefinition(functionDefinition + "_binding", functionBindableProxyDefinition);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (StringUtils.hasText(this.environment.getProperty(SOURCE_PROPERY))) {
|
||||
String[] sourceNames = this.environment.getProperty(SOURCE_PROPERY).split(";");
|
||||
|
||||
for (String sourceName : sourceNames) {
|
||||
if (functionCatalog.lookup(sourceName) == null) {
|
||||
RootBeanDefinition functionBindableProxyDefinition = new RootBeanDefinition(BindableFunctionProxyFactory.class);
|
||||
functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(sourceName);
|
||||
functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(0);
|
||||
functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(1);
|
||||
functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.streamFunctionProperties);
|
||||
registry.registerBeanDefinition(sourceName + "_binding", functionBindableProxyDefinition);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
else {
|
||||
logger.info("Functional binding is disabled due to the presense of @EnableBinding annotation in your configuration");
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
/*
|
||||
* Copyright 2020-2020 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.Function;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.cloud.function.context.catalog.BeanFactoryAwareFunctionRegistry.FunctionInvocationWrapper;
|
||||
import org.springframework.cloud.stream.binder.BinderHeaders;
|
||||
import org.springframework.cloud.stream.binder.PartitionHandler;
|
||||
import org.springframework.cloud.stream.binder.ProducerProperties;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.expression.spel.support.StandardEvaluationContext;
|
||||
import org.springframework.integration.expression.ExpressionUtils;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
|
||||
/**
|
||||
* This class is effectively a wrapper which is aware of the stream related partition information
|
||||
* for outgoing messages. It has only one responsibility and that is to modify the result message
|
||||
* with 'scst_partition' header if necessary.
|
||||
*/
|
||||
class PartitionAwareFunctionWrapper implements Function<Object, Object>, Supplier<Object> {
|
||||
|
||||
protected final Log logger = LogFactory.getLog(PartitionAwareFunctionWrapper.class);
|
||||
|
||||
private final FunctionInvocationWrapper function;
|
||||
|
||||
@SuppressWarnings("rawtypes")
|
||||
private final Function<Message, Message> outputMessageEnricher;
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
PartitionAwareFunctionWrapper(FunctionInvocationWrapper function, ConfigurableApplicationContext context, ProducerProperties producerProperties) {
|
||||
this.function = function;
|
||||
if (producerProperties != null && producerProperties.isPartitioned()) {
|
||||
StandardEvaluationContext evaluationContext = ExpressionUtils.createStandardEvaluationContext(context.getBeanFactory());
|
||||
PartitionHandler partitionHandler = new PartitionHandler(evaluationContext, producerProperties, context.getBeanFactory());
|
||||
|
||||
this.outputMessageEnricher = outputMessage -> {
|
||||
int partitionId = partitionHandler.determinePartition(outputMessage);
|
||||
return MessageBuilder
|
||||
.fromMessage(outputMessage)
|
||||
.setHeader(BinderHeaders.PARTITION_HEADER, partitionId).build();
|
||||
};
|
||||
}
|
||||
else {
|
||||
this.outputMessageEnricher = null;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object apply(Object input) {
|
||||
if (this.outputMessageEnricher == null) { // to avoid breaking change
|
||||
return this.function.apply(input);
|
||||
}
|
||||
try {
|
||||
return this.function.apply(input, this.outputMessageEnricher);
|
||||
}
|
||||
catch (NoSuchMethodError e) {
|
||||
logger.warn("Versions of spring-cloud-function older then 3.0.2.RELEASE do not support generation of partition information. "
|
||||
+ "Output message will not contain any partition header unless spring-cloud-function dependency is 3.0.2.RELEASE or higher.");
|
||||
return this.function.apply(input);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object get() {
|
||||
if (this.outputMessageEnricher == null) { // to avoid breaking change
|
||||
return this.function.get();
|
||||
}
|
||||
return this.function.get(this.outputMessageEnricher);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,140 @@
|
||||
/*
|
||||
* Copyright 2020-2020 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.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.Map.Entry;
|
||||
import java.util.function.Function;
|
||||
|
||||
import org.springframework.beans.factory.SmartInitializingSingleton;
|
||||
import org.springframework.cloud.function.context.FunctionCatalog;
|
||||
import org.springframework.cloud.function.context.FunctionRegistration;
|
||||
import org.springframework.cloud.function.context.FunctionRegistry;
|
||||
import org.springframework.cloud.function.context.FunctionType;
|
||||
import org.springframework.cloud.function.context.catalog.BeanFactoryAwareFunctionRegistry.FunctionInvocationWrapper;
|
||||
import org.springframework.cloud.stream.binder.ProducerProperties;
|
||||
import org.springframework.cloud.stream.config.BindingProperties;
|
||||
import org.springframework.cloud.stream.config.BindingServiceProperties;
|
||||
import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.MimeType;
|
||||
import org.springframework.util.MimeTypeUtils;
|
||||
|
||||
/**
|
||||
* A class which allows user to send data to an output binding.
|
||||
* While in a common scenario of a typical spring-cloud-stream application user rarely
|
||||
* has to manually send data, there are times when the sources of data are outside of
|
||||
* spring-cloud-stream context and therefore we need to bridge such foreign sources
|
||||
* with spring-cloud-stream.
|
||||
* <br><br>
|
||||
* This utility class allows user to do just that - <i>bridge non-spring-cloud-stream applications
|
||||
* with spring-cloud-stream</i> by providing a mechanism (bridge) to send data to an output binding while
|
||||
* maintaining the same invocation contract (i.e., type conversion, partitioning etc) as if it was
|
||||
* done through a declared function.
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
* @since 3.0.3
|
||||
*
|
||||
*/
|
||||
public final class StreamBridge implements SmartInitializingSingleton {
|
||||
|
||||
private final Map<String, DirectWithAttributesChannel> outputChannelsOnly = new HashMap<>();
|
||||
|
||||
private final FunctionCatalog functionCatalog;
|
||||
|
||||
private final FunctionRegistry functionRegistry;
|
||||
|
||||
private BindingServiceProperties bindingServiceProperties;
|
||||
|
||||
private ConfigurableApplicationContext applicationContext;
|
||||
|
||||
private boolean initialized;
|
||||
|
||||
/**
|
||||
*
|
||||
* @param functionCatalog instance of {@link FunctionCatalog}
|
||||
* @param functionRegistry instance of {@link FunctionRegistry}
|
||||
* @param bindingServiceProperties instance of {@link BindingServiceProperties}
|
||||
* @param applicationContext instance of {@link ConfigurableApplicationContext}
|
||||
*/
|
||||
StreamBridge(FunctionCatalog functionCatalog, FunctionRegistry functionRegistry,
|
||||
BindingServiceProperties bindingServiceProperties, ConfigurableApplicationContext applicationContext) {
|
||||
this.functionCatalog = functionCatalog;
|
||||
this.functionRegistry = functionRegistry;
|
||||
this.applicationContext = applicationContext;
|
||||
this.bindingServiceProperties = bindingServiceProperties;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends 'data' to an output binding specified by 'bindingName' argument while
|
||||
* using default content type to deal with output type conversion (if necessary).
|
||||
* @param bindingName the name of the output binding
|
||||
* @param data the data to send
|
||||
* @return true if data was sent successfully, otherwise false or throws an exception.
|
||||
*/
|
||||
public boolean send(String bindingName, Object data) {
|
||||
return this.send(bindingName, data, MimeTypeUtils.APPLICATION_JSON);
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends 'data' to an output binding specified by 'bindingName' argument while
|
||||
* using the content type specified by the 'outputContentType' argument to deal
|
||||
* with output type conversion (if necessary).
|
||||
* @param bindingName the name of the output binding
|
||||
* @param data the data to send
|
||||
* @param outputContentType content type to be used to deal with output type conversion
|
||||
* @return true if data was sent successfully, otherwise false or throws an exception.
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public boolean send(String bindingName, Object data, MimeType outputContentType) {
|
||||
Assert.isTrue(this.outputChannelsOnly.containsKey(bindingName), "Binding name '" + bindingName + "' does not exist.");
|
||||
FunctionInvocationWrapper functionWrapper = this.functionCatalog.lookup(bindingName, outputContentType.toString());
|
||||
|
||||
BindingProperties bindingProperties = this.bindingServiceProperties.getBindings().get(bindingName);
|
||||
ProducerProperties producerProperties = bindingProperties.getProducer();
|
||||
Function<Object, Object> functionToInvoke = functionWrapper;
|
||||
if (producerProperties != null && producerProperties.isPartitioned()) {
|
||||
functionToInvoke = new PartitionAwareFunctionWrapper(functionWrapper, this.applicationContext, producerProperties);
|
||||
}
|
||||
|
||||
Message<byte[]> resultMessage = (Message<byte[]>) functionToInvoke.apply(data);
|
||||
this.outputChannelsOnly.get(bindingName).send(resultMessage);
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterSingletonsInstantiated() {
|
||||
if (this.initialized) {
|
||||
return;
|
||||
}
|
||||
Map<String, DirectWithAttributesChannel> channels = applicationContext.getBeansOfType(DirectWithAttributesChannel.class);
|
||||
for (Entry<String, DirectWithAttributesChannel> channelEntry : channels.entrySet()) {
|
||||
if (channelEntry.getValue().getAttribute("type").equals("output")) {
|
||||
outputChannelsOnly.put(channelEntry.getKey(), channelEntry.getValue());
|
||||
// we're registering a dummy pass-through function to ensure that it goes through the
|
||||
// same process (type conversion, etc) as other function invocation.
|
||||
FunctionRegistration<Function<Object, Object>> fr = new FunctionRegistration<>(v -> v, channelEntry.getKey());
|
||||
fr.type(FunctionType.from(Object.class).to(Object.class));
|
||||
this.functionRegistry.register(fr);
|
||||
this.initialized = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,136 @@
|
||||
/*
|
||||
* Copyright 2020-2020 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.Supplier;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
|
||||
import org.springframework.boot.WebApplicationType;
|
||||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
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.messaging.Message;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.junit.Assert.fail;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
*
|
||||
*/
|
||||
public class StreamBridgeTests {
|
||||
|
||||
@Before
|
||||
public void before() {
|
||||
System.clearProperty("spring.cloud.function.definition");
|
||||
}
|
||||
|
||||
@Test(expected = NoSuchBeanDefinitionException.class)
|
||||
public void testNoBridgeIfNoSourcePropertyDefined() {
|
||||
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
|
||||
.getCompleteConfiguration())
|
||||
.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false")) {
|
||||
|
||||
context.getBean(StreamBridge.class);
|
||||
fail();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testBridgeFunctions() {
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
|
||||
.getCompleteConfiguration(EmptyConfiguration.class))
|
||||
.web(WebApplicationType.NONE).run("--spring.cloud.stream.source=foo;bar",
|
||||
"--spring.jmx.enabled=false")) {
|
||||
|
||||
StreamBridge bridge = context.getBean(StreamBridge.class);
|
||||
bridge.send("foo-out-0", "hello foo");
|
||||
bridge.send("bar-out-0", "hello bar");
|
||||
|
||||
|
||||
OutputDestination outputDestination = context.getBean(OutputDestination.class);
|
||||
assertThat(new String(outputDestination.receive(100, "foo-out-0").getPayload())).isEqualTo("hello foo");
|
||||
assertThat(new String(outputDestination.receive(100, "bar-out-0").getPayload())).isEqualTo("hello bar");
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testBridgeFunctionsWitthPartitionInformation() {
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
|
||||
.getCompleteConfiguration(EmptyConfiguration.class))
|
||||
.web(WebApplicationType.NONE).run("--spring.cloud.stream.source=foo;bar",
|
||||
"--spring.cloud.stream.bindings.foo-out-0.producer.partitionKeyExpression=payload",
|
||||
"--spring.cloud.stream.bindings.foo-out-0.producer.partitionCount=2",
|
||||
"--spring.jmx.enabled=false")) {
|
||||
|
||||
StreamBridge bridge = context.getBean(StreamBridge.class);
|
||||
bridge.send("foo-out-0", "a");
|
||||
bridge.send("bar-out-0", "b");
|
||||
|
||||
|
||||
OutputDestination outputDestination = context.getBean(OutputDestination.class);
|
||||
Message<byte[]> message = outputDestination.receive(100, "foo-out-0");
|
||||
assertThat(new String(message.getPayload())).isEqualTo("a");
|
||||
assertThat(message.getHeaders().get("scst_partition")).isEqualTo(1);
|
||||
assertThat(new String(outputDestination.receive(100, "bar-out-0").getPayload())).isEqualTo("b");
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSendingMessageToOutputOfExistingSupplier() {
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
|
||||
.getCompleteConfiguration(TestConfiguration.class))
|
||||
.web(WebApplicationType.NONE).run("--spring.cloud.stream.source=supplier;foo",
|
||||
"--spring.jmx.enabled=false")) {
|
||||
|
||||
StreamBridge bridge = context.getBean(StreamBridge.class);
|
||||
bridge.send("supplier-out-0", "blah");
|
||||
bridge.send("foo-out-0", "b");
|
||||
|
||||
|
||||
OutputDestination outputDestination = context.getBean(OutputDestination.class);
|
||||
Message<byte[]> message = outputDestination.receive(100, "foo-out-0");
|
||||
assertThat(new String(message.getPayload())).isEqualTo("b");
|
||||
message = outputDestination.receive(100, "supplier-out-0");
|
||||
assertThat(new String(message.getPayload())).isEqualTo("hello");
|
||||
message = outputDestination.receive(100, "supplier-out-0");
|
||||
assertThat(new String(message.getPayload())).isEqualTo("blah");
|
||||
}
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
public static class EmptyConfiguration {
|
||||
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
public static class TestConfiguration {
|
||||
|
||||
@Bean
|
||||
public Supplier<String> supplier() {
|
||||
return () -> "hello";
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user