diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 92ddbf90f..f7c102165 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -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 <>). 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 <> 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 <> 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>` 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 diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java index 6ecd4f5c5..0dfe9b72c 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java @@ -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)); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java index 7352b71e9..bea6c8264 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java @@ -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); } - } 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 86294b211..8b1eaeab9 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 @@ -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, Function { - private final FunctionInvocationWrapper function; - - @SuppressWarnings("rawtypes") - private final Function 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"); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/PartitionAwareFunctionWrapper.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/PartitionAwareFunctionWrapper.java new file mode 100644 index 000000000..d5fc01387 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/PartitionAwareFunctionWrapper.java @@ -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, Supplier { + + protected final Log logger = LogFactory.getLog(PartitionAwareFunctionWrapper.class); + + private final FunctionInvocationWrapper function; + + @SuppressWarnings("rawtypes") + private final Function 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); + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java new file mode 100644 index 000000000..563df07d2 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java @@ -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. + *

+ * This utility class allows user to do just that - bridge non-spring-cloud-stream applications + * with spring-cloud-stream 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 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 functionToInvoke = functionWrapper; + if (producerProperties != null && producerProperties.isPartitioned()) { + functionToInvoke = new PartitionAwareFunctionWrapper(functionWrapper, this.applicationContext, producerProperties); + } + + Message resultMessage = (Message) functionToInvoke.apply(data); + this.outputChannelsOnly.get(bindingName).send(resultMessage); + return true; + } + + @Override + public void afterSingletonsInstantiated() { + if (this.initialized) { + return; + } + Map channels = applicationContext.getBeansOfType(DirectWithAttributesChannel.class); + for (Entry 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> fr = new FunctionRegistration<>(v -> v, channelEntry.getKey()); + fr.type(FunctionType.from(Object.class).to(Object.class)); + this.functionRegistry.register(fr); + this.initialized = true; + } + } + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java new file mode 100644 index 000000000..a1b2053d1 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java @@ -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 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 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 supplier() { + return () -> "hello"; + } + } +}