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

Fix a logic in the IntFlowDef.toReactivePublisher (#2553)

Browse Source 
* Fix a logic in the IntFlowDef.toReactivePublisher

* We may consider to start a `Publisher<Message<?>>` just from one
channel.
So, and an implicit `bridge()` to meet and internal `IntegrationFlow`
 logic
* The `MessageChannelReference` and `FixedSubscriberChannelPrototype`
can't be converted to the reactive `Publisher`.
So, an implicit `bridge()` in between them and target `FluxMessageChannel`

NOTE: This maybe considered for back-port, but the workaround is
simple: just add extra `bridge()` after the mentioned channels

* * Allow `log()` before `toReactivePublisher()` and the same time fix
the problem with not resetted `implicitChannel` flag in the
`IntegrationFlowDefinition`
This commit is contained in:
Artem Bilan
2018-09-17 10:50:29 -04:00
committed by Gary Russell
parent 54094da76b
commit 6c379d7199
2 changed files with 62 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
View File

@@ -210,6 +210,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
*/
public B channel(MessageChannel messageChannel) {
Assert.notNull(messageChannel, "'messageChannel' must not be null");
this.implicitChannel = false;
if (this.currentMessageChannel != null) {
bridge();
}
@@ -431,9 +432,9 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
*/
public B wireTap(WireTapSpec wireTapSpec) {
WireTap interceptor = wireTapSpec.get();
if (this.currentMessageChannel == null || !(this.currentMessageChannel instanceof ChannelInterceptorAware)) {
this.implicitChannel = true;
if (!(this.currentMessageChannel instanceof ChannelInterceptorAware)) {
channel(new DirectChannel());
this.implicitChannel = true;
}
addComponent(wireTapSpec);
((ChannelInterceptorAware) this.currentMessageChannel).addInterceptor(interceptor);
@@ -2941,7 +2942,9 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
publisher = (Publisher<Message<T>>) channelForPublisher;
}
else {
if (channelForPublisher != null) {
if (channelForPublisher != null && this.integrationComponents.size() > 1
&& !(channelForPublisher instanceof MessageChannelReference) &&
!(channelForPublisher instanceof FixedSubscriberChannelPrototype)) {
publisher = MessageChannelReactiveUtils.toPublisher(channelForPublisher);
}
else {
@@ -2951,6 +2954,8 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
}
}
this.implicitChannel = false;
get();
return new PublisherIntegrationFlow<>(this.integrationComponents, publisher);

54
spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java
View File

@@ -89,6 +89,18 @@ public class ReactiveStreamsTests {
@Autowired
private IntegrationFlowContext integrationFlowContext;
@Autowired
private MessageChannel singleChannel;
@Autowired
private Publisher<Message<String>> singleChannelFlow;
@Autowired
private MessageChannel fixedSubscriberChannel;
@Autowired
private Publisher<Message<String>> fixedSubscriberChannelFlow;
@Test
public void testReactiveFlow() throws Exception {
List<String> results = new ArrayList<>();
@@ -207,6 +219,30 @@ public class ReactiveStreamsTests {
integrationFlowRegistration.destroy();
}
@Test
public void singleChannelFlowTest() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
Flux.from(this.singleChannelFlow)
.map(m -> m.getPayload().toUpperCase())
.subscribe(p -> {
latch.countDown();
});
this.singleChannel.send(new GenericMessage<>("foo"));
assertTrue(latch.await(10, TimeUnit.SECONDS));
}
@Test
public void fixedSubscriberChannelFlowTest() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
Flux.from(this.fixedSubscriberChannelFlow)
.map(m -> m.getPayload().toUpperCase())
.subscribe(p -> {
latch.countDown();
});
this.fixedSubscriberChannel.send(new GenericMessage<>("bar"));
assertTrue(latch.await(10, TimeUnit.SECONDS));
}
@Configuration
@EnableIntegration
public static class ContextConfiguration {
@@ -221,6 +257,7 @@ public class ReactiveStreamsTests {
.autoStartup(false)
.id("reactiveStreamsMessageSource"))
.split(String.class, p -> p.split(","))
.log()
.toReactivePublisher();
}
@@ -231,6 +268,23 @@ public class ReactiveStreamsTests {
.split(s -> s.delimiters(","))
.<String, Integer>transform(Integer::parseInt)
.channel(MessageChannels.queue())
.log()
.toReactivePublisher();
}
@Bean
public Publisher<Message<String>> singleChannelFlow() {
return IntegrationFlows
.from(MessageChannels.direct("singleChannel"))
.log()
.toReactivePublisher();
}
@Bean
public Publisher<Message<String>> fixedSubscriberChannelFlow() {
return IntegrationFlows
.from("fixedSubscriberChannel", true)
.log()
.toReactivePublisher();
}