From 24a15520baa7677a18c8088a42798166f54e47ed Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 25 Apr 2008 15:18:39 +0000 Subject: [PATCH] PollingSourceEndpoint now provides initializeTask() for one-time only as well as refreshTask() to reconfigure proxies at runtime. --- .../endpoint/PollingSourceEndpoint.java | 108 +++++---- .../endpoint/PollingSourceEndpointTests.java | 218 ++++++++++++++++++ 2 files changed, 281 insertions(+), 45 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java index cbd35b9f7d..d17afe459c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java @@ -43,19 +43,21 @@ import org.springframework.util.Assert; */ public class PollingSourceEndpoint extends AbstractSourceEndpoint implements MessagingTask, InitializingBean { + private final Schedule schedule; + private final DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); private volatile PollingDispatcher dispatcher; + private volatile List dispatchAdviceChain; + private volatile MessagingTask task; private volatile List taskAdviceChain; - private volatile List dispatchAdviceChain; + private volatile boolean taskInitialized; - private volatile boolean proxiesInitialized; - - private final Object proxyInitializationMonitor = new Object(); + private final Object taskMonitor = new Object(); public PollingSourceEndpoint(PollableSource source, MessageChannel channel, PollingSchedule schedule) { @@ -63,48 +65,10 @@ public class PollingSourceEndpoint extends AbstractSourceEndpoint implements Mes Assert.notNull(schedule, "schedule must not be null"); this.dispatcher = new DefaultPollingDispatcher(source, this.dispatcherPolicy); this.dispatcher.subscribe(this.getChannel()); - this.task = new PollingDispatcherTask(this.dispatcher, schedule); + this.schedule = schedule; } - public void setTaskAdviceChain(List taskAdviceChain) { - this.taskAdviceChain = taskAdviceChain; - } - - public void setDispatchAdviceChain(List dispatchAdviceChain) { - this.dispatchAdviceChain = dispatchAdviceChain; - } - - public void afterPropertiesSet() { - this.initializeProxies(); - } - - public void initializeProxies() { - synchronized (this.proxyInitializationMonitor) { - if (this.proxiesInitialized) { - return; - } - if (this.dispatchAdviceChain != null && this.dispatchAdviceChain.size() > 0) { - ProxyFactory proxyFactory = new ProxyFactory(this.dispatcher); - proxyFactory.setInterfaces(new Class[] { PollingDispatcher.class }); - for (Advice advice : this.dispatchAdviceChain) { - proxyFactory.addAdvisor(new MethodNameAdvisor(advice, "dispatch")); - } - this.dispatcher = (PollingDispatcher) proxyFactory.getProxy(); - this.task = new PollingDispatcherTask(this.dispatcher, this.task.getSchedule()); - } - if (this.taskAdviceChain != null && this.taskAdviceChain.size() > 0) { - ProxyFactory proxyFactory = new ProxyFactory(this.task); - proxyFactory.setInterfaces(new Class[] { MessagingTask.class }); - for (Advice advice : this.taskAdviceChain) { - proxyFactory.addAdvisor(new MethodNameAdvisor(advice, "run")); - } - this.task = (MessagingTask) proxyFactory.getProxy(); - } - this.proxiesInitialized = true; - } - } - public void setMaxMessagesPerTask(int maxMessagesPerTask) { this.dispatcherPolicy.setMaxMessagesPerTask(maxMessagesPerTask); } @@ -113,12 +77,66 @@ public class PollingSourceEndpoint extends AbstractSourceEndpoint implements Mes this.dispatcher.setSendTimeout(sendTimeout); } + public void setTaskAdviceChain(List taskAdviceChain) { + this.taskAdviceChain = taskAdviceChain; + } + + public void setDispatchAdviceChain(List dispatchAdviceChain) { + this.dispatchAdviceChain = dispatchAdviceChain; + } + public Schedule getSchedule() { - return this.task.getSchedule(); + return this.schedule; + } + + public void afterPropertiesSet() { + this.initializeTask(); + } + + public void initializeTask() { + synchronized (this.taskMonitor) { + if (this.taskInitialized) { + return; + } + this.refreshTask(); + this.taskInitialized = true; + } + } + + public void refreshTask() { + synchronized (this.taskMonitor) { + PollingDispatcher dispatcherProxy = null; + if (this.dispatchAdviceChain != null && this.dispatchAdviceChain.size() > 0) { + ProxyFactory proxyFactory = new ProxyFactory(this.dispatcher); + proxyFactory.setInterfaces(new Class[] { PollingDispatcher.class }); + for (Advice advice : this.dispatchAdviceChain) { + proxyFactory.addAdvisor(new MethodNameAdvisor(advice, "dispatch")); + } + dispatcherProxy = (PollingDispatcher) proxyFactory.getProxy(); + } + this.task = new PollingDispatcherTask((dispatcherProxy != null) ? dispatcherProxy : this.dispatcher, this.schedule); + if (this.taskAdviceChain != null && this.taskAdviceChain.size() > 0) { + ProxyFactory proxyFactory = new ProxyFactory(this.task); + proxyFactory.setInterfaces(new Class[] { MessagingTask.class }); + for (Advice advice : this.taskAdviceChain) { + proxyFactory.addAdvisor(new MethodNameAdvisor(advice, "run")); + } + this.task = (MessagingTask) proxyFactory.getProxy(); + } + } + } + + private MessagingTask getTask() { + synchronized (this.taskMonitor) { + if (!this.taskInitialized) { + this.initializeTask(); + } + return this.task; + } } public void run() { - this.task.run(); + this.getTask().run(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java index efcafb9f2d..cea7d90800 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java @@ -210,6 +210,224 @@ public class PollingSourceEndpointTests { assertEquals("12abcabcabc3", buffer.toString()); } + @Test + public void testRefreshTaskAtRuntime() { + TestSource source = new TestSource("testing", 3); + QueueChannel channel = new QueueChannel(); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + List dispatchAdviceChain = new ArrayList(); + List taskAdviceChain = new ArrayList(); + final StringBuffer buffer = new StringBuffer(); + dispatchAdviceChain.add(new MethodBeforeAdvice() { + public void before(Method method, Object[] args, Object target) throws Throwable { + buffer.append("a"); + } + }); + dispatchAdviceChain.add(new MethodInterceptor() { + public Object invoke(MethodInvocation invocation) throws Throwable { + buffer.append("b"); + Object retval = invocation.proceed(); + buffer.append("c"); + return retval; + } + }); + taskAdviceChain.add(new MethodInterceptor() { + public Object invoke(MethodInvocation invocation) throws Throwable { + buffer.append(1); + Object retval = invocation.proceed(); + buffer.append(3); + return retval; + } + }); + taskAdviceChain.add(new MethodBeforeAdvice() { + public void before(Method method, Object[] args, Object target) throws Throwable { + buffer.append(2); + } + }); + endpoint.setDispatchAdviceChain(dispatchAdviceChain); + endpoint.afterPropertiesSet(); + endpoint.setMaxMessagesPerTask(5); + endpoint.run(); + assertEquals("abcabcabc", buffer.toString()); + buffer.delete(0, buffer.length()); + source.resetCounter(); + endpoint.setTaskAdviceChain(taskAdviceChain); + endpoint.refreshTask(); + endpoint.run(); + assertEquals("12abcabcabc3", buffer.toString()); + buffer.delete(0, buffer.length()); + source.resetCounter(); + endpoint.setDispatchAdviceChain(null); + endpoint.refreshTask(); + endpoint.run(); + assertEquals("123", buffer.toString()); + buffer.delete(0, buffer.length()); + source.resetCounter(); + endpoint.setTaskAdviceChain(null); + endpoint.setDispatchAdviceChain(dispatchAdviceChain); + endpoint.refreshTask(); + endpoint.run(); + assertEquals("abcabcabc", buffer.toString()); + } + + @Test + public void testInitializeTaskDoesNotRefreshWithDispatchAdviceOnly() { + TestSource source = new TestSource("testing", 3); + QueueChannel channel = new QueueChannel(); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + List dispatchAdviceChain = new ArrayList(); + List taskAdviceChain = new ArrayList(); + final StringBuffer buffer = new StringBuffer(); + dispatchAdviceChain.add(new MethodBeforeAdvice() { + public void before(Method method, Object[] args, Object target) throws Throwable { + buffer.append("a"); + } + }); + taskAdviceChain.add(new MethodBeforeAdvice() { + public void before(Method method, Object[] args, Object target) throws Throwable { + buffer.append(1); + } + }); + endpoint.setDispatchAdviceChain(dispatchAdviceChain); + endpoint.afterPropertiesSet(); + endpoint.setMaxMessagesPerTask(5); + endpoint.run(); + assertEquals("aaa", buffer.toString()); + buffer.delete(0, buffer.length()); + source.resetCounter(); + endpoint.setTaskAdviceChain(taskAdviceChain); + endpoint.initializeTask(); + endpoint.run(); + assertEquals("aaa", buffer.toString()); + buffer.delete(0, buffer.length()); + source.resetCounter(); + endpoint.setDispatchAdviceChain(null); + endpoint.initializeTask(); + endpoint.run(); + assertEquals("aaa", buffer.toString()); + } + + @Test + public void testInitializeTaskDoesNotRefreshWithTaskAdviceOnly() { + TestSource source = new TestSource("testing", 3); + QueueChannel channel = new QueueChannel(); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + List dispatchAdviceChain = new ArrayList(); + List taskAdviceChain = new ArrayList(); + final StringBuffer buffer = new StringBuffer(); + dispatchAdviceChain.add(new MethodBeforeAdvice() { + public void before(Method method, Object[] args, Object target) throws Throwable { + buffer.append("a"); + } + }); + taskAdviceChain.add(new MethodBeforeAdvice() { + public void before(Method method, Object[] args, Object target) throws Throwable { + buffer.append(1); + } + }); + endpoint.setTaskAdviceChain(taskAdviceChain); + endpoint.afterPropertiesSet(); + endpoint.setMaxMessagesPerTask(5); + endpoint.run(); + assertEquals("1", buffer.toString()); + buffer.delete(0, buffer.length()); + source.resetCounter(); + endpoint.setDispatchAdviceChain(dispatchAdviceChain); + endpoint.initializeTask(); + endpoint.run(); + assertEquals("1", buffer.toString()); + buffer.delete(0, buffer.length()); + source.resetCounter(); + endpoint.setTaskAdviceChain(null); + endpoint.initializeTask(); + endpoint.run(); + assertEquals("1", buffer.toString()); + } + + @Test + public void testInitializeTaskDoesNotRefreshWithTaskAndDispatchAdvice() { + TestSource source = new TestSource("testing", 3); + QueueChannel channel = new QueueChannel(); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + List dispatchAdviceChain = new ArrayList(); + List taskAdviceChain = new ArrayList(); + final StringBuffer buffer = new StringBuffer(); + dispatchAdviceChain.add(new MethodBeforeAdvice() { + public void before(Method method, Object[] args, Object target) throws Throwable { + buffer.append("a"); + } + }); + taskAdviceChain.add(new MethodBeforeAdvice() { + public void before(Method method, Object[] args, Object target) throws Throwable { + buffer.append(1); + } + }); + endpoint.setTaskAdviceChain(taskAdviceChain); + endpoint.setDispatchAdviceChain(dispatchAdviceChain); + endpoint.afterPropertiesSet(); + endpoint.setMaxMessagesPerTask(5); + endpoint.run(); + assertEquals("1aaa", buffer.toString()); + buffer.delete(0, buffer.length()); + source.resetCounter(); + endpoint.setDispatchAdviceChain(null); + endpoint.initializeTask(); + endpoint.run(); + assertEquals("1aaa", buffer.toString()); + buffer.delete(0, buffer.length()); + source.resetCounter(); + endpoint.setTaskAdviceChain(null); + endpoint.initializeTask(); + endpoint.run(); + assertEquals("1aaa", buffer.toString()); + } + + @Test + public void testInitializeTaskDoesNotRefreshWithNoAdvice() { + TestSource source = new TestSource("testing", 3); + QueueChannel channel = new QueueChannel(); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + List dispatchAdviceChain = new ArrayList(); + List taskAdviceChain = new ArrayList(); + final StringBuffer buffer = new StringBuffer(); + dispatchAdviceChain.add(new MethodBeforeAdvice() { + public void before(Method method, Object[] args, Object target) throws Throwable { + buffer.append("a"); + } + }); + taskAdviceChain.add(new MethodBeforeAdvice() { + public void before(Method method, Object[] args, Object target) throws Throwable { + buffer.append(1); + } + }); + endpoint.afterPropertiesSet(); + endpoint.setMaxMessagesPerTask(5); + endpoint.run(); + assertEquals("", buffer.toString()); + buffer.delete(0, buffer.length()); + source.resetCounter(); + endpoint.setDispatchAdviceChain(dispatchAdviceChain); + endpoint.initializeTask(); + endpoint.run(); + assertEquals("", buffer.toString()); + buffer.delete(0, buffer.length()); + source.resetCounter(); + endpoint.setTaskAdviceChain(taskAdviceChain); + endpoint.initializeTask(); + endpoint.run(); + assertEquals("", buffer.toString()); + } + private static class TestSource implements PollableSource {