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

Add IntegrationFlowDefinition.implicitChannel state to be sure that we don't add nullChannel if we wire-tap the real channel.

Browse Source 
Otherwise we end up with the case to break the flow which expect message from the explicit channel.
Add JavDocs to `.wireTap()` to explain behaviour in the end of flow
This commit is contained in:
Artem Bilan
2017-01-20 10:40:23 -05:00
committed by Gary Russell
parent 36f3c533aa
commit 6ac0c7316a
2 changed files with 13 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -127,6 +127,8 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
private StandardIntegrationFlow integrationFlow;
private boolean implicitChannel;
IntegrationFlowDefinition() {
}
@@ -396,7 +398,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* Populate the {@code Wire Tap} EI Pattern specific
* {@link org.springframework.messaging.support.ChannelInterceptor} implementation
* to the current {@link #currentMessageChannel}.
* It is useful when an implicit {@link MessageChannel} is used between endpoints:
* <p> It is useful when an implicit {@link MessageChannel} is used between endpoints:
* <pre class="code">
* {@code
* .transform("payload")
@@ -407,11 +409,14 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* This method can be used after any {@link #channel} for explicit {@link MessageChannel},
* but with the caution do not impact existing {@link org.springframework.messaging.support.ChannelInterceptor}s.
* @param wireTapSpec the {@link WireTapSpec} to use.
* <p> When this EIP-method is used in the end of flow, it appends {@code nullChannel} to terminate flow properly,
* Otherwise {@code Dispatcher has no subscribers} exception is thrown for implicit {@link DirectChannel}.
* @return the current {@link IntegrationFlowDefinition}.
*/
public B wireTap(WireTapSpec wireTapSpec) {
WireTap interceptor = wireTapSpec.get();
if (this.currentMessageChannel == null || !(this.currentMessageChannel instanceof ChannelInterceptorAware)) {
this.implicitChannel = true;
channel(new DirectChannel());
}
addComponent(wireTapSpec);
@@ -2889,9 +2894,11 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
}
}
Optional<Object> lastComponent = this.integrationComponents.stream().reduce((first, second) -> second);
if (lastComponent.get() instanceof WireTapSpec) {
// channel(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME);
if (this.implicitChannel) {
Optional<Object> lastComponent = this.integrationComponents.stream().reduce((first, second) -> second);
if (lastComponent.get() instanceof WireTapSpec) {
channel(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME);
}
}
this.integrationFlow = new StandardIntegrationFlow(this.integrationComponents);

3
spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java
View File

@@ -167,7 +167,8 @@ public class FlowServiceTests {
.enrichHeaders(Collections.singletonMap("foo", "FOO"))
.filter(this)
.handle(this)
.channel(MessageChannels.queue("myFlowAdapterOutput"));
.channel(MessageChannels.queue("myFlowAdapterOutput"))
.log();
}
public String messageSource() {