SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3054 Clear channel cache in StreamBridge during the refresh

Browse Source 
Resolves #3054
This commit is contained in:
Oleg Zhurakousky
2024-12-19 12:04:55 +01:00
parent 637636dfa8
commit 3f8b7003e7
1 changed files with 13 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -53,6 +53,8 @@ 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.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.core.ResolvableType;
import org.springframework.integration.channel.AbstractMessageChannel;
@@ -95,7 +97,7 @@ import static org.springframework.cloud.stream.utils.CacheKeyCreatorUtils.create
*
*/
@SuppressWarnings("rawtypes")
public final class StreamBridge implements StreamOperations, SmartInitializingSingleton, DisposableBean {
public final class StreamBridge implements StreamOperations, SmartInitializingSingleton, DisposableBean, ApplicationListener<ApplicationEvent> {
private static final String STREAM_BRIDGE_FUNC_NAME = "streamBridge";
@@ -135,6 +137,7 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
* @param bindingServiceProperties instance of {@link BindingServiceProperties}
* @param applicationContext instance of {@link ConfigurableApplicationContext}
*/
@SuppressWarnings("serial")
StreamBridge(FunctionCatalog functionCatalog, BindingServiceProperties bindingServiceProperties,
ConfigurableApplicationContext applicationContext, @Nullable NewDestinationBindingCallback destinationBindingCallback, ObjectProvider<ObservationRegistry> observationRegistries) {
this.executorService = Executors.newCachedThreadPool();
@@ -292,12 +295,9 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
messageChannel = this.isAsync() ? new ExecutorChannel(this.executorService) : new DirectWithAttributesChannel();
((AbstractSubscribableChannel) messageChannel).setApplicationContext(applicationContext);
((AbstractSubscribableChannel) messageChannel).setComponentName(destinationName);
//<<<<<<< HEAD
BinderWrapper binderWrapper = bindingService.createBinderWrapper(binderName, destinationName, messageChannel.getClass());
//=======
((AbstractSubscribableChannel) messageChannel).registerObservationRegistry(observationRegistry);
//>>>>>>> a1418283c (GH-3033: Register ObservationRegistry for Dynamic MessageChannels)
if (this.destinationBindingCallback != null) {
Object extendedProducerProperties = this.bindingService
.getExtendedProducerProperties(binderWrapper.binder(), destinationName);
@@ -369,6 +369,15 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
this.async = async;
}
// see https://github.com/spring-cloud/spring-cloud-stream/issues/3054
@Override
public void onApplicationEvent(ApplicationEvent event) {
// we need to do it by String to avoid cloud-bus and context dependencies
if (event.getClass().getName().equals("org.springframework.cloud.bus.event.RefreshRemoteApplicationEvent")) {
this.channelCache.clear();
}
}
private static final class ContextPropagationHelper {
static ExecutorService wrap(ExecutorService executorService) {
return ContextExecutorService.wrap(executorService, () -> ContextSnapshotFactory.builder().build().captureAll());