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 #2805
Resolves #2813

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 @@ void testWithOutputContentTypeWildCardBindings() throws Exception {
 				.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 {

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

@@ -39,7 +39,6 @@
 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 @@ protected boolean removeEldestEntry(Map.Entry<String, MessageChannel> eldest) {
 
 	@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 boolean send(String bindingName, Object data, MimeType outputContentType)
 	}
 	@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