Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-8950: Fix PublisherIntegrationFlow for AbstractEndpoint

Browse Source 
Fixes: #8950

The `IntegrationFlow.toReactivePublisher(true)` makes `PublisherIntegrationFlow`
to be stopped from the beginning and waiting for the `Publisher.subscribe()`.
However only `EndpointSpec` components are marked as `autoStartup(false)`.
The `MessageProducerSupport` is not included, therefore unexpected messages
are produced to the channel in the end of flow without subscribers.

* Check for `AbstractEndpoint` component type in the `PublisherIntegrationFlow`
and mark it as `setAutoStartup(false)`
* Ensure in a new `ReactiveStreamsTests.messageProducerIsNotStartedAutomatically()` test
that `MessageProducerSupport` is not started from the beginning,
but rather when we subscribe to the `Publisher` via `StepVerifier`

(cherry picked from commit 2b576c57d6)
This commit is contained in:
Artem Bilan
2024-02-23 13:00:16 -05:00
committed by Spring Builds
parent 7acb4c37a7
commit 30fa11f621
2 changed files with 53 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-integration-core/src/main/java/org/springframework/integration/dsl/PublisherIntegrationFlow.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-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.
@@ -22,6 +22,7 @@ import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import reactor.core.publisher.Flux;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.messaging.Message;
/**
@@ -48,8 +49,11 @@ class PublisherIntegrationFlow<T> extends StandardIntegrationFlow implements Pub
if (autoStartOnSubscribe) {
flux = flux.doOnSubscribe((sub) -> start());
for (Object component : integrationComponents.keySet()) {
if (component instanceof EndpointSpec) {
((EndpointSpec<?, ?, ?>) component).autoStartup(false);
if (component instanceof EndpointSpec<?, ?, ?> endpointSpec) {
endpointSpec.autoStartup(false);
}
else if (component instanceof AbstractEndpoint endpoint) {
endpoint.setAutoStartup(false);
}
}
}