From 7e7688ce1bedf476e4670e9f3670a020a3792678 Mon Sep 17 00:00:00 2001
From: Soby Chacko <chackos@vmware.com>
Date: Tue, 19 Sep 2023 18:38:27 -0400
Subject: [PATCH] GH-2805: StreamBridge send and custom content-type

 - When StreamBridge#send is called with binder-name and custom content-type,
   it does not honor the content-type value, but default to application/json.
   Fixing this issue for this call path by explicitly checking for any custom
   content-type provided on the binding.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2805
Resolves #2813
---
 .../stream/function/StreamBridgeTests.java    | 21 +++++++++++++++++--
 .../cloud/stream/function/StreamBridge.java   | 13 ++++++++----
 2 files changed, 28 insertions(+), 6 deletions(-)

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 5428a4c78..57f1c8837 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
@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2022 the original author or authors.
+ * Copyright 2020-2023 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.
@@ -268,10 +268,27 @@ public class StreamBridgeTests {
 				.isEqualTo(MimeType.valueOf("application/json+foo"));
 			assertThat(output.receive(1000, "bar").getHeaders().get(MessageHeaders.CONTENT_TYPE))
 				.isEqualTo(MimeType.valueOf("application/blahblah+non-registered-foo"));
-
 		}
 	}
 
+	// See this issue for more details: https://github.com/spring-cloud/spring-cloud-stream/issues/2805
+	@Test
+	void testStreamBridgeSendWithBinderNameAndCustomContentType() throws Exception {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
+			.getCompleteConfiguration(ConsumerConfiguration.class, EmptyConfigurationWithCustomConverters.class))
+			.web(WebApplicationType.NONE).run(
+				"--spring.cloud.stream.bindings.foo.content-type=application/*+foo")) {
+			StreamBridge bridge = context.getBean(StreamBridge.class);
+			bridge.send("foo", "test-binder", "hello foo");
+
+			OutputDestination output = context.getBean(OutputDestination.class);
+
+			assertThat(output.receive(1000, "foo").getHeaders().get(MessageHeaders.CONTENT_TYPE))
+				.isEqualTo(MimeType.valueOf("application/json+foo"));
+		}
+	}
+
+
 	@SuppressWarnings("unchecked")
 	@Test
 	void testNoCachingOfStreamBridgeFunction() throws Exception {
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 8123f8a7f..af76d8f15 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
@@ -39,7 +39,6 @@ import org.springframework.cloud.stream.binder.BinderFactory;
 import org.springframework.cloud.stream.binder.ProducerProperties;
 import org.springframework.cloud.stream.binding.BindingService;
 import org.springframework.cloud.stream.binding.NewDestinationBindingCallback;
-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;
@@ -136,8 +135,7 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
 
 	@Override
 	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;
+		var contentType = determineContentType(bindingName, this.bindingServiceProperties);
 		return this.send(bindingName, data, contentType);
 	}
 
@@ -147,7 +145,14 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
 	}
 	@Override
 	public boolean send(String bindingName, @Nullable String binderName, Object data) {
-		return this.send(bindingName, binderName, data, MimeTypeUtils.APPLICATION_JSON);
+		var contentType = determineContentType(bindingName, this.bindingServiceProperties);
+		return this.send(bindingName, binderName, data, contentType);
+	}
+
+	private static MimeType determineContentType(String bindingName, BindingServiceProperties bindingServiceProperties) {
+		var bindingProperties = bindingServiceProperties.getBindingProperties(bindingName);
+		return StringUtils.hasText(bindingProperties.getContentType()) ?
+			MimeType.valueOf(bindingProperties.getContentType()) : MimeTypeUtils.APPLICATION_JSON;
 	}
 
 	@Override