GH-2885: Channel used by StreamBridge missing name

Fixes #2885

The `DirectWithAttributesChannel` used by `StreamBridge` is missing naming
information. Adding the proper application context and component name data
to the channel so that it is able to construct a name when queried.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2023 the original author or authors.
+ * Copyright 2020-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.
@@ -50,6 +50,7 @@
 import org.springframework.cloud.stream.binding.BindingService;
 import org.springframework.cloud.stream.binding.NewDestinationBindingCallback;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
+import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.integration.channel.AbstractMessageChannel;
@@ -414,6 +415,25 @@ void test_2268() {
 		}
 	}
 
+	// See https://github.com/spring-cloud/spring-cloud-stream/issues/2885 for more context on the following test
+	@SuppressWarnings("unchecked")
+	@Test
+	void ensureDirectWithAttributesChannelIsPopulatedWithName() throws Exception {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
+			.getCompleteConfiguration(InterceptorConfiguration.class))
+			.web(WebApplicationType.NONE).run(
+				"--spring.jmx.enabled=false")) {
+			StreamBridge bridge = context.getBean(StreamBridge.class);
+			bridge.send("test-channel", "blah");
+
+			Field field = ReflectionUtils.findField(StreamBridge.class, "channelCache");
+			Objects.requireNonNull(field).setAccessible(true);
+			Map<String, MessageChannel> map = (Map<String, MessageChannel>) field.get(bridge);
+			final MessageChannel messageChannel = map.get("test-channel");
+			assertThat(((DirectWithAttributesChannel) messageChannel).getFullChannelName()).isEqualTo("application.test-channel");
+		}
+	}
+
 	@Test
 	void interceptorIsNotAddedMultipleTimesToTheMessageChannel() {
 		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2023 the original author or authors.
+ * Copyright 2020-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.
@@ -254,6 +254,8 @@ synchronized MessageChannel resolveDestination(String destinationName, ProducerP
 			}
 			else {
 				messageChannel = new DirectWithAttributesChannel();
+				((DirectWithAttributesChannel) messageChannel).setApplicationContext(applicationContext);
+				((DirectWithAttributesChannel) messageChannel).setComponentName(destinationName);
 				if (this.destinationBindingCallback != null) {
 					Object extendedProducerProperties = this.bindingService
 							.getExtendedProducerProperties(messageChannel, destinationName);