From 872edb2f24be8231badc84a62dfe6bb39dfbf4cd Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Fri, 14 Jan 2022 15:55:12 +0100 Subject: [PATCH] GH-2265 Add support for creating explicit bindings Resolves #2265 --- README.adoc | 30 ++++------ docs/src/main/asciidoc/_configprops.adoc | 4 +- .../main/asciidoc/spring-cloud-stream.adoc | 39 +++++++++++++ .../config/BindingServiceProperties.java | 40 ++++++++++++- .../function/FunctionConfiguration.java | 56 +++++++++++++----- .../stream/binding/ExplicitBindingTests.java | 58 +++++++++++++++++++ 6 files changed, 189 insertions(+), 38 deletions(-) create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ExplicitBindingTests.java diff --git a/README.adoc b/README.adoc index d5b27d22c..ebcca497c 100644 --- a/README.adoc +++ b/README.adoc @@ -175,31 +175,21 @@ You can also build and package your application into a boot jar (by using `./mvn Now you have a working (albeit very basic) Spring Cloud Stream application. -== What's New in 3.x? +[[spel-and-streaming-data]] -[[spring-cloud-stream-preface-new-features]] -=== New Features and Enhancements +== Spring Expression Language (SpEL) in the context of Streaming data -- *Routing Function* - see <> for more details. -- *StreamBridge* - for dynamic destinations. See <> for more details. -- *Multiple bindings with functions* (multiple message handlers) - see <> for more details. -- *Functions with multiple inputs/outputs* (single function that can subscribe or target multiple destinations) - see <> for more details. -- *Native support for reactive programming* - since v3.0.0 we no longer distribute spring-cloud-stream-reactive modules and instead -relying on native reactive support provided by spring cloud function. For backward -compatibility you can still bring `spring-cloud-stream-reactive` from previous versions. +Throughout this reference manual you will encounter many features and examples where you can utilize Spring Expression Language (SpEL). It is important to understand certain limitations when it comes to using it. +SpEL gives you access to the current Message as well as the Application Context you are running in. +However it is important to understand what type of data SpEL can see especially in the context of the incoming Message. +From the broker, the message arrives in a form of a byte[]. It is then transformed to a `Message` by the binders where as you can see the payload of the message maintains its raw form. The headers of the message are ``, where values are typically another primitive or a collection/array of primitives, hence Object. +That is because binder does not know the required input type as it has no access to the user code (function). So effectively binder delivered an envelope with the payload and some readable meta-data in the form of message headers, just like the letter delivered by mail. +This means that while accessing payload of the message is possible you will only have access to it as raw data (i.e., byte[]). And while it may be very common for developers to ask for ability to have SpEL access to fields of a payload object as concrete type (e.g., Foo, Bar etc), you can see how difficult or even impossible would it be to achieve. +Here is one example to demonstrate the problem; Imagine you have a routing expression to route to different functions based on payload type. This requirement would imply payload conversion from byte[] to a specific type and then applying the SpEL. However, in order to perform such conversion we would need to know the actual type to pass to converter and that comes from function's signature which we don’t know which one. A better approach to solve this requirement would be to pass the type information as message headers (e.g., `application/json;type=foo.bar.Baz` ). You’ll get a clear readable String value that could be accessed and evaluated in a year and easy to read SpEL expression. -[[spring-cloud-stream-preface-notable-deprecations]] -=== Notable Deprecations and Removals - -- Annotation-based programming model is now fully removed. Basically the @EnableBInding, @StreamListener and all related annotations are now removed in -favor of the functional programming model. See <> for more details. -- _Reactive module_ (`spring-cloud-stream-reactive`) is discontinued and no longer distributed in favor of native support via spring-cloud-function. -For backward -compatibility you can still bring `spring-cloud-stream-reactive` from previous versions. -- _Test support binder_ `spring-cloud-stream-test-support` with MessageCollector in favor of a new test binder. See <> for more details. -- _@StreamMessageConverter_ - Removed as it is no longer required. +Additionally it is considered very bad practice to use payload for routing decisions, since the payload is considered to be privileged data - data only to be read by its final recipient. Again, using the mail delivery analogy you would not want the mailman to open your envelope and read the contents of the letter to make some delivery decisions. The same concept applies here, especially when it is relatively easy to include such information when generating a Message. It enforces certain level of discipline related to the design of data to be transmitted over the network and which pieces of such data can be considered as public and which are privileged. = Appendices [appendix] diff --git a/docs/src/main/asciidoc/_configprops.adoc b/docs/src/main/asciidoc/_configprops.adoc index 45e42323f..55f3103e3 100644 --- a/docs/src/main/asciidoc/_configprops.adoc +++ b/docs/src/main/asciidoc/_configprops.adoc @@ -9,6 +9,7 @@ |spring.cloud.stream.dynamic-destinations | `[]` | A list of destinations that can be bound dynamically. If set, only listed destinations can be bound. |spring.cloud.stream.function.batch-mode | `false` | |spring.cloud.stream.function.bindings | | +|spring.cloud.stream.input-bindings | | A semi-colon delimited string to explicitly define input bindings (specifically for cases when there is no implicit trigger to create such bindings such as Function, Supplier or Consumer). |spring.cloud.stream.instance-count | `1` | The number of deployed instances of an application. Default: 1. NOTE: Could also be managed per individual binding "spring.cloud.stream.bindings.foo.consumer.instance-count" where 'foo' is the name of the binding. |spring.cloud.stream.instance-index | `0` | 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 "spring.cloud.stream.bindings.foo.consumer.instance-index" where 'foo' is the name of the binding. |spring.cloud.stream.instance-index-list | | A list of instance id's from 0 to instanceCount-1. Used for partitioning and with Kafka. NOTE: Could also be managed per individual binding "spring.cloud.stream.bindings.foo.consumer.instance-index-list" where 'foo' is the name of the binding. This setting will override the one set in 'spring.cloud.stream.instance-index' @@ -18,9 +19,10 @@ |spring.cloud.stream.metrics.meter-filter | | Pattern to control the 'meters' one wants to capture. By default all 'meters' will be captured. For example, 'spring.integration.*' will only capture metric information for meters whose name starts with 'spring.integration'. |spring.cloud.stream.metrics.properties | | Application properties that should be added to the metrics payload For example: `spring.application**`. |spring.cloud.stream.metrics.schedule-interval | `60s` | Interval expressed as Duration for scheduling metrics snapshots publishing. Defaults to 60 seconds +|spring.cloud.stream.output-bindings | | A semi-colon delimited string to explicitly define output bindings (specifically for cases when there is no implicit trigger to create such bindings such as Function, Supplier or Consumer). |spring.cloud.stream.override-cloud-connectors | `false` | This property is only applicable when the cloud profile is active and Spring Cloud Connectors are provided with the application. If the property is false (the default), the binder detects a suitable bound service (for example, a RabbitMQ service bound in Cloud Foundry for the RabbitMQ binder) and uses it for creating connections (usually through Spring Cloud Connectors). When set to true, this property instructs binders to completely ignore the bound services and rely on Spring Boot properties (for example, relying on the spring.rabbitmq.* properties provided in the environment for the RabbitMQ binder). The typical usage of this property is to be nested in a customized environment when connecting to multiple systems. |spring.cloud.stream.pollable-source | `none` | A semi-colon delimited list of binding names of pollable sources. Binding names follow the same naming convention as functions. For example, name '...pollable-source=foobar' will be accessible as 'foobar-iin-0'' binding |spring.cloud.stream.sendto.destination | `none` | The name of the header used to determine the name of the output destination -|spring.cloud.stream.source | | 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) +|spring.cloud.stream.source | | A semi-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) @deprecated use {@link #outputBindings} |=== \ No newline at end of file diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 678563d14..99a632239 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -336,6 +336,45 @@ where you are clearly correlating the input of `uppercase` function to `sample-t For more on properties and other configuration options please see <> section. [[spring-cloud-stream-overview-producing-consuming-messages]] + +===== Explicit binding creation + +In the previous section we explained how bindings are created implicitly driven by Function, Supplier or Consumer provided by your application. +However, there are times when you may need to create binding explicitly where bindings are not tied to any function. This is typically done to +support integrations with other frameworks (e.g., Spring Integration framework) where you may need direct access to the underlying `MessageChannel`. + +Spring Cloud Stream allows you to define input and output bindings explicitly via `spring.cloud.stream.input-bindings` and `spring.cloud.stream.output-bindings` +properties. Noticed the plural in the property names allowing you to define multiple bindings by simply using `;` as a delimiter. +Just look at the following test case as an example: + +---- +@Test +public void testExplicitBindings() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(EmptyConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.stream.input-bindings=fooin;barin", + "--spring.cloud.stream.output-bindings=fooout;barout")) { + + assertThat(context.getBean("fooin-in-0", MessageChannel.class)).isNotNull(); + assertThat(context.getBean("barin-in-0", MessageChannel.class)).isNotNull(); + assertThat(context.getBean("fooout-out-0", MessageChannel.class)).isNotNull(); + assertThat(context.getBean("barout-out-0", MessageChannel.class)).isNotNull(); + } +} + +@EnableAutoConfiguration +@Configuration +public static class EmptyConfiguration { +} +---- + +As you can see we have declared two input bindings and two output bindings while our configuration had no functions defined, yet we were able to successfully create these bindings and access their corresponding channels. + +The rest of the binding rules that apply to implicit bindings apply here as well (for example, you can see that `fooin` turned into `fooin-in-0` binding/channel etc). + + === Producing and Consuming Messages You can write a Spring Cloud Stream application by simply writing functions and exposing them as `@Bean`s. 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 fcebda21f..015d5dea1 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 @@ -61,12 +61,26 @@ 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. + * A semi-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) + * @deprecated use {@link #outputBindings} */ + @Deprecated private String source; + /** + * A semi-colon delimited string to explicitly define input bindings (specifically for cases when there + * is no implicit trigger to create such bindings such as Function, Supplier or Consumer). + */ + private String inputBindings; + + /** + * A semi-colon delimited string to explicitly define output bindings (specifically for cases when there + * is no implicit trigger to create such bindings such as Function, Supplier or Consumer). + */ + private String outputBindings; + /** * 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 @@ -302,10 +316,18 @@ public class BindingServiceProperties this.bindingRetryInterval = bindingRetryInterval; } + /** + * @deprecated in favor of {@link #getOutputBindings()} + */ + @Deprecated public String getSource() { return source; } + /** + * @deprecated in favor of {@link #setOutputBindings()} + */ + @Deprecated public void setSource(String source) { this.source = source; } @@ -325,6 +347,22 @@ public class BindingServiceProperties this.dynamicDestinationCacheSize = dynamicDestinationCacheSize; } + public String getInputBindings() { + return inputBindings; + } + + public void setInputBindings(String inputBindings) { + this.inputBindings = inputBindings; + } + + public String getOutputBindings() { + return outputBindings; + } + + public void setOutputBindings(String outputBindings) { + this.outputBindings = outputBindings; + } + /* * The "necessary" implies the scenario where only defaults are defined. */ 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 fd0b7cd1f..0ab3784af 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 @@ -129,6 +129,8 @@ public class FunctionConfiguration { private final static String SOURCE_PROPERY = "spring.cloud.stream.source"; +// private final static String OUT_BINDINGS = "spring.cloud.stream.output-bindings"; + @Bean public StreamBridge streamBridgeUtils(FunctionCatalog functionCatalog, FunctionRegistry functionRegistry, BindingServiceProperties bindingServiceProperties, ConfigurableApplicationContext applicationContext, @@ -847,29 +849,51 @@ public class FunctionConfiguration { } } - if (StringUtils.hasText(this.environment.getProperty(SOURCE_PROPERY))) { - String[] sourceNames = this.environment.getProperty(SOURCE_PROPERY).split(";"); + this.createStandAloneBindingsIfNecessary(registry, applicationContext.getBean(BindingServiceProperties.class)); - for (String sourceName : sourceNames) { - FunctionInvocationWrapper sourceFunc = functionCatalog.lookup(sourceName); - - if (sourceFunc == null || //see https://github.com/spring-cloud/spring-cloud-stream/issues/2229 - (!sourceFunc.getFunctionDefinition().equals(sourceName) && applicationContext.containsBean(sourceName))) { - 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"); } } + private void createStandAloneBindingsIfNecessary(BeanDefinitionRegistry registry, BindingServiceProperties bindingProperties) { + String[] inputBindings = StringUtils.hasText(bindingProperties.getInputBindings()) + ? bindingProperties.getInputBindings().split(";") : new String[0]; + + String[] outputBindings = StringUtils.hasText(bindingProperties.getSource()) ? bindingProperties.getSource().split(";") : ( + StringUtils.hasText(bindingProperties.getOutputBindings()) ? bindingProperties.getOutputBindings().split(";") : new String[0] + ); + for (String inputBindingName : inputBindings) { + FunctionInvocationWrapper sourceFunc = functionCatalog.lookup(inputBindingName); + + if (sourceFunc == null || //see https://github.com/spring-cloud/spring-cloud-stream/issues/2229 + (!sourceFunc.getFunctionDefinition().equals(inputBindingName) && applicationContext.containsBean(inputBindingName))) { + RootBeanDefinition functionBindableProxyDefinition = new RootBeanDefinition(BindableFunctionProxyFactory.class); + functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(inputBindingName); + functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(1); + functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(0); + functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.streamFunctionProperties); + registry.registerBeanDefinition(inputBindingName + "_binding", functionBindableProxyDefinition); + } + } + + for (String outputBindingName : outputBindings) { + FunctionInvocationWrapper sourceFunc = functionCatalog.lookup(outputBindingName); + + if (sourceFunc == null || //see https://github.com/spring-cloud/spring-cloud-stream/issues/2229 + (!sourceFunc.getFunctionDefinition().equals(outputBindingName) && applicationContext.containsBean(outputBindingName))) { + RootBeanDefinition functionBindableProxyDefinition = new RootBeanDefinition(BindableFunctionProxyFactory.class); + functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(outputBindingName); + functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(0); + functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(1); + functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.streamFunctionProperties); + registry.registerBeanDefinition(outputBindingName + "_binding", functionBindableProxyDefinition); + } + } + + } + @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = (ConfigurableApplicationContext) applicationContext; diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ExplicitBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ExplicitBindingTests.java new file mode 100644 index 000000000..5ab5f9bd5 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ExplicitBindingTests.java @@ -0,0 +1,58 @@ +/* + * Copyright 2022-2022 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.binding; + +import org.junit.jupiter.api.Test; + +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.MessageChannel; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * + * + */ +public class ExplicitBindingTests { + + @Test + public void testExplicitBindings() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(EmptyConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.stream.input-bindings=fooin;barin", + "--spring.cloud.stream.output-bindings=fooout;barout")) { + + assertThat(context.getBean("fooin-in-0", MessageChannel.class)).isNotNull(); + assertThat(context.getBean("barin-in-0", MessageChannel.class)).isNotNull(); + assertThat(context.getBean("fooout-out-0", MessageChannel.class)).isNotNull(); + assertThat(context.getBean("barout-out-0", MessageChannel.class)).isNotNull(); + } + } + + @EnableAutoConfiguration + @Configuration + public static class EmptyConfiguration { + + } +}