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

GH-2563 Add warning when SB is sending to input binding

Browse Source 
Resolves #2563
This commit is contained in:
Oleg Zhurakousky
2023-01-11 18:53:20 +01:00
parent 130a295d04
commit 3be95190be
3 changed files with 32 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
View File

@@ -1128,7 +1128,7 @@ public class ImplicitFunctionBindingTests {
@EnableAutoConfiguration
public static class SupplierAndProcessorConfiguration {
Many<Message<String>> sink = Sinks.many().unicast().onBackpressureBuffer();
@Bean
public Supplier<Flux<Message<String>>> supplier() {
return () -> sink.asFlux().doOnNext(v -> {

37
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java
View File

@@ -85,26 +85,39 @@ public class StreamBridgeTests {
public static void before() {
System.clearProperty("spring.cloud.function.definition");
}
@Test // see SCF-985
void ensurePassThruFunctionIsNotPrePostProcessed() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(EmptyConfiguration.class))
.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false")) {
TestChannelBinderConfiguration.getCompleteConfiguration(EmptyConfiguration.class))
.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false")) {
StreamBridge streamBridge = context.getBean(StreamBridge.class);
OutputDestination outputDestination = context.getBean(OutputDestination.class);
Message<String> message = CloudEventMessageBuilder.withData("foo") // cloud event
.setType("CustomType")
.setSource("CustomSource")
.setSubject("CustomSubject")
.build();
.setType("CustomType").setSource("CustomSource").setSubject("CustomSubject").build();
streamBridge.send("fooDestination", message);
Message<byte[]> messageReceived = outputDestination.receive(1000, "fooDestination");
assertThat(CloudEventMessageUtils.getSource(messageReceived)).isEqualTo(URI.create("CustomSource"));
assertThat(CloudEventMessageUtils.getSubject(messageReceived)).isEqualTo("CustomSubject");
assertThat(CloudEventMessageUtils.getType(messageReceived)).isEqualTo("CustomType");
Message<byte[]> messageReceived = outputDestination.receive(1000, "fooDestination");
assertThat(CloudEventMessageUtils.getSource(messageReceived)).isEqualTo(URI.create("CustomSource"));
assertThat(CloudEventMessageUtils.getSubject(messageReceived)).isEqualTo("CustomSubject");
assertThat(CloudEventMessageUtils.getType(messageReceived)).isEqualTo("CustomType");
}
}
@Test //
void validateWARNifSendingToINputBinding() { // GH-2563
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(SimpleConfiguration.class))
.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false",
"--spring.cloud.function.definition=echo",
"--spring.cloud.stream.function.bindings.echo-in-0=input")) {
StreamBridge streamBridge = context.getBean(StreamBridge.class);
OutputDestination outputDestination = context.getBean(OutputDestination.class);
streamBridge.send("input", "foo");
// just validate that there is a WARN message about sending to the input binding
}
}

6
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
View File

@@ -55,6 +55,7 @@ import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
/**
@@ -220,6 +221,11 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
if (messageChannel == null) {
if (this.applicationContext.containsBean(destinationName)) {
messageChannel = this.applicationContext.getBean(destinationName, MessageChannel.class);
String[] consumerBindingNames = this.bindingService.getConsumerBindingNames();
if (ObjectUtils.containsElement(consumerBindingNames, destinationName)) { //GH-2563
logger.warn("You seem to be sending data to the input binding. It is not "
+ "recommended, since you are bypassing the binder and this the messaging system exposed by the binder.");
}
}
else {
messageChannel = new DirectWithAttributesChannel();