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

INT-4210: DSL: Make a final .log() Terminal

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4210

Previously use-case like:
```
.transform()
.log()
.get();
```
failed with the `Dispatcher has no subscribers` for an implicit channel populated by the `.wireTap()` because actually `LoggingHandler` is placed on the wire-tapped channel

Since logically it looks like end-user tries to have `LoggingHandler` as a terminal of the flow, we add `nullChannel` if `WireTapSpec` is in the end of the flow definition.

If user would like to continue flow (e.g. return reply to the `replyChannel`), he must end more EIP-methods after `.log()` (like before, of course)
This commit is contained in:
Artem Bilan
2017-01-19 17:27:31 -05:00
committed by Gary Russell
parent 8b591482c1
commit 36f3c533aa
3 changed files with 45 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -20,6 +20,7 @@ import java.util.Collection;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.function.Consumer;
@@ -42,6 +43,7 @@ import org.springframework.integration.channel.ReactiveChannel;
import org.springframework.integration.channel.interceptor.WireTap;
import org.springframework.integration.config.ConsumerEndpointFactoryBean;
import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.GenericSelector;
import org.springframework.integration.core.MessageSelector;
import org.springframework.integration.dsl.channel.MessageChannelSpec;
@@ -412,7 +414,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
if (this.currentMessageChannel == null || !(this.currentMessageChannel instanceof ChannelInterceptorAware)) {
channel(new DirectChannel());
}
addComponents(wireTapSpec.getComponentsToRegister());
addComponent(wireTapSpec);
((ChannelInterceptorAware) this.currentMessageChannel).addInterceptor(interceptor);
return _this();
}
@@ -2361,6 +2363,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* as a default logging category.
* <p> The full request {@link Message} will be logged.
* @return the current {@link IntegrationFlowDefinition}.
* @see #wireTap(WireTapSpec)
*/
public B log() {
return log(LoggingHandler.Level.INFO);
@@ -2374,6 +2377,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* <p> The full request {@link Message} will be logged.
* @param level the {@link LoggingHandler.Level}.
* @return the current {@link IntegrationFlowDefinition}.
* @see #wireTap(WireTapSpec)
*/
public B log(LoggingHandler.Level level) {
return log(level, (String) null);
@@ -2386,6 +2390,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* <p> The full request {@link Message} will be logged.
* @param category the logging category to use.
* @return the current {@link IntegrationFlowDefinition}.
* @see #wireTap(WireTapSpec)
*/
public B log(String category) {
return log(LoggingHandler.Level.INFO, category);
@@ -2399,6 +2404,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @param level the {@link LoggingHandler.Level}.
* @param category the logging category to use.
* @return the current {@link IntegrationFlowDefinition}.
* @see #wireTap(WireTapSpec)
*/
public B log(LoggingHandler.Level level, String category) {
return log(level, category, (Expression) null);
@@ -2414,6 +2420,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @param logExpression the SpEL expression to evaluate logger message at runtime
* against the request {@link Message}.
* @return the current {@link IntegrationFlowDefinition}.
* @see #wireTap(WireTapSpec)
*/
public B log(LoggingHandler.Level level, String category, String logExpression) {
Assert.hasText(logExpression);
@@ -2429,6 +2436,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @param <P> the expected payload type.
* against the request {@link Message}.
* @return the current {@link IntegrationFlowDefinition}.
* @see #wireTap(WireTapSpec)
*/
public <P> B log(Function<Message<P>, Object> function) {
Assert.notNull(function);
@@ -2444,6 +2452,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @param logExpression the {@link Expression} to evaluate logger message at runtime
* against the request {@link Message}.
* @return the current {@link IntegrationFlowDefinition}.
* @see #wireTap(WireTapSpec)
*/
public B log(Expression logExpression) {
return log(LoggingHandler.Level.INFO, logExpression);
@@ -2460,6 +2469,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @param logExpression the {@link Expression} to evaluate logger message at runtime
* against the request {@link Message}.
* @return the current {@link IntegrationFlowDefinition}.
* @see #wireTap(WireTapSpec)
*/
public B log(LoggingHandler.Level level, Expression logExpression) {
return log(level, null, logExpression);
@@ -2476,6 +2486,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @param logExpression the {@link Expression} to evaluate logger message at runtime
* against the request {@link Message}.
* @return the current {@link IntegrationFlowDefinition}.
* @see #wireTap(WireTapSpec)
*/
public B log(String category, Expression logExpression) {
return log(LoggingHandler.Level.INFO, category, logExpression);
@@ -2492,6 +2503,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @param <P> the expected payload type.
* against the request {@link Message}.
* @return the current {@link IntegrationFlowDefinition}.
* @see #wireTap(WireTapSpec)
*/
public <P> B log(LoggingHandler.Level level, Function<Message<P>, Object> function) {
return log(level, null, function);
@@ -2507,6 +2519,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @param <P> the expected payload type.
* against the request {@link Message}.
* @return the current {@link IntegrationFlowDefinition}.
* @see #wireTap(WireTapSpec)
*/
public <P> B log(String category, Function<Message<P>, Object> function) {
return log(LoggingHandler.Level.INFO, category, function);
@@ -2523,6 +2536,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @param <P> the expected payload type.
* against the request {@link Message}.
* @return the current {@link IntegrationFlowDefinition}.
* @see #wireTap(WireTapSpec)
*/
public <P> B log(LoggingHandler.Level level, String category, Function<Message<P>, Object> function) {
Assert.notNull(function);
@@ -2540,6 +2554,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @param logExpression the {@link Expression} to evaluate logger message at runtime
* against the request {@link Message}.
* @return the current {@link IntegrationFlowDefinition}.
* @see #wireTap(WireTapSpec)
*/
public B log(LoggingHandler.Level level, String category, Expression logExpression) {
LoggingHandler loggingHandler = new LoggingHandler(level);
@@ -2653,8 +2668,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
}
RecipientListRouter recipientListRouter = recipientListRouterSpec.get();
addComponent(recipientListRouter);
addComponents(recipientListRouterSpec.getComponentsToRegister());
addComponent(recipientListRouterSpec);
AggregatingMessageHandler aggregatingMessageHandler = aggregatorSpec.get().getT2();
addComponent(aggregatingMessageHandler);
ScatterGatherHandler messageHandler = new ScatterGatherHandler(recipientListRouter, aggregatingMessageHandler);
@@ -2874,6 +2888,12 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
+ "Add at lest '.bridge()' EIP-method before the end of flow.");
}
}
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);
}
return this.integrationFlow;

7
spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/WireTapSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2017 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.
@@ -16,7 +16,6 @@
package org.springframework.integration.dsl.channel;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -104,10 +103,10 @@ public class WireTapSpec extends IntegrationComponentSpec<WireTapSpec, WireTap>
@Override
public Collection<Object> getComponentsToRegister() {
if (this.selector != null) {
return Arrays.asList(this.selector, this.target);
return Collections.singleton(this.selector);
}
else {
return Collections.singletonList(this.target);
return null;
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2017 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.
@@ -49,6 +49,7 @@ import org.springframework.integration.dsl.IntegrationFlowAdapter;
import org.springframework.integration.dsl.IntegrationFlowDefinition;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.channel.MessageChannels;
import org.springframework.integration.handler.LoggingHandler;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -57,8 +58,7 @@ import org.springframework.messaging.handler.annotation.Header;
import org.springframework.scheduling.TriggerContext;
import org.springframework.stereotype.Component;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.util.StringUtils;
/**
@@ -66,8 +66,7 @@ import org.springframework.util.StringUtils;
*
* @since 5.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@RunWith(SpringRunner.class)
@DirtiesContext
public class FlowServiceTests {
@@ -82,13 +81,12 @@ public class FlowServiceTests {
private PollableChannel myFlowAdapterOutput;
@Test
public void testFlowService() {
public void testFlowServiceAndLogAsLastNoError() {
assertNotNull(this.myFlow);
QueueChannel replyChannel = new QueueChannel();
this.input.send(MessageBuilder.withPayload("foo").setReplyChannel(replyChannel).build());
Message<?> receive = replyChannel.receive(1000);
assertNotNull(receive);
assertEquals("FOO", receive.getPayload());
this.input.send(MessageBuilder.withPayload("foo").build());
Object result = this.myFlow.resultOverLoggingHandler.get();
assertNotNull(result);
assertEquals("FOO", result);
}
@Test
@@ -119,7 +117,9 @@ public class FlowServiceTests {
@Bean
public IntegrationFlow testGateway() {
return f -> f.gateway("processChannel", g -> g.replyChannel("replyChannel"));
return f -> f.gateway("processChannel", g -> g.replyChannel("replyChannel"))
.log()
.bridge(null);
}
@Bean
@@ -136,9 +136,15 @@ public class FlowServiceTests {
@Component
public static class MyFlow implements IntegrationFlow {
private final AtomicReference<Object> resultOverLoggingHandler = new AtomicReference<>();
@Override
public void configure(IntegrationFlowDefinition<?> f) {
f.<String, String>transform(String::toUpperCase);
f.<String, String>transform(String::toUpperCase)
.log(LoggingHandler.Level.ERROR, m -> {
resultOverLoggingHandler.set(m.getPayload());
return m;
});
}
}