From 83fb344bd3d47adda9d05e8b974918e38403add9 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Sat, 22 Feb 2014 16:49:50 +0200 Subject: [PATCH] Add `delay()` EIP-method --- .../integration/dsl/DelayerEndpointSpec.java | 62 +++++++++++++++++++ .../integration/dsl/FilterEndpointSpec.java | 1 - .../integration/dsl/GenericEndpointSpec.java | 1 - .../dsl/IntegrationFlowBuilder.java | 14 +++++ ...slIntegrationConfigurationInitializer.java | 2 - .../dsl/test/IntegrationFlowTests.java | 32 +++++++++- 6 files changed, 107 insertions(+), 5 deletions(-) create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/DelayerEndpointSpec.java diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/DelayerEndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/DelayerEndpointSpec.java new file mode 100644 index 0000000..4eef421 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/DelayerEndpointSpec.java @@ -0,0 +1,62 @@ +/* + * Copyright 2014 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.dsl; + +import java.util.Arrays; +import java.util.LinkedList; +import java.util.List; + +import org.aopalliance.aop.Advice; + +import org.springframework.integration.dsl.core.ConsumerEndpointSpec; +import org.springframework.integration.handler.DelayHandler; +import org.springframework.integration.store.MessageGroupStore; + +/** + * @author Artem Bilan + */ +public final class DelayerEndpointSpec extends ConsumerEndpointSpec { + + private final List delayedAdvice = new LinkedList(); + + DelayerEndpointSpec(DelayHandler delayHandler) { + super(delayHandler); + this.target.getT2().setDelayedAdviceChain(this.delayedAdvice); + } + + public DelayerEndpointSpec defaultDelay(long defaultDelay) { + this.target.getT2().setDefaultDelay(defaultDelay); + return _this(); + } + + + public DelayerEndpointSpec ignoreExpressionFailures(boolean ignoreExpressionFailures) { + this.target.getT2().setIgnoreExpressionFailures(ignoreExpressionFailures); + return _this(); + } + + public DelayerEndpointSpec messageStore(MessageGroupStore messageStore) { + this.target.getT2().setMessageStore(messageStore); + return _this(); + } + + public DelayerEndpointSpec delayedAdvice(Advice... advice) { + this.delayedAdvice.addAll(Arrays.asList(advice)); + return _this(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java index d22f6df..99c3fda 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java @@ -22,7 +22,6 @@ import org.springframework.messaging.MessageChannel; /** * @author Artem Bilan - */ public final class FilterEndpointSpec extends ConsumerEndpointSpec { diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java index fd04107..d03037e 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java @@ -21,7 +21,6 @@ import org.springframework.messaging.MessageHandler; /** * @author Artem Bilan - */ public final class GenericEndpointSpec extends ConsumerEndpointSpec, H> { diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java index f0004d7..c4a14e3 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java @@ -31,6 +31,7 @@ import org.springframework.integration.filter.ExpressionEvaluatingSelector; import org.springframework.integration.filter.MessageFilter; import org.springframework.integration.filter.MethodInvokingSelector; import org.springframework.integration.handler.BridgeHandler; +import org.springframework.integration.handler.DelayHandler; import org.springframework.integration.transformer.ExpressionEvaluatingTransformer; import org.springframework.integration.transformer.GenericTransformer; import org.springframework.integration.transformer.MessageTransformingHandler; @@ -39,6 +40,7 @@ import org.springframework.integration.transformer.Transformer; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * @author Artem Bilan @@ -131,6 +133,18 @@ public final class IntegrationFlowBuilder { return this.register(new GenericEndpointSpec(new BridgeHandler()), endpointConfigurer); } + public IntegrationFlowBuilder delay(String groupId, String expression) { + return this.delay(groupId, expression, null); + } + + public IntegrationFlowBuilder delay(String groupId, String expression, EndpointConfigurer> endpointConfigurer) { + DelayHandler delayHandler = new DelayHandler(groupId); + if (StringUtils.hasText(expression)) { + delayHandler.setDelayExpression(PARSER.parseExpression(expression)); + } + return this.register(new GenericEndpointSpec(delayHandler), endpointConfigurer); + } + private IntegrationFlowBuilder registerOutputChannelIfCan(MessageChannel outputChannel) { this.flow.addComponent(outputChannel); if (this.currentComponent != null) { diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java index 5e28cb9..a2fe70f 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java @@ -125,8 +125,6 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig BeanDefinitionReaderUtils.registerWithGeneratedName(component, registry); } } -// registry.removeBeanDefinition(flowName); -// beanFactory.destroyBean(flowName); } } diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java index 1309bca..3679018 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java @@ -24,10 +24,13 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import java.io.File; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import org.aopalliance.aop.Advice; +import org.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; import org.hamcrest.Matchers; import org.junit.Test; import org.junit.runner.RunWith; @@ -137,6 +140,10 @@ public class IntegrationFlowTests { @Qualifier("methodInvokingInput") private DirectChannel methodInvokingInput; + @Autowired + @Qualifier("delayedAdvice") + private DelayedAdvice delayedAdvice; + @Test public void testPollingFlow() { for (int i = 0; i < 10; i++) { @@ -202,6 +209,7 @@ public class IntegrationFlowTests { reply = this.bridgeFlow2Output.receive(5000); assertNotNull(reply); assertEquals("test", reply.getPayload()); + assertTrue(this.delayedAdvice.getInvoked()); } @Test @@ -340,6 +348,10 @@ public class IntegrationFlowTests { @Configuration public static class ContextConfiguration3 { + @Autowired + @Qualifier("delayedAdvice") + private MethodInterceptor delayedAdvice; + @Bean public QueueChannel successChannel() { return MessageChannels.queue().get(); @@ -379,12 +391,30 @@ public class IntegrationFlowTests { public IntegrationFlow bridgeFlow2() { return IntegrationFlows.from(MessageChannels.direct("bridgeFlow2Input")) .bridge(c -> c.autoStartup(false).id("bridge")) + .delay("delayer", "200", c -> c.advice(this.delayedAdvice)) .channel(MessageChannels.queue("bridgeFlow2Output")) .get(); } } + @Component("delayedAdvice") + public static class DelayedAdvice implements MethodInterceptor { + + private final AtomicBoolean invoked = new AtomicBoolean(); + + @Override + public Object invoke(MethodInvocation invocation) throws Throwable { + this.invoked.set(true); + return invocation.proceed(); + } + + public Boolean getInvoked() { + return invoked.get(); + } + + } + @Configuration public static class ContextConfiguration4 { @@ -403,13 +433,13 @@ public class IntegrationFlowTests { }) .get(); } - @Bean public IntegrationFlow methodInvokingFlow() { return IntegrationFlows.from(MessageChannels.direct("methodInvokingInput")) .handle("greetingService", null) .get(); } + } @Component("greetingService")