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

MessageDispatcher now extends SubscribableSource. The 'addTarget' and 'removeTarget' methods have been replaced with 'subscribe' and 'unsubscribe' respectively.

Browse Source
This commit is contained in:
Mark Fisher
2008-08-14 11:53:24 +00:00
parent 234c87da1d
commit 3e15df93ed
8 changed files with 96 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

88
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/mail/SubscribableMailSource.java
View File

@@ -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));
}
}
}
}

4
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java
View File

@@ -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);
}

24
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/DirectChannel.java
View File

@@ -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);
}
}

8
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java
View File

@@ -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);
}

4
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java
View File

@@ -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() {

4
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PublishSubscribeChannel.java
View File

@@ -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

32
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java
View File

@@ -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());

24
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java
View File

@@ -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<MessageHandler>(handler1);
DefaultEndpoint<?> endpoint2 = new DefaultEndpoint<MessageHandler>(handler2);
DefaultEndpoint<?> endpoint3 = new DefaultEndpoint<MessageHandler>(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);