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 9b1620aa2..c47923cd5 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 @@ -217,6 +217,21 @@ class StreamBridgeTests { } } + // See for more details: https://github.com/spring-cloud/spring-cloud-stream/issues/2921 + @Test + void outputBindingsAndRabbitRoutingKeyExpressionComboDoesNotCauseIssues() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + EmptyConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.cloud.stream.output-bindings=the-exchange", + "--spring.cloud.stream.rabbit.bindings.the-exchange.producer.routing-key-expression=headers['key']")) { + // Ensure no exception is thrown from the above properties. + // See this issue for more details: https://github.com/spring-cloud/spring-cloud-stream/issues/2921 + // Without the corresponding chages added in this PR, this config would throw an exception. + + } + } + // Fore more context on this test: https://github.com/spring-cloud/spring-cloud-stream/issues/2848 @Test void ensurePartitioningWorksWhenNativeEncodingEnabledAndOutputBindingsExist() { diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java index dd7e4439f..64045feee 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2023 the original author or authors. + * Copyright 2015-2024 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. @@ -167,7 +167,24 @@ public class MessageConverterConfigurer if (environment != null && environment.containsProperty("spring.cloud.stream.rabbit.bindings." + channelName + ".producer.routing-key-expression")) { Map channelNameToFunctions = BindableFunctionProxyFactory.getChannelNameToFunctions(); String functionName = channelNameToFunctions.get(channelName); - this.setSkipOutputConversionIfNecessary(functionName); + String outputBindings = environment.getProperty("spring.cloud.stream.output-bindings"); + FunctionInvocationWrapper function = null; + if (outputBindings != null && outputBindings.contains(functionName)) { + try { + function = retrieveFunction(functionName); + } + catch (Exception e) { + // we expect an exception in this case since we didn't have a matching function + // for the function name due to output-bindings, hence passing through. + // See https://github.com/spring-cloud/spring-cloud-stream/issues/2921 for more details. + } + } + else { + function = retrieveFunction(functionName); + } + if (function != null) { + function.setSkipOutputConversion(true); + } functional = false; } if (!functional) { @@ -177,13 +194,19 @@ public class MessageConverterConfigurer } } - private void setSkipOutputConversionIfNecessary(String functionName) { + private FunctionInvocationWrapper retrieveFunction(String functionName) { FunctionCatalog catalog = this.beanFactory.getBean(FunctionCatalog.class); - if (catalog != null) { - FunctionInvocationWrapper function = catalog.lookup(functionName); - if (function != null) { - function.setSkipOutputConversion(true); - } + FunctionInvocationWrapper function = catalog.lookup(functionName); + if (function != null) { + function.setSkipOutputConversion(true); + } + return function; + } + + private void skipOutputConversionIfNecessary(String functionName) { + FunctionInvocationWrapper function = retrieveFunction(functionName); + if (function != null) { + function.setSkipOutputConversion(true); } } diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java index 38855bd98..0849f3522 100644 --- a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java +++ b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2023 the original author or authors. + * Copyright 2017-2024 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. @@ -25,26 +25,25 @@ import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.integration.channel.QueueChannel; -import org.springframework.messaging.Message; +import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.AbstractMessageConverter; -import org.springframework.messaging.converter.MessageConversionException; import org.springframework.messaging.converter.MessageConverter; import org.springframework.messaging.support.GenericMessage; import org.springframework.util.MimeType; -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.fail; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** * @author Gary Russell * @author Oleg Zhurakousky + * @author Soby Chacko * @since 1.3 * */ class MessageConverterConfigurerTests { -// @Test + @Test void configureOutputChannelWithBadContentType() { BindingServiceProperties props = new BindingServiceProperties(); BindingProperties bindingProps = new BindingProperties(); @@ -56,11 +55,8 @@ class MessageConverterConfigurerTests { converterFactory.getMessageConverterForAllRegistered()); QueueChannel out = new QueueChannel(); configurer.configureOutputChannel(out, "foo"); - out.send(new GenericMessage(new Foo(), Collections - .singletonMap(MessageHeaders.CONTENT_TYPE, "bad/ct"))); - Message received = out.receive(0); - assertThat(received).isNotNull(); - assertThat(received.getPayload()).isInstanceOf(Foo.class); + assertThatThrownBy(() -> out.send(new GenericMessage<>(new Foo(), Collections + .singletonMap(MessageHeaders.CONTENT_TYPE, "bad/ct")))).isInstanceOf(MessageDeliveryException.class); } @Test @@ -91,16 +87,9 @@ class MessageConverterConfigurerTests { converterFactory.getMessageConverterForAllRegistered()); QueueChannel out = new QueueChannel(); configurer.configureOutputChannel(out, "foo"); - try { - out.send(new GenericMessage(new Foo(), - Collections.singletonMap(MessageHeaders.CONTENT_TYPE, - "bad/ct"))); - fail("Expected MessageConversionException: " + out.receive(0)); - } - catch (MessageConversionException e) { - assertThat(e.getMessage()) - .endsWith("to the configured output type: 'foo/bar'"); - } + assertThatThrownBy(() -> out.send(new GenericMessage<>(new Foo(), + Collections.singletonMap(MessageHeaders.CONTENT_TYPE, + "bad/ct")))).isInstanceOf(MessageDeliveryException.class); } public static class Foo {