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

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 {