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

PollingSourceEndpoint now provides initializeTask() for one-time only as well as refreshTask() to reconfigure proxies at runtime.

Browse Source
This commit is contained in:
Mark Fisher
2008-04-25 15:18:39 +00:00
parent 2147ac43ed
commit 24a15520ba
2 changed files with 281 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

108
spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java
View File

@@ -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<Advice> dispatchAdviceChain;
private volatile MessagingTask task;
private volatile List<Advice> taskAdviceChain;
private volatile List<Advice> 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<Advice> taskAdviceChain) {
this.taskAdviceChain = taskAdviceChain;
}
public void setDispatchAdviceChain(List<Advice> 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<Advice> taskAdviceChain) {
this.taskAdviceChain = taskAdviceChain;
}
public void setDispatchAdviceChain(List<Advice> 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();
}

218
spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java
View File

@@ -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<Advice> dispatchAdviceChain = new ArrayList<Advice>();
List<Advice> taskAdviceChain = new ArrayList<Advice>();
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<Advice> dispatchAdviceChain = new ArrayList<Advice>();
List<Advice> taskAdviceChain = new ArrayList<Advice>();
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<Advice> dispatchAdviceChain = new ArrayList<Advice>();
List<Advice> taskAdviceChain = new ArrayList<Advice>();
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<Advice> dispatchAdviceChain = new ArrayList<Advice>();
List<Advice> taskAdviceChain = new ArrayList<Advice>();
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<Advice> dispatchAdviceChain = new ArrayList<Advice>();
List<Advice> taskAdviceChain = new ArrayList<Advice>();
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<String> {