From 7a3f0f819746887d6e5d08edbe75a608755b0549 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Mon, 12 Dec 2022 16:26:50 -0500 Subject: [PATCH] StreamOperations interface for StreamBridge Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2548 --- .../stream/function/StreamBridgeTests.java | 11 ++ .../cloud/stream/function/StreamBridge.java | 74 +----------- .../stream/function/StreamOperations.java | 110 ++++++++++++++++++ .../main/asciidoc/spring-cloud-stream.adoc | 2 + 4 files changed, 128 insertions(+), 69 deletions(-) create mode 100644 core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamOperations.java diff --git a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java index a36fc8e4e..f33002af3 100644 --- a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java +++ b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java @@ -85,6 +85,17 @@ public class StreamBridgeTests { System.clearProperty("spring.cloud.function.definition"); } + @Test + void extractStreamBridgeAsStreamOperations() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(EmptyConfiguration.class)) + .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false")) { + StreamOperations streamOperations = context.getBean(StreamOperations.class); + StreamBridge streamBridge = context.getBean(StreamBridge.class); + assertThat(streamOperations).isSameAs(streamBridge); + } + } + @Test void test_SCF_856() throws Exception { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java index 56e49f33a..a2551421d 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java @@ -75,9 +75,9 @@ import org.springframework.util.StringUtils; * */ @SuppressWarnings("rawtypes") -public final class StreamBridge implements SmartInitializingSingleton { +public final class StreamBridge implements StreamOperations, SmartInitializingSingleton { - private static String STREAM_BRIDGE_FUNC_NAME = "streamBridge"; + private static final String STREAM_BRIDGE_FUNC_NAME = "streamBridge"; private final Log logger = LogFactory.getLog(getClass()); @@ -87,9 +87,9 @@ public final class StreamBridge implements SmartInitializingSingleton { private final NewDestinationBindingCallback destinationBindingCallback; - private BindingServiceProperties bindingServiceProperties; + private final BindingServiceProperties bindingServiceProperties; - private ConfigurableApplicationContext applicationContext; + private final ConfigurableApplicationContext applicationContext; private boolean initialized; @@ -97,7 +97,7 @@ public final class StreamBridge implements SmartInitializingSingleton { private final Map streamBridgeFunctionCache; - private FunctionInvocationHelper functionInvocationHelper; + private final FunctionInvocationHelper functionInvocationHelper; /** * @@ -133,83 +133,19 @@ public final class StreamBridge implements SmartInitializingSingleton { this.streamBridgeFunctionCache = new HashMap<>(); } - /** - * Sends 'data' to an output binding specified by 'bindingName' argument while - * using default content type to deal with output type conversion (if necessary). - * For typical cases `bindingName` is configured using 'spring.cloud.stream.source' property. - * However, this operation also supports sending to truly dynamic destinations. This means if the name - * provided via 'bindingName' does not have a corresponding binding such name will be - * treated as dynamic destination.
- * Will use default binder. For specific binder type see {@link #send(String, String, Object)} and {@link #send(String, String, Object, MimeType)} methods. - * @param bindingName the name of the output binding. That said it requires a bit of clarification. - * When using bridge.send("foo"...), the 'foo' typically represents the binding name. However - * if such binding does not exist, the new binding will be created to support dynamic destinations. - * @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) { BindingProperties bindingProperties = this.bindingServiceProperties.getBindingProperties(bindingName); MimeType contentType = StringUtils.hasText(bindingProperties.getContentType()) ? MimeType.valueOf(bindingProperties.getContentType()) : MimeTypeUtils.APPLICATION_JSON; return this.send(bindingName, data, contentType); } - /** - * 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). - * For typical cases `bindingName` is configured using 'spring.cloud.stream.source' property. - * However, this operation also supports sending to truly dynamic destinations. This means if the name - * provided via 'bindingName' does not have a corresponding binding such name will be - * treated as dynamic destination.
- * Will use default binder. For specific binder type see {@link #send(String, String, Object)} and {@link #send(String, String, Object, MimeType)} methods. - * @param bindingName the name of the output binding. That said it requires a bit of clarification. - * When using bridge.send("foo"...), the 'foo' typically represents the binding name. However - * if such binding does not exist, the new binding will be created to support dynamic destinations. - * @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. - */ public boolean send(String bindingName, Object data, MimeType outputContentType) { return this.send(bindingName, null, data, outputContentType); } - - /** - * 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). - * For typical cases `bindingName` is configured using 'spring.cloud.stream.source' property. - * However, this operation also supports sending to truly dynamic destinations. This means if the name - * provided via 'bindingName' does not have a corresponding binding such name will be - * treated as dynamic destination. - * - * @param bindingName the name of the output binding. That said it requires a bit of clarification. - * When using bridge.send("foo"...), the 'foo' typically represents the binding name. However - * if such binding does not exist, the new binding will be created to support dynamic destinations. - * @param binderName the name of the binder to use (e.g., 'kafka', 'rabbit') for cases where multiple binders are used. Can be null. - * @param data the data to send - * @return true if data was sent successfully, otherwise false or throws an exception. - */ public boolean send(String bindingName, @Nullable String binderName, Object data) { return this.send(bindingName, binderName, 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). - * For typical cases `bindingName` is configured using 'spring.cloud.stream.source' property. - * However, this operation also supports sending to truly dynamic destinations. This means if the name - * provided via 'bindingName' does not have a corresponding binding such name will be - * treated as dynamic destination. - * - * @param bindingName the name of the output binding. That said it requires a bit of clarification. - * When using bridge.send("foo"...), the 'foo' typically represents the binding name. However - * if such binding does not exist, the new binding will be created to support dynamic destinations. - * @param binderName the name of the binder to use (e.g., 'kafka', 'rabbit') for cases where multiple binders are used. Can be null. - * @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, @Nullable String binderName, Object data, MimeType outputContentType) { if (!this.initialized) { diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamOperations.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamOperations.java new file mode 100644 index 000000000..1cf7843a9 --- /dev/null +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamOperations.java @@ -0,0 +1,110 @@ +/* + * 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.function; + +import org.springframework.lang.Nullable; +import org.springframework.util.MimeType; + +/** + * Basic contract for {@link StreamBridge} operations. + * + * @author Oleg Zhurakousky + * @author Soby Chacko + * + * @since 4.0.0 + */ +public interface StreamOperations { + + /** + * Sends 'data' to an output binding specified by 'bindingName' argument while + * using default content type to deal with output type conversion (if necessary). + * For typical cases `bindingName` is configured using 'spring.cloud.stream.source' property. + * However, this operation also supports sending to truly dynamic destinations. This means if the name + * provided via 'bindingName' does not have a corresponding binding such name will be + * treated as dynamic destination.
+ * Will use default binder. For specific binder type see {@link #send(String, String, Object)} and + * {@link #send(String, String, Object, MimeType)} methods. + * + * @param bindingName the name of the output binding. That said it requires a bit of clarification. + * When using send("foo"...), the 'foo' typically represents the binding name. + * However, if such binding does not exist, the new binding will be created + * to support dynamic destinations. + * @param data the data to send + * @return true if data was sent successfully, otherwise false or throws an exception. + */ + boolean send(String bindingName, Object data); + + /** + * 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). + * For typical cases `bindingName` is configured using 'spring.cloud.stream.source' property. + * However, this operation also supports sending to truly dynamic destinations. This means if the name + * provided via 'bindingName' does not have a corresponding binding such name will be + * treated as dynamic destination.
+ * Will use default binder. For specific binder type see {@link #send(String, String, Object)} and + * {@link #send(String, String, Object, MimeType)} methods. + * + * @param bindingName the name of the output binding. That said it requires a bit of clarification. + * When using bridge.send("foo"...), the 'foo' typically represents the binding name. + * However, if such binding does not exist, the new binding will be created to support + * dynamic destinations. + * @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. + */ + boolean send(String bindingName, Object data, MimeType outputContentType); + + /** + * 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). + * For typical cases `bindingName` is configured using 'spring.cloud.stream.source' property. + * However, this operation also supports sending to truly dynamic destinations. This means if the name + * provided via 'bindingName' does not have a corresponding binding such name will be + * treated as dynamic destination. + * + * @param bindingName the name of the output binding. That said it requires a bit of clarification. + * When using bridge.send("foo"...), the 'foo' typically represents the binding name. + * However, if such binding does not exist, the new binding will be created to support dynamic destinations. + * @param binderName the name of the binder to use (e.g., 'kafka', 'rabbit') for cases where multiple + * binders are used. Can be null. + * @param data the data to send + * @return true if data was sent successfully, otherwise false or throws an exception. + */ + boolean send(String bindingName, @Nullable String binderName, Object data); + + /** + * 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). + * For typical cases `bindingName` is configured using 'spring.cloud.stream.source' property. + * However, this operation also supports sending to truly dynamic destinations. This means if the name + * provided via 'bindingName' does not have a corresponding binding such name will be + * treated as dynamic destination. + * + * @param bindingName the name of the output binding. That said it requires a bit of clarification. + * When using bridge.send("foo"...), the 'foo' typically represents the binding name. + * However, if such binding does not exist, the new binding will be created to support dynamic destinations. + * @param binderName the name of the binder to use (e.g., 'kafka', 'rabbit') for cases where multiple binders are used. Can be null. + * @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. + */ + boolean send(String bindingName, @Nullable String binderName, Object data, MimeType outputContentType); + +} diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index afb2274f0..de881f520 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -824,6 +824,8 @@ You have the flexibility to make the patterns more strict or customized to your With this approach, the application gets the ability to decide which interceptors to inject in `StreamBridge` rather than applying all the available interceptors. +NOTE: `StreamBridge` provides a contract through the `StreamOperations` interface that contains all the `send` methods of `StreamBridge`. Therefore, applications may choose to autowire using `StreamOperations`. This is handy when it comes to unit testing code that uses `StreamBridge` by providing a mock or similar mechanisms for the `StreamOperations` interface. + ===== Reactive Functions support