diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java index 756bab80ad..35117ed65e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -30,6 +30,7 @@ import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.support.MessageBuilder; +import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.Assert; @@ -62,7 +63,9 @@ import org.springframework.util.Assert; * @author Artem Bilan * @since 1.0.3 */ -public class DelayHandler extends AbstractReplyProducingMessageHandler implements ApplicationListener { + +@ManagedResource +public class DelayHandler extends AbstractReplyProducingMessageHandler implements DelayHandlerManagement, ApplicationListener { private final String messageGroupId; @@ -226,6 +229,10 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement } } + public int getDelayedMessageCount() { + return this.messageStore.messageGroupSize(this.messageGroupId); + } + /** * Used for reading persisted Messages in the 'messageStore' * to reschedule them e.g. upon application restart. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandlerManagement.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandlerManagement.java new file mode 100644 index 0000000000..0bae6d3b20 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandlerManagement.java @@ -0,0 +1,31 @@ +/* + * Copyright 2002-2012 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.handler; + +import org.springframework.jmx.export.annotation.ManagedAttribute; +import org.springframework.jmx.export.annotation.ManagedOperation; + +/** + * @author Artem Bilan + * @since 2.2 + */ + +public interface DelayHandlerManagement { + + @ManagedAttribute + int getDelayedMessageCount(); + + @ManagedOperation + void reschedulePersistedMessages(); +} diff --git a/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/config/GroovyControlBusFactoryBean.java b/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/config/GroovyControlBusFactoryBean.java index 8976dd7d56..7cd6c31637 100644 --- a/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/config/GroovyControlBusFactoryBean.java +++ b/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/config/GroovyControlBusFactoryBean.java @@ -61,7 +61,6 @@ public class GroovyControlBusFactoryBean extends AbstractSimpleMessageHandlerFac protected MessageHandler createHandler() { Binding binding = new ManagedBeansBinding(this.getBeanFactory()); GroovyCommandMessageProcessor processor = new GroovyCommandMessageProcessor(binding, new ScriptVariableGenerator() { - @Override public Map generateScriptVariables(Message message) { Map variables = new HashMap(); variables.put("headers", message.getHeaders()); diff --git a/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyControlBusDelayerManagementTest.groovy b/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyControlBusDelayerManagementTest.groovy new file mode 100644 index 0000000000..c685902dab --- /dev/null +++ b/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyControlBusDelayerManagementTest.groovy @@ -0,0 +1,11 @@ +import org.springframework.integration.test.util.TestUtils +import org.springframework.integration.handler.DelayHandler + +def delayHandler = TestUtils.getPropertyValue(testDelayer, 'handler', DelayHandler) + +def delayedMessageCount = delayHandler.delayedMessageCount +assert 2 == delayedMessageCount + +delayHandler.reschedulePersistedMessages() + +return true diff --git a/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyControlBusIntegrationTests-context.xml b/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyControlBusIntegrationTests-context.xml new file mode 100644 index 0000000000..c55ef0b749 --- /dev/null +++ b/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyControlBusIntegrationTests-context.xml @@ -0,0 +1,28 @@ + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyControlBusIntegrationTests.java b/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyControlBusIntegrationTests.java new file mode 100644 index 0000000000..15c2e2a00f --- /dev/null +++ b/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyControlBusIntegrationTests.java @@ -0,0 +1,78 @@ +/* + * Copyright 2002-2012 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.groovy; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import java.io.IOException; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.core.io.ClassPathResource; +import org.springframework.core.io.Resource; +import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.scripting.ScriptSource; +import org.springframework.scripting.support.ResourceScriptSource; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Artem Bilan + * @since 2.2 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class GroovyControlBusIntegrationTests { + + @Autowired + private MessageChannel controlBus; + + @Autowired + private PollableChannel controlBusOutput; + + @Autowired + private PollableChannel output; + + @Autowired + private MessageChannel delayerInput; + + @Autowired + ThreadPoolTaskScheduler scheduler; + + @Test + public void testDelayerManagement() throws IOException { + Message testMessage = MessageBuilder.withPayload("test").build(); + this.delayerInput.send(testMessage); + this.delayerInput.send(testMessage); + + this.scheduler.destroy(); + assertNull(this.output.receive(100)); + this.scheduler.afterPropertiesSet(); + + Resource scriptResource = new ClassPathResource("GroovyControlBusDelayerManagementTest.groovy", this.getClass()); + ScriptSource scriptSource = new ResourceScriptSource(scriptResource); + Message message = MessageBuilder.withPayload(scriptSource.getScriptAsString()).build(); + this.controlBus.send(message); + + assertNotNull(this.output.receive(100)); + assertNotNull(this.output.receive(100)); + } +}