diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/SubscribableMailSource.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/SubscribableMailSource.java index ea396b689a..3cfd34dec5 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/SubscribableMailSource.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/SubscribableMailSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * Copyright 2002-2008 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. @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.adapter.mail; import javax.mail.Message; @@ -20,6 +21,7 @@ import javax.mail.internet.MimeMessage; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import org.springframework.beans.factory.DisposableBean; import org.springframework.context.Lifecycle; import org.springframework.core.task.TaskExecutor; @@ -27,17 +29,18 @@ import org.springframework.integration.adapter.mail.monitor.AsyncMonitoringStrat import org.springframework.integration.dispatcher.BroadcastingDispatcher; import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.SubscribableSource; +import org.springframework.util.Assert; /** - * Broadcasts all mail messages recovered to subscribed {@link MessageTarget} - * The given {@link FolderConnection} should be using an - * {@link AsyncMonitoringStrategy} to retrieve mail + * Broadcasts all mail messages recovered to subscribed {@link MessageTarget MessageTargets}. + * The given {@link FolderConnection} should be using an {@link AsyncMonitoringStrategy} to + * retrieve mail. * * @author Jonas Partner - * */ -public class SubscribableMailSource implements SubscribableSource, Lifecycle, - DisposableBean { +public class SubscribableMailSource implements SubscribableSource, Lifecycle, DisposableBean { + + private final Log logger = LogFactory.getLog(this.getClass()); private final BroadcastingDispatcher dispatcher = new BroadcastingDispatcher(); @@ -45,29 +48,29 @@ public class SubscribableMailSource implements SubscribableSource, Lifecycle, private final MonitorRunnable monitorRunnable; - private boolean monitorRunning = false; + private volatile boolean monitorRunning = false; - private final Log logger = LogFactory.getLog(getClass()); + private volatile MailMessageConverter converter = new DefaultMailMessageConverter(); - private MailMessageConverter converter = new DefaultMailMessageConverter(); - public SubscribableMailSource(FolderConnection folderConnection, - TaskExecutor taskExecutor) { + public SubscribableMailSource(FolderConnection folderConnection, TaskExecutor taskExecutor) { + Assert.notNull(folderConnection, "FolderConnection must not be null"); + Assert.notNull(taskExecutor, "TaskExecutor must not be null"); this.monitorRunnable = new MonitorRunnable(folderConnection); this.taskExecutor = taskExecutor; - } + public void setApplySequence(boolean applySequence) { this.dispatcher.setApplySequence(applySequence); } public boolean subscribe(MessageTarget target) { - return this.dispatcher.addTarget(target); + return this.dispatcher.subscribe(target); } public boolean unsubscribe(MessageTarget target) { - return this.dispatcher.removeTarget(target); + return this.dispatcher.unsubscribe(target); } public void setConverter(MailMessageConverter converter) { @@ -75,60 +78,77 @@ public class SubscribableMailSource implements SubscribableSource, Lifecycle, } public void destroy() throws Exception { - stop(); + this.stop(); } public void start() { - logger.info("Starting to monitor mailbox"); - startMonitor(); - logger.info("Started to monitor mailbox"); + if (logger.isInfoEnabled()) { + logger.info("Starting to monitor mailbox"); + } + this.startMonitor(); + if (logger.isInfoEnabled()) { + logger.info("Started to monitor mailbox"); + } } public void stop() { - logger.info("Stopping monitoring of mailbox"); - stopMonitor(); - logger.info("Stopped monitoring mailbox"); + if (logger.isInfoEnabled()) { + logger.info("Stopping monitoring of mailbox"); + } + this.stopMonitor(); + if (logger.isInfoEnabled()) { + logger.info("Stopped monitoring mailbox"); + } } public boolean isRunning() { - return monitorRunning; + return this.monitorRunning; } - protected synchronized void startMonitor() { - if (!monitorRunning) { - taskExecutor.execute(monitorRunnable); + protected void startMonitor() { + synchronized (this.monitorRunnable) { + if (!this.monitorRunning) { + this.taskExecutor.execute(this.monitorRunnable); + } + this.monitorRunning = true; } } - protected synchronized void stopMonitor() { - if (monitorRunning) { - monitorRunnable.interrupt(); + protected void stopMonitor() { + synchronized (this.monitorRunnable) { + if (this.monitorRunning) { + this.monitorRunnable.interrupt(); + } + this.monitorRunning = false; } } + private class MonitorRunnable implements Runnable { + private volatile Thread thread; private final FolderConnection folderConnection; - protected MonitorRunnable(FolderConnection folderConnection) { + + private MonitorRunnable(FolderConnection folderConnection) { this.folderConnection = folderConnection; } + public synchronized void interrupt() { - thread.interrupt(); + this.thread.interrupt(); } public void run() { - thread = Thread.currentThread(); + this.thread = Thread.currentThread(); while (!Thread.currentThread().isInterrupted()) { - Message[] messages = folderConnection.receive(); + Message[] messages = this.folderConnection.receive(); for (Message message : messages) { dispatcher.send(converter.create((MimeMessage) message)); } } - } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java index 5cbe34b224..0a8cadc388 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java @@ -47,11 +47,11 @@ public abstract class AbstractDispatcher implements MessageDispatcher { this.messageExchangeTemplate.setSendTimeout(timeout); } - public boolean addTarget(MessageTarget target) { + public boolean subscribe(MessageTarget target) { return this.targets.add(target); } - public boolean removeTarget(MessageTarget target) { + public boolean unsubscribe(MessageTarget target) { return this.targets.remove(target); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/DirectChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/DirectChannel.java index 36503bc2cd..8065c9beaf 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/DirectChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/DirectChannel.java @@ -16,16 +16,13 @@ package org.springframework.integration.dispatcher; -import java.util.concurrent.atomic.AtomicInteger; - import org.springframework.integration.channel.AbstractMessageChannel; -import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; import org.springframework.integration.message.SubscribableSource; import org.springframework.integration.message.MessageTarget; /** - * A channel that invokes the subscribed {@link MessageHandler handler(s)} in + * A channel that invokes the subscribed {@link MessageTarget target(s)} in * the sender's thread (returning after at most one handles the message). * * @author Dave Syer @@ -35,31 +32,18 @@ public class DirectChannel extends AbstractMessageChannel implements Subscribabl private final SimpleDispatcher dispatcher = new SimpleDispatcher(); - private final AtomicInteger handlerCount = new AtomicInteger(); - public boolean subscribe(MessageTarget target) { - boolean added = this.dispatcher.addTarget(target); - if (added) { - this.handlerCount.incrementAndGet(); - } - return added; + return this.dispatcher.subscribe(target); } public boolean unsubscribe(MessageTarget target) { - boolean removed = this.dispatcher.removeTarget(target); - if (removed) { - this.handlerCount.decrementAndGet(); - } - return removed; + return this.dispatcher.unsubscribe(target); } @Override protected boolean doSend(Message message, long timeout) { - if (message != null && this.handlerCount.get() > 0) { - return this.dispatcher.send(message); - } - return false; + return this.dispatcher.send(message); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java index cc288934ba..03f352d70d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java @@ -18,25 +18,25 @@ package org.springframework.integration.dispatcher; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.message.SubscribableSource; /** * Strategy interface for dispatching messages. * * @author Mark Fisher */ -public interface MessageDispatcher extends MessageTarget { +public interface MessageDispatcher extends MessageTarget, SubscribableSource { boolean send(Message message); /** * Specify the timeout for sending to a target (in milliseconds). * Note that this value will only be applicable for blocking targets. - * The default value is 0. */ void setTimeout(long timeout); - boolean addTarget(MessageTarget target); + boolean subscribe(MessageTarget target); - boolean removeTarget(MessageTarget target); + boolean unsubscribe(MessageTarget target); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java index fcff959c14..cf25dde1bb 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java @@ -110,11 +110,11 @@ public class PollingDispatcher implements SchedulableTask, SubscribableSource { } public boolean subscribe(MessageTarget target) { - return this.dispatcher.addTarget(target); + return this.dispatcher.subscribe(target); } public boolean unsubscribe(MessageTarget target) { - return this.dispatcher.removeTarget(target); + return this.dispatcher.unsubscribe(target); } public Schedule getSchedule() { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PublishSubscribeChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PublishSubscribeChannel.java index cba5b5c8ef..f9e340f896 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PublishSubscribeChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PublishSubscribeChannel.java @@ -49,11 +49,11 @@ public class PublishSubscribeChannel extends AbstractMessageChannel implements S } public boolean subscribe(MessageTarget target) { - return this.dispatcher.addTarget(target); + return this.dispatcher.subscribe(target); } public boolean unsubscribe(MessageTarget target) { - return this.dispatcher.removeTarget(target); + return this.dispatcher.unsubscribe(target); } @Override diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java index 20ed026685..24b05194a5 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java @@ -71,8 +71,8 @@ public class BroadcastingDispatcherTests { @Test public void publishSubcribe() throws Exception { dispatcher.setTaskExecutor(null); - dispatcher.addTarget(targetMock); - dispatcher.addTarget(targetMock); + dispatcher.subscribe(targetMock); + dispatcher.subscribe(targetMock); expect(targetMock.send(messageMock)).andReturn(true).times(2); replay(globalMocks); dispatcher.send(messageMock); @@ -82,9 +82,9 @@ public class BroadcastingDispatcherTests { @Test public void multipleTargetsWithExecutor() { // should the same target be allowed to be added twice? - dispatcher.addTarget(targetMock); - dispatcher.addTarget(targetMock); - dispatcher.addTarget(targetMock); + dispatcher.subscribe(targetMock); + dispatcher.subscribe(targetMock); + dispatcher.subscribe(targetMock); expect(targetMock.send(messageMock)).andReturn(true).times(3); replay(globalMocks); dispatcher.send(messageMock); @@ -94,9 +94,9 @@ public class BroadcastingDispatcherTests { @Test public void multipleTargetsPartialFailure() { reset(taskExecutorMock); - dispatcher.addTarget(targetMock); - dispatcher.addTarget(targetMock); - dispatcher.addTarget(targetMock); + dispatcher.subscribe(targetMock); + dispatcher.subscribe(targetMock); + dispatcher.subscribe(targetMock); partialFailingExecutorMock(true, false, true); expect(targetMock.send(messageMock)).andReturn(true).times(2); replay(globalMocks); @@ -107,9 +107,9 @@ public class BroadcastingDispatcherTests { @Test(timeout = 500) public void multipleTargetsPartialTimeout() throws Exception { reset(taskExecutorMock); - dispatcher.addTarget(targetMock); - dispatcher.addTarget(targetMock); - dispatcher.addTarget(targetMock); + dispatcher.subscribe(targetMock); + dispatcher.subscribe(targetMock); + dispatcher.subscribe(targetMock); dispatcher.setTimeout(50); // three threads invoking targets final CountDownLatch latch = new CountDownLatch(3); @@ -160,8 +160,8 @@ public class BroadcastingDispatcherTests { return true; } }; - dispatcher.addTarget(target); - dispatcher.addTarget(target); + dispatcher.subscribe(target); + dispatcher.subscribe(target); dispatcher.send(new StringMessage("test")); assertEquals(2, messages.size()); assertEquals(0, (int) messages.get(0).getHeaders().getSequenceNumber()); @@ -181,9 +181,9 @@ public class BroadcastingDispatcherTests { return true; } }; - dispatcher.addTarget(target); - dispatcher.addTarget(target); - dispatcher.addTarget(target); + dispatcher.subscribe(target); + dispatcher.subscribe(target); + dispatcher.subscribe(target); dispatcher.send(new StringMessage("test")); assertEquals(3, messages.size()); assertEquals(1, (int) messages.get(0).getHeaders().getSequenceNumber()); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java index ea4dc00bd4..2d2eb004db 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java @@ -44,7 +44,7 @@ public class SimpleDispatcherTests { public void testSingleMessage() throws InterruptedException { SimpleDispatcher dispatcher = new SimpleDispatcher(); final CountDownLatch latch = new CountDownLatch(1); - dispatcher.addTarget(createEndpoint(TestHandlers.countDownHandler(latch))); + dispatcher.subscribe(createEndpoint(TestHandlers.countDownHandler(latch))); dispatcher.send(new StringMessage("test")); latch.await(500, TimeUnit.MILLISECONDS); assertEquals(0, latch.getCount()); @@ -56,8 +56,8 @@ public class SimpleDispatcherTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); - dispatcher.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch))); - dispatcher.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch))); + dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch))); + dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch))); dispatcher.send(new StringMessage("test")); latch.await(500, TimeUnit.MILLISECONDS); assertEquals(0, latch.getCount()); @@ -80,9 +80,9 @@ public class SimpleDispatcherTests { endpoint1.setSelector(new TestMessageSelector(selectorCounter, false)); endpoint2.setSelector(new TestMessageSelector(selectorCounter, false)); endpoint3.setSelector(new TestMessageSelector(selectorCounter, true)); - dispatcher.addTarget(endpoint1); - dispatcher.addTarget(endpoint2); - dispatcher.addTarget(endpoint3); + dispatcher.subscribe(endpoint1); + dispatcher.subscribe(endpoint2); + dispatcher.subscribe(endpoint3); dispatcher.send(new StringMessage("test")); assertEquals(0, latch.getCount()); assertEquals("selectors should have been invoked one time each", 3, selectorCounter.get()); @@ -107,9 +107,9 @@ public class SimpleDispatcherTests { endpoint1.setSelector(new TestMessageSelector(selectorCounter, false)); endpoint2.setSelector(new TestMessageSelector(selectorCounter, false)); endpoint3.setSelector(new TestMessageSelector(selectorCounter, false)); - dispatcher.addTarget(endpoint1); - dispatcher.addTarget(endpoint2); - dispatcher.addTarget(endpoint3); + dispatcher.subscribe(endpoint1); + dispatcher.subscribe(endpoint2); + dispatcher.subscribe(endpoint3); boolean exceptionThrown = false; try { dispatcher.send(new StringMessage("test")); @@ -136,9 +136,9 @@ public class SimpleDispatcherTests { DefaultEndpoint endpoint1 = new DefaultEndpoint(handler1); DefaultEndpoint endpoint2 = new DefaultEndpoint(handler2); DefaultEndpoint endpoint3 = new DefaultEndpoint(handler3); - dispatcher.addTarget(endpoint1); - dispatcher.addTarget(endpoint2); - dispatcher.addTarget(endpoint3); + dispatcher.subscribe(endpoint1); + dispatcher.subscribe(endpoint2); + dispatcher.subscribe(endpoint3); dispatcher.send(new StringMessage("test")); assertEquals("handlers should have been invoked 9 times in total", 9, handlerCounter.get()); assertFalse("first handler should not have handled the message", handler1.handledMessage);