From 0aa00b969b96f7526b73267a3a7354829022d5d0 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 4 Jul 2008 18:48:33 +0000 Subject: [PATCH] Major refactoring of endpoint subscription, scheduling, and activation. This includes significant changes to the dispatcher implementations and the addition of endpoint "triggers" to drive polling and dispatching of messages. Also introduces an explicit PublishSubscribeChannel implementation and thereby removes support for the "publish-subscribe" attribute from channel elements in the namespace. --- .../adapter/stream/ByteStreamSourceTests.java | 12 +- .../adapter/stream/ByteStreamTargetTests.java | 100 ++-- .../stream/CharacterStreamSourceTests.java | 4 +- .../stream/CharacterStreamTargetTests.java | 82 ++- .../integration/bus/MessageBus.java | 186 +++---- .../integration/bus/SubscriptionManager.java | 172 ------- .../channel/PublishSubscribeChannel.java | 78 +++ .../config/PublishSubscribeChannelParser.java | 46 ++ .../config/IntegrationNamespaceHandler.java | 2 + .../config/spring-integration-core-1.0.xsd | 16 +- .../dispatcher/AbstractDispatcher.java | 81 +++ .../dispatcher/BroadcastingDispatcher.java | 50 ++ .../integration/dispatcher/DirectChannel.java | 4 +- .../dispatcher/MessageDispatcher.java | 11 +- .../dispatcher/PollingDispatcher.java | 119 +++++ .../dispatcher/PollingDispatcherTask.java | 94 ---- .../dispatcher/SimpleDispatcher.java | 31 +- .../endpoint/AbstractEndpoint.java | 73 ++- .../integration/endpoint/EndpointTrigger.java | 59 +++ .../integration/endpoint/HandlerEndpoint.java | 24 +- .../endpoint/MessageConsumingEndpoint.java | 28 -- .../integration/endpoint/MessageEndpoint.java | 19 +- .../endpoint/MessageProducingEndpoint.java | 30 -- .../integration/endpoint/SourceEndpoint.java | 13 +- .../integration/endpoint/TargetEndpoint.java | 44 +- .../integration/bus/MessageBusTests.java | 4 +- .../bus/SubscriptionManagerTests.java | 470 ------------------ .../channel/config/ChannelParserTests.java | 98 +--- .../channel/config/channelParserTests.xml | 11 +- ...MessageEndpointBeanPostProcessorTests.java | 9 +- .../messageEndpointBeanPostProcessorTests.xml | 2 +- .../dispatcher/SimpleDispatcherTests.java | 10 +- .../endpoint/SourceEndpointTests.java | 4 +- .../TransactionInterceptorTests.java | 18 +- 34 files changed, 782 insertions(+), 1222 deletions(-) delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/bus/SubscriptionManager.java create mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java create mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/channel/config/PublishSubscribeChannelParser.java create mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java create mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java create mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcherTask.java create mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointTrigger.java delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageConsumingEndpoint.java delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageProducingEndpoint.java delete mode 100644 org.springframework.integration/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceTests.java index e6e33a2dae..2a74ff2f57 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceTests.java @@ -44,7 +44,7 @@ public class ByteStreamSourceTests { ByteStreamSource source = new ByteStreamSource(stream); SourceEndpoint endpoint = new SourceEndpoint(source, channel); endpoint.afterPropertiesSet(); - endpoint.invoke(new CommandMessage(new PollCommand())); + endpoint.send(new CommandMessage(new PollCommand())); Message message1 = channel.receive(500); byte[] payload = (byte[]) message1.getPayload(); assertEquals(3, payload.length); @@ -53,7 +53,7 @@ public class ByteStreamSourceTests { assertEquals(3, payload[2]); Message message2 = channel.receive(0); assertNull(message2); - endpoint.invoke(new CommandMessage(new PollCommand())); + endpoint.send(new CommandMessage(new PollCommand())); Message message3 = channel.receive(0); assertNull(message3); } @@ -69,12 +69,12 @@ public class ByteStreamSourceTests { schedule.setInitialDelay(10000); SourceEndpoint endpoint = new SourceEndpoint(source, channel); endpoint.afterPropertiesSet(); - endpoint.invoke(new CommandMessage(new PollCommand())); + endpoint.send(new CommandMessage(new PollCommand())); Message message1 = channel.receive(0); assertEquals(4, ((byte[]) message1.getPayload()).length); Message message2 = channel.receive(0); assertNull(message2); - endpoint.invoke(new CommandMessage(new PollCommand())); + endpoint.send(new CommandMessage(new PollCommand())); Message message3 = channel.receive(0); assertEquals(2, ((byte[]) message3.getPayload()).length); } @@ -91,12 +91,12 @@ public class ByteStreamSourceTests { schedule.setInitialDelay(10000); SourceEndpoint endpoint = new SourceEndpoint(source, channel); endpoint.afterPropertiesSet(); - endpoint.invoke(new CommandMessage(new PollCommand())); + endpoint.send(new CommandMessage(new PollCommand())); Message message1 = channel.receive(0); assertEquals(4, ((byte[]) message1.getPayload()).length); Message message2 = channel.receive(0); assertNull(message2); - endpoint.invoke(new CommandMessage(new PollCommand())); + endpoint.send(new CommandMessage(new PollCommand())); Message message3 = channel.receive(0); assertEquals(4, ((byte[]) message3.getPayload()).length); assertEquals(0, ((byte[]) message3.getPayload())[3]); diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetTests.java index 1e55ccd6ae..0323ffefab 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetTests.java @@ -21,19 +21,34 @@ import static org.junit.Assert.assertEquals; import java.io.ByteArrayOutputStream; import java.io.IOException; +import org.junit.Before; import org.junit.Test; -import org.springframework.integration.channel.DispatcherPolicy; +import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.dispatcher.PollingDispatcherTask; +import org.springframework.integration.dispatcher.BroadcastingDispatcher; +import org.springframework.integration.dispatcher.PollingDispatcher; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; +import org.springframework.integration.scheduling.PollingSchedule; /** * @author Mark Fisher */ public class ByteStreamTargetTests { + private MessageChannel channel; + + private PollingDispatcher dispatcher; + + + @Before + public void initialize() { + this.channel = new QueueChannel(10); + this.dispatcher = new PollingDispatcher(channel, new BroadcastingDispatcher(), new PollingSchedule(0)); + } + + @Test public void testSingleByteArray() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); @@ -60,15 +75,12 @@ public class ByteStreamTargetTests { public void testMaxMessagesPerTaskSameAsMessageCount() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); - QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcherTask task = new PollingDispatcherTask(channel, null); - task.setMaxMessagesPerTask(3); - task.subscribe(target); + dispatcher.setMaxMessagesPerTask(3); + dispatcher.addTarget(target); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - task.run(); + dispatcher.run(); byte[] result = stream.toByteArray(); assertEquals(9, result.length); assertEquals(1, result[0]); @@ -79,15 +91,12 @@ public class ByteStreamTargetTests { public void testMaxMessagesPerTaskLessThanMessageCount() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); - QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcherTask task = new PollingDispatcherTask(channel, null); - task.setMaxMessagesPerTask(2); - task.subscribe(target); + dispatcher.setMaxMessagesPerTask(2); + dispatcher.addTarget(target); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - task.run(); + dispatcher.run(); byte[] result = stream.toByteArray(); assertEquals(6, result.length); assertEquals(1, result[0]); @@ -97,16 +106,13 @@ public class ByteStreamTargetTests { public void testMaxMessagesPerTaskExceedsMessageCount() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); - QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcherTask task = new PollingDispatcherTask(channel, null); - task.setMaxMessagesPerTask(5); - task.setReceiveTimeout(0); - task.subscribe(target); + dispatcher.setMaxMessagesPerTask(5); + dispatcher.setReceiveTimeout(0); + dispatcher.addTarget(target); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - task.run(); + dispatcher.run(); byte[] result = stream.toByteArray(); assertEquals(9, result.length); assertEquals(1, result[0]); @@ -116,20 +122,17 @@ public class ByteStreamTargetTests { public void testMaxMessagesLessThanMessageCountWithMultipleDispatches() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); - QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcherTask task = new PollingDispatcherTask(channel, null); - task.setMaxMessagesPerTask(2); - task.setReceiveTimeout(0); - task.subscribe(target); + dispatcher.setMaxMessagesPerTask(2); + dispatcher.setReceiveTimeout(0); + dispatcher.addTarget(target); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - task.run(); + dispatcher.run(); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); assertEquals(1, result1[0]); - task.run(); + dispatcher.run(); byte[] result2 = stream.toByteArray(); assertEquals(9, result2.length); assertEquals(1, result2[0]); @@ -140,20 +143,17 @@ public class ByteStreamTargetTests { public void testMaxMessagesExceedsMessageCountWithMultipleDispatches() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); - QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcherTask task = new PollingDispatcherTask(channel, null); - task.setMaxMessagesPerTask(5); - task.setReceiveTimeout(0); - task.subscribe(target); + dispatcher.setMaxMessagesPerTask(5); + dispatcher.setReceiveTimeout(0); + dispatcher.addTarget(target); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - task.run(); + dispatcher.run(); byte[] result1 = stream.toByteArray(); assertEquals(9, result1.length); assertEquals(1, result1[0]); - task.run(); + dispatcher.run(); byte[] result2 = stream.toByteArray(); assertEquals(9, result2.length); assertEquals(1, result2[0]); @@ -163,20 +163,17 @@ public class ByteStreamTargetTests { public void testStreamResetBetweenDispatches() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); - QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcherTask task = new PollingDispatcherTask(channel, null); - task.setMaxMessagesPerTask(2); - task.setReceiveTimeout(0); - task.subscribe(target); + dispatcher.setMaxMessagesPerTask(2); + dispatcher.setReceiveTimeout(0); + dispatcher.addTarget(target); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - task.run(); + dispatcher.run(); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); stream.reset(); - task.run(); + dispatcher.run(); byte[] result2 = stream.toByteArray(); assertEquals(3, result2.length); assertEquals(7, result2[0]); @@ -186,21 +183,18 @@ public class ByteStreamTargetTests { public void testStreamWriteBetweenDispatches() throws IOException { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); - QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcherTask task = new PollingDispatcherTask(channel, null); - task.setMaxMessagesPerTask(2); - task.setReceiveTimeout(0); - task.subscribe(target); + dispatcher.setMaxMessagesPerTask(2); + dispatcher.setReceiveTimeout(0); + dispatcher.addTarget(target); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - task.run(); + dispatcher.run(); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); stream.write(new byte[] {123}); stream.flush(); - task.run(); + dispatcher.run(); byte[] result2 = stream.toByteArray(); assertEquals(10, result2.length); assertEquals(1, result2[0]); diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceTests.java index 09ee9e60b5..1a000078fb 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceTests.java @@ -45,12 +45,12 @@ public class CharacterStreamSourceTests { schedule.setInitialDelay(10000); SourceEndpoint endpoint = new SourceEndpoint(source, channel); endpoint.afterPropertiesSet(); - endpoint.invoke(new CommandMessage(new PollCommand())); + endpoint.send(new CommandMessage(new PollCommand())); Message message1 = channel.receive(0); assertEquals("test", message1.getPayload()); Message message2 = channel.receive(0); assertNull(message2); - endpoint.invoke(new CommandMessage(new PollCommand())); + endpoint.send(new CommandMessage(new PollCommand())); Message message3 = channel.receive(0); assertNull(message3); } diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetTests.java index 40f93800ec..bdb65ef480 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetTests.java @@ -20,20 +20,34 @@ import static org.junit.Assert.assertEquals; import java.io.StringWriter; +import org.junit.Before; import org.junit.Test; -import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.dispatcher.PollingDispatcherTask; +import org.springframework.integration.dispatcher.BroadcastingDispatcher; +import org.springframework.integration.dispatcher.PollingDispatcher; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; +import org.springframework.integration.scheduling.PollingSchedule; /** * @author Mark Fisher */ public class CharacterStreamTargetTests { + private MessageChannel channel; + + private PollingDispatcher dispatcher; + + + @Before + public void initialize() { + this.channel = new QueueChannel(10); + this.dispatcher = new PollingDispatcher(channel, new BroadcastingDispatcher(), new PollingSchedule(0)); + } + + @Test public void testSingleString() { StringWriter writer = new StringWriter(); @@ -44,33 +58,29 @@ public class CharacterStreamTargetTests { @Test public void testTwoStringsAndNoNewLinesByDefault() { - MessageChannel channel = new QueueChannel(); StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); - PollingDispatcherTask task = new PollingDispatcherTask(channel, null); - task.subscribe(target); + dispatcher.addTarget(target); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - task.run(); + dispatcher.run(); assertEquals("foo", writer.toString()); - task.run(); + dispatcher.run(); assertEquals("foobar", writer.toString()); } @Test public void testTwoStringsWithNewLines() { - MessageChannel channel = new QueueChannel(); StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); target.setShouldAppendNewLine(true); - PollingDispatcherTask task = new PollingDispatcherTask(channel, null); - task.subscribe(target); + dispatcher.addTarget(target); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - task.run(); + dispatcher.run(); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine, writer.toString()); - task.run(); + dispatcher.run(); assertEquals("foo" + newLine + "bar" + newLine, writer.toString()); } @@ -78,14 +88,11 @@ public class CharacterStreamTargetTests { public void testMaxMessagesPerTaskSameAsMessageCount() { StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); - DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); - QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcherTask task = new PollingDispatcherTask(channel, null); - task.setMaxMessagesPerTask(2); - task.subscribe(target); + dispatcher.setMaxMessagesPerTask(2); + dispatcher.addTarget(target); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - task.run(); + dispatcher.run(); assertEquals("foobar", writer.toString()); } @@ -93,30 +100,25 @@ public class CharacterStreamTargetTests { public void testMaxMessagesPerTaskExceedsMessageCountWithAppendedNewLines() { StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); - DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); - QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcherTask task = new PollingDispatcherTask(channel, null); - task.setMaxMessagesPerTask(10); - task.setReceiveTimeout(0); - task.subscribe(target); + dispatcher.setMaxMessagesPerTask(10); + dispatcher.setReceiveTimeout(0); + dispatcher.addTarget(target); target.setShouldAppendNewLine(true); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - task.run(); + dispatcher.run(); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine + "bar" + newLine, writer.toString()); } @Test public void testSingleNonStringObject() { - MessageChannel channel = new QueueChannel(); StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); - PollingDispatcherTask task = new PollingDispatcherTask(channel, null); - task.subscribe(target); + dispatcher.addTarget(target); TestObject testObject = new TestObject("foo"); channel.send(new GenericMessage(testObject)); - task.run(); + dispatcher.run(); assertEquals("foo", writer.toString()); } @@ -124,17 +126,14 @@ public class CharacterStreamTargetTests { public void testTwoNonStringObjectWithOutNewLines() { StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); - DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); - QueueChannel channel = new QueueChannel(5, dispatcherPolicy); - PollingDispatcherTask task = new PollingDispatcherTask(channel, null); - task.setReceiveTimeout(0); - task.setMaxMessagesPerTask(2); - task.subscribe(target); + dispatcher.setReceiveTimeout(0); + dispatcher.setMaxMessagesPerTask(2); + dispatcher.addTarget(target); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); channel.send(new GenericMessage(testObject1), 0); channel.send(new GenericMessage(testObject2), 0); - task.run(); + dispatcher.run(); assertEquals("foobar", writer.toString()); } @@ -142,18 +141,15 @@ public class CharacterStreamTargetTests { public void testTwoNonStringObjectWithNewLines() { StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); - DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); - QueueChannel channel = new QueueChannel(5, dispatcherPolicy); target.setShouldAppendNewLine(true); - PollingDispatcherTask task = new PollingDispatcherTask(channel, null); - task.setReceiveTimeout(0); - task.setMaxMessagesPerTask(2); - task.subscribe(target); + dispatcher.setReceiveTimeout(0); + dispatcher.setMaxMessagesPerTask(2); + dispatcher.addTarget(target); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); channel.send(new GenericMessage(testObject1), 0); channel.send(new GenericMessage(testObject2), 0); - task.run(); + dispatcher.run(); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine + "bar" + newLine, writer.toString()); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java index 109bb77f62..2063caf2f8 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -16,12 +16,11 @@ package org.springframework.integration.bus; -import java.util.Collection; import java.util.List; import java.util.Map; import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.ScheduledThreadPoolExecutor; import org.apache.commons.logging.Log; @@ -29,7 +28,6 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.BeansException; import org.springframework.beans.factory.DisposableBean; -import org.springframework.beans.factory.InitializingBean; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.ApplicationEvent; @@ -45,7 +43,6 @@ import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.channel.factory.ChannelFactory; import org.springframework.integration.channel.factory.QueueChannelFactory; import org.springframework.integration.endpoint.DefaultEndpointRegistry; @@ -53,16 +50,13 @@ import org.springframework.integration.endpoint.EndpointRegistry; import org.springframework.integration.endpoint.HandlerEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.endpoint.MessagingGateway; -import org.springframework.integration.endpoint.MessageProducingEndpoint; -import org.springframework.integration.endpoint.SourceEndpoint; -import org.springframework.integration.endpoint.MessageConsumingEndpoint; import org.springframework.integration.endpoint.TargetEndpoint; +import org.springframework.integration.endpoint.EndpointTrigger; import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.message.CommandMessage; import org.springframework.integration.message.MessageTarget; -import org.springframework.integration.message.PollCommand; +import org.springframework.integration.message.Subscribable; import org.springframework.integration.scheduling.MessagePublishingErrorHandler; -import org.springframework.integration.scheduling.SchedulableTask; +import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.TaskScheduler; import org.springframework.integration.scheduling.Schedule; import org.springframework.integration.scheduling.SimpleTaskScheduler; @@ -91,12 +85,14 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, private final EndpointRegistry endpointRegistry = new DefaultEndpointRegistry(); - private final Map subscriptionManagers = new ConcurrentHashMap(); + private final Set endpointTriggers = new CopyOnWriteArraySet(); private final List lifecycleEndpoints = new CopyOnWriteArrayList(); private final MessageBusInterceptorsList interceptors = new MessageBusInterceptorsList(); + private volatile Schedule defaultPollerSchedule = new PollingSchedule(0); + private volatile TaskScheduler taskScheduler; private volatile boolean configureAsyncEventMulticaster = false; @@ -232,8 +228,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, this.initialize(); } channel.setName(name); - SubscriptionManager manager = new SubscriptionManager(channel, this.taskScheduler); - this.subscriptionManagers.put(channel, manager); this.channelRegistry.registerChannel(name, channel); if (logger.isInfoEnabled()) { logger.info("registered channel '" + name + "'"); @@ -241,14 +235,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, } public MessageChannel unregisterChannel(String name) { - MessageChannel removedChannel = this.channelRegistry.unregisterChannel(name); - if (removedChannel != null) { - SubscriptionManager manager = this.subscriptionManagers.remove(removedChannel); - if (manager != null && manager.isRunning()) { - manager.stop(); - } - } - return removedChannel; + return this.channelRegistry.unregisterChannel(name); } public void registerHandler(String name, MessageHandler handler, Subscription subscription) { @@ -276,12 +263,9 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, if (endpoint instanceof ChannelRegistryAware) { ((ChannelRegistryAware) endpoint).setChannelRegistry(this.channelRegistry); } - if (endpoint instanceof SourceEndpoint) { - this.registerSourceEndpoint(name, (SourceEndpoint) endpoint); - } this.endpointRegistry.registerEndpoint(name, endpoint); if (this.isRunning()) { - activateEndpoint(endpoint); + this.activateEndpoint(endpoint); } if (logger.isInfoEnabled()) { logger.info("registered endpoint '" + name + "'"); @@ -293,17 +277,8 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, if (endpoint == null) { return null; } - if (endpoint instanceof TargetEndpoint) { - Collection managers = this.subscriptionManagers.values(); - boolean removed = false; - for (SubscriptionManager manager : managers) { - removed = (removed || manager.removeTarget((TargetEndpoint) endpoint)); - } - if (removed) { - return endpoint; - } - } - return null; + this.deactivateEndpoint(endpoint); + return endpoint; } public MessageEndpoint lookupEndpoint(String endpointName) { @@ -325,84 +300,63 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, } private void activateEndpoint(MessageEndpoint endpoint) { - if (endpoint instanceof MessageProducingEndpoint) { - String channelName = ((MessageProducingEndpoint) endpoint).getOutputChannelName(); - if (channelName != null && this.lookupChannel(channelName) == null) { - if (!this.autoCreateChannels) { - throw new ConfigurationException("Unknown channel '" + channelName - + "' configured as output channel for endpoint '" + endpoint - + "'. Consider enabling the 'autoCreateChannels' option for the message bus."); - } - this.registerChannel(channelName, new QueueChannel()); - } + Assert.notNull(endpoint, "'endpoint' must not be null"); + if (endpoint.getOutputChannel() == null) { + this.lookupOrCreateChannel(endpoint.getOutputChannelName()); } - if (endpoint instanceof InitializingBean) { - try { - ((InitializingBean) endpoint).afterPropertiesSet(); - } - catch (Exception e) { - throw new ConfigurationException("failed to initialize endpoint", e); - } + try { + endpoint.afterPropertiesSet(); } - if (endpoint instanceof MessageConsumingEndpoint && endpoint instanceof MessageTarget) { - this.activateSubscriber((MessageConsumingEndpoint) endpoint); + catch (Exception e) { + throw new ConfigurationException("failed to initialize endpoint", e); } - } - - private void activateSubscriber(MessageConsumingEndpoint subscriber) { - Subscription subscription = subscriber.getSubscription(); - if (subscription == null) { - throw new ConfigurationException("Unable to register endpoint '" + subscriber - + "'. No subscription information is available."); - } - MessageChannel channel = subscription.getChannel(); - if (channel == null) { - String channelName = subscription.getChannelName(); - if (channelName == null) { - throw new ConfigurationException("endpoint '" + subscriber - + "' must provide either 'channel' or 'channelName' in its subscription metadata"); - } - channel = this.lookupChannel(channelName); + Schedule schedule = null; + Subscription subscription = endpoint.getSubscription(); + if (subscription != null) { + schedule = subscription.getSchedule(); + MessageChannel channel = subscription.getChannel(); if (channel == null) { - if (!this.autoCreateChannels) { - throw new ConfigurationException("Cannot activate subscription, unknown channel '" + channelName - + "'. Consider enabling the 'autoCreateChannels' option for the message bus."); + channel = this.lookupOrCreateChannel(subscription.getChannelName()); + } + if (channel != null && channel instanceof Subscribable) { + ((Subscribable) channel).subscribe(endpoint); + if (logger.isInfoEnabled()) { + logger.info("activated subscription to channel '" + + channel.getName() + "' for endpoint '" + endpoint + "'"); } - if (this.logger.isInfoEnabled()) { - logger.info("auto-creating channel '" + channelName + "'"); - } - channel = channelFactory.getChannel(null, null); - this.registerChannel(channelName, channel); + return; } } - this.activateSubscription(channel, (MessageTarget) subscriber, subscription.getSchedule()); - if (logger.isInfoEnabled()) { - logger.info("activated subscription to channel '" + channel.getName() - + "' for endpoint '" + subscriber + "' of type '" + subscriber.getClass() + "'"); + if (schedule == null) { + schedule = endpoint.getSchedule(); + } + EndpointTrigger trigger = endpoint.getTrigger(); + if (trigger == null) { + trigger = new EndpointTrigger(schedule != null ? schedule : this.defaultPollerSchedule); + } + trigger.addTarget(endpoint); + if (this.endpointTriggers.add(trigger)) { + this.taskScheduler.schedule(trigger); } } - private void registerSourceEndpoint(String name, final SourceEndpoint endpoint) { - if (!this.initialized) { - this.initialize(); + private MessageChannel lookupOrCreateChannel(String channelName) { + if (channelName == null) { + return null; } - final Schedule schedule = endpoint.getSchedule(); - if (schedule != null) { - this.taskScheduler.schedule(new SchedulableTask() { - public Schedule getSchedule() { - return schedule; - } - public void run() { - endpoint.invoke(new CommandMessage(new PollCommand())); - } - }); - } - if (endpoint instanceof Lifecycle) { - this.lifecycleEndpoints.add((Lifecycle) endpoint); - if (this.isRunning()) { - ((Lifecycle) endpoint).start(); + MessageChannel channel = this.lookupChannel(channelName); + if (channel == null) { + if (!this.autoCreateChannels) { + throw new ConfigurationException("Cannot activate endpoint, unknown channel '" + channelName + + "'. Consider enabling the 'autoCreateChannels' option for the message bus."); } + if (this.logger.isInfoEnabled()) { + logger.info("auto-creating channel '" + channelName + "'"); + } + channel = channelFactory.getChannel(null, null); + this.registerChannel(channelName, channel); } + return channel; } private void registerGateway(String name, MessagingGateway gateway) { @@ -417,18 +371,16 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, } } - private void activateSubscription(MessageChannel channel, MessageTarget target, Schedule schedule) { - SubscriptionManager manager = this.subscriptionManagers.get(channel); - if (manager == null) { - if (logger.isWarnEnabled()) { - logger.warn("no subscription manager available for channel '" + channel - + "', be sure to register the channel"); + public void deactivateEndpoint(MessageEndpoint endpoint) { + Assert.notNull(endpoint, "'endpoint' must not be null"); + for (EndpointTrigger trigger : this.endpointTriggers) { + boolean removed = trigger.removeTarget(endpoint); + if (removed && this.logger.isInfoEnabled()) { + logger.info("removed endpoint '" + endpoint + "' from dispatcher"); } - return; } - manager.addTarget(target, schedule); - if (this.isRunning() && !manager.isRunning()) { - manager.start(); + if (endpoint instanceof Lifecycle) { + ((Lifecycle) endpoint).stop(); } } @@ -451,12 +403,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, this.activateEndpoints(); this.taskScheduler.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel())); this.taskScheduler.start(); - for (SubscriptionManager manager : this.subscriptionManagers.values()) { - manager.start(); - if (logger.isInfoEnabled()) { - logger.info("started subscription manager '" + manager + "'"); - } - } for (Lifecycle endpoint : this.lifecycleEndpoints) { endpoint.start(); if (logger.isInfoEnabled()) { @@ -486,12 +432,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, logger.info("stopped endpoint '" + endpoint + "'"); } } - for (SubscriptionManager manager : this.subscriptionManagers.values()) { - manager.stop(); - if (logger.isInfoEnabled()) { - logger.info("stopped subscription manager '" + manager + "'"); - } - } } this.interceptors.postStop(); if (logger.isInfoEnabled()) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/SubscriptionManager.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/SubscriptionManager.java deleted file mode 100644 index 8c707c9d9b..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/SubscriptionManager.java +++ /dev/null @@ -1,172 +0,0 @@ -/* - * 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. - * 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.bus; - -import java.util.Collection; -import java.util.List; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.CopyOnWriteArrayList; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.context.Lifecycle; -import org.springframework.integration.ConfigurationException; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.dispatcher.DirectChannel; -import org.springframework.integration.dispatcher.PollingDispatcherTask; -import org.springframework.integration.message.MessageTarget; -import org.springframework.integration.scheduling.TaskScheduler; -import org.springframework.integration.scheduling.PollingSchedule; -import org.springframework.integration.scheduling.Schedule; -import org.springframework.util.Assert; - -/** - * Manages subscriptions for {@link MessageTarget Targets} to a {@link MessageChannel} - * including the creation, scheduling, and lifecycle management of dispatchers. - * - * @author Mark Fisher - */ -public class SubscriptionManager { - - private final Log logger = LogFactory.getLog(this.getClass()); - - private final MessageChannel channel; - - private final TaskScheduler scheduler; - - private volatile Schedule defaultSchedule = new PollingSchedule(5); - - private final ConcurrentMap dispatcherTasks = new ConcurrentHashMap(); - - private final List lifecycleTargets = new CopyOnWriteArrayList(); - - private volatile boolean running; - - private final Object lifecycleMonitor = new Object(); - - - public SubscriptionManager(MessageChannel channel, TaskScheduler scheduler) { - Assert.notNull(channel, "channel must not be null"); - Assert.notNull(scheduler, "scheduler must not be null"); - this.channel = channel; - this.scheduler = scheduler; - } - - - public void setDefaultSchedule(Schedule defaultSchedule) { - Assert.notNull(defaultSchedule, "'defaultSchedule' must not be null"); - this.defaultSchedule = defaultSchedule; - } - - public void addTarget(MessageTarget target) { - this.addTarget(target, null); - } - - public void addTarget(MessageTarget target, Schedule schedule) { - Assert.notNull(target, "'target' must not be null"); - if (schedule == null) { - schedule = this.defaultSchedule; - } - else if (this.channel instanceof DirectChannel) { - if (logger.isInfoEnabled()) { - logger.info("Subscribing to a DirectChannel. The provided schedule will be ignored."); - } - } - else if (this.channel.getDispatcherPolicy().isPublishSubscribe()) { - if (logger.isInfoEnabled()) { - logger.info("This dispatcher broadcasts messages for a publish-subscribe channel. " + - "Therefore all targets are scheduled with its 'defaultSchedule', " + - "and the provided schedule will be ignored."); - } - schedule = this.defaultSchedule; - } - if (target instanceof Lifecycle) { - this.lifecycleTargets.add((Lifecycle) target); - if (this.isRunning()) { - ((Lifecycle) target).start(); - } - } - if (this.channel instanceof DirectChannel) { - ((DirectChannel) this.channel).subscribe(target); - return; - } - PollingDispatcherTask dispatcherTask = this.dispatcherTasks.get(schedule); - if (dispatcherTask == null) { - dispatcherTask = this.dispatcherTasks.putIfAbsent(schedule, new PollingDispatcherTask(this.channel, schedule)); - } - this.dispatcherTasks.get(schedule).subscribe(target); - if (dispatcherTask == null && this.isRunning()) { - this.scheduleDispatcherTask(schedule); - } - } - - public boolean removeTarget(MessageTarget target) { - boolean removed = false; - Collection dispatcherTaskValues = this.dispatcherTasks.values(); - for (PollingDispatcherTask dispatcherTask : dispatcherTaskValues) { - removed = (removed || dispatcherTask.unsubscribe(target)); - } - return removed; - } - - public boolean isRunning() { - return this.running; - } - - public void start() { - synchronized (this.lifecycleMonitor) { - if (this.running) { - return; - } - if (this.scheduler == null) { - throw new ConfigurationException("scheduler is required"); - } - if (!this.scheduler.isRunning()) { - this.scheduler.start(); - } - for (Lifecycle target : lifecycleTargets) { - target.start(); - } - for (Schedule schedule : this.dispatcherTasks.keySet()) { - this.scheduleDispatcherTask(schedule); - } - this.running = true; - } - } - - private void scheduleDispatcherTask(Schedule schedule) { - PollingDispatcherTask dispatcherTask = this.dispatcherTasks.get(schedule); - if (dispatcherTask != null) { - this.scheduler.schedule(dispatcherTask); - } - } - - public void stop() { - synchronized (this.lifecycleMonitor) { - if (!this.running) { - return; - } - for (Lifecycle target : lifecycleTargets) { - target.stop(); - } - this.running = false; - } - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java new file mode 100644 index 0000000000..c2a82abad3 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java @@ -0,0 +1,78 @@ +/* + * 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. + * 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.channel; + +import java.util.List; + +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.dispatcher.BroadcastingDispatcher; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.message.Subscribable; +import org.springframework.integration.message.selector.MessageSelector; + +/** + * @author Mark Fisher + */ +public class PublishSubscribeChannel extends AbstractMessageChannel implements Subscribable { + + private final BroadcastingDispatcher dispatcher = new BroadcastingDispatcher(); + + + public PublishSubscribeChannel() { + super(new DispatcherPolicy(true)); + } + + /** + * Create a PublishSubscribeChannel that will use a {@link TaskExecutor} + * to publish its Messages. + */ + public PublishSubscribeChannel(TaskExecutor taskExecutor) { + this(); + if (taskExecutor != null) { + this.dispatcher.setTaskExecutor(taskExecutor); + } + } + + + public boolean subscribe(MessageTarget target) { + return this.dispatcher.addTarget(target); + } + + public boolean unsubscribe(MessageTarget target) { + return this.dispatcher.removeTarget(target); + } + + @Override + protected boolean doSend(Message message, long timeout) { + return this.dispatcher.send(message); + } + + @Override + protected Message doReceive(long timeout) { + return null; + } + + public List> clear() { + return null; + } + + public List> purge(MessageSelector selector) { + return null; + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/config/PublishSubscribeChannelParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/config/PublishSubscribeChannelParser.java new file mode 100644 index 0000000000..1dddfbe49b --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/config/PublishSubscribeChannelParser.java @@ -0,0 +1,46 @@ +/* + * 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. + * 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.channel.config; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.integration.channel.DispatcherPolicy; +import org.springframework.integration.channel.PublishSubscribeChannel; +import org.springframework.util.StringUtils; + +/** + * Parser for the <publish-subscribe-channel> element. + * + * @author Mark Fisher + */ +public class PublishSubscribeChannelParser extends AbstractChannelParser { + + @Override + protected Class getBeanClass(Element element) { + return PublishSubscribeChannel.class; + } + + @Override + protected void configureConstructorArgs(BeanDefinitionBuilder builder, Element element, DispatcherPolicy dispatcherPolicy) { + String taskExecutorRef = element.getAttribute("task-executor"); + if (StringUtils.hasText(taskExecutorRef)) { + builder.addConstructorArgReference(taskExecutorRef); + } + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java index 793bd25b76..cc15097c5b 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java @@ -29,6 +29,7 @@ import org.springframework.beans.factory.xml.BeanDefinitionParser; import org.springframework.beans.factory.xml.NamespaceHandlerSupport; import org.springframework.core.io.support.PropertiesLoaderUtils; import org.springframework.integration.channel.config.PriorityChannelParser; +import org.springframework.integration.channel.config.PublishSubscribeChannelParser; import org.springframework.integration.channel.config.QueueChannelParser; import org.springframework.integration.channel.config.RendezvousChannelParser; import org.springframework.integration.channel.config.ThreadLocalChannelParser; @@ -58,6 +59,7 @@ public class IntegrationNamespaceHandler extends NamespaceHandlerSupport { registerBeanDefinitionParser("annotation-driven", new AnnotationDrivenParser()); registerBeanDefinitionParser("channel", new DefaultChannelParser()); registerBeanDefinitionParser("queue-channel", new QueueChannelParser()); + registerBeanDefinitionParser("publish-subscribe-channel", new PublishSubscribeChannelParser()); registerBeanDefinitionParser("direct-channel", new DirectChannelParser()); registerBeanDefinitionParser("priority-channel", new PriorityChannelParser()); registerBeanDefinitionParser("rendezvous-channel", new RendezvousChannelParser()); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd index 8247de852d..a7ce65fedd 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd @@ -63,6 +63,21 @@ + + + + + Defines a publish-subscribe channel that broadcasts to its targets. + + + + + + + + + + @@ -121,7 +136,6 @@ - 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 new file mode 100644 index 0000000000..1ad36e7fce --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java @@ -0,0 +1,81 @@ +/* + * 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. + * 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.dispatcher; + +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.message.BlockingTarget; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageTarget; + +/** + * Base class for {@link MessageDispatcher} implementations. + * + * @author Mark Fisher + */ +public abstract class AbstractDispatcher implements MessageDispatcher { + + protected final Log logger = LogFactory.getLog(this.getClass()); + + protected final List targets = new CopyOnWriteArrayList(); + + private volatile long timeout = 0; + + private volatile TaskExecutor taskExecutor; + + + public void setTimeout(long timeout) { + this.timeout = timeout; + } + + public boolean addTarget(MessageTarget target) { + return this.targets.add(target); + } + + public boolean removeTarget(MessageTarget target) { + return this.targets.remove(target); + } + + /** + * Specify a {@link TaskExecutor} for invoking the target endpoints. + * If none is provided, the invocation will occur in the thread + * that runs this polling dispatcher. + */ + public void setTaskExecutor(TaskExecutor taskExecutor) { + this.taskExecutor = taskExecutor; + } + + protected TaskExecutor getTaskExecutor() { + return this.taskExecutor; + } + + protected final boolean sendMessageToTarget(Message message, MessageTarget target) { + boolean sent = (target instanceof BlockingTarget && this.timeout >= 0) + ? ((BlockingTarget) target).send(message, this.timeout) + : target.send(message); + if (!sent && logger.isDebugEnabled()) { + logger.debug("dispatcher failed to send message to target '" + target + "'"); + } + return sent; + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java new file mode 100644 index 0000000000..f770badf56 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java @@ -0,0 +1,50 @@ +/* + * 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. + * 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.dispatcher; + +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageTarget; + +/** + * A broadcasting dispatcher implementation. It makes a best effort to + * send the message to each of its targets. If it fails to send to any + * one target, it will log a warn-level message but continue to send + * to the other targets. + * + * @author Mark Fisher + */ +public class BroadcastingDispatcher extends AbstractDispatcher { + + public boolean send(final Message message) { + for (final MessageTarget target : this.targets) { + TaskExecutor executor = this.getTaskExecutor(); + if (executor != null) { + executor.execute(new Runnable() { + public void run() { + sendMessageToTarget(message, target); + } + }); + } + else { + this.sendMessageToTarget(message, target); + } + } + return true; + } + +} 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 8f09a993b4..1211a4fd1a 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 @@ -59,7 +59,7 @@ public class DirectChannel extends AbstractMessageChannel implements Subscribabl public boolean subscribe(MessageTarget target) { - boolean added = this.dispatcher.subscribe(target); + boolean added = this.dispatcher.addTarget(target); if (added) { this.handlerCount.incrementAndGet(); } @@ -67,7 +67,7 @@ public class DirectChannel extends AbstractMessageChannel implements Subscribabl } public boolean unsubscribe(MessageTarget target) { - boolean removed = this.dispatcher.unsubscribe(target); + boolean removed = this.dispatcher.removeTarget(target); if (removed) { this.handlerCount.decrementAndGet(); } 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 5adeb2e3fe..cc288934ba 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 @@ -28,6 +28,15 @@ public interface MessageDispatcher extends MessageTarget { boolean send(Message message); - void setSendTimeout(long timeout); + /** + * 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 removeTarget(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 new file mode 100644 index 0000000000..21da91044d --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java @@ -0,0 +1,119 @@ +/* + * 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. + * 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.dispatcher; + +import org.springframework.integration.message.BlockingSource; +import org.springframework.integration.message.BlockingTarget; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageSource; +import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.scheduling.SchedulableTask; +import org.springframework.integration.scheduling.Schedule; +import org.springframework.util.Assert; + +/** + * @author Mark Fisher + */ +public class PollingDispatcher implements SchedulableTask { + + private final MessageSource source; + + private final MessageDispatcher dispatcher; + + private final Schedule schedule; + + private volatile long receiveTimeout = 5000; + + private volatile int maxMessagesPerTask = 1; + + + /** + * Create a PollingDispatcher for the provided {@link MessageSource}. + * It can be scheduled according to the specified {@link Schedule}. + */ + public PollingDispatcher(MessageSource source, MessageDispatcher dispatcher, Schedule schedule) { + Assert.notNull(source, "source must not be null"); + Assert.notNull(dispatcher, "dispatcher must not be null"); + this.source = source; + this.dispatcher = dispatcher; + this.schedule = schedule; + } + + + /** + * Specify the timeout to use when receiving from the source (in milliseconds). + * Note that this value will only be applicable if the source is an instance + * of {@link BlockingSource}. + *

+ * A negative value indicates that receive calls should block indefinitely. + * The default value is 5000 (5 seconds). + */ + public void setReceiveTimeout(long receiveTimeout) { + this.receiveTimeout = receiveTimeout; + } + + /** + * Specify the timeout to use when sending to a target (in milliseconds). + * Note that this value will only be applicable if the target is an instance + * of {@link BlockingTarget}. The default value is 0. + */ + public void setSendTimeout(long sendTimeout) { + this.dispatcher.setTimeout(sendTimeout); + } + + /** + * Set the maximum number of messages to receive for each poll. + * A non-positive value indicates that polling should repeat as long + * as non-null messages are being received and successfully sent. + */ + public void setMaxMessagesPerTask(int maxMessagesPerTask) { + this.maxMessagesPerTask = maxMessagesPerTask; + } + + public boolean addTarget(MessageTarget target) { + return this.dispatcher.addTarget(target); + } + + public boolean removeTarget(MessageTarget target) { + return this.dispatcher.removeTarget(target); + } + + public Schedule getSchedule() { + return this.schedule; + } + + public void run() { + int count = 0; + while (this.maxMessagesPerTask <= 0 || count < this.maxMessagesPerTask) { + if (!this.dispatch()) { + return; + } + count++; + } + } + + private boolean dispatch() { + final Message message = (this.source instanceof BlockingSource && this.receiveTimeout >= 0) + ? ((BlockingSource) this.source).receive(this.receiveTimeout) + : this.source.receive(); + if (message == null) { + return false; + } + return this.dispatcher.send(message); + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcherTask.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcherTask.java deleted file mode 100644 index 9f439cb617..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcherTask.java +++ /dev/null @@ -1,94 +0,0 @@ -/* - * 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. - * 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.dispatcher; - -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.Subscribable; -import org.springframework.integration.message.MessageTarget; -import org.springframework.integration.scheduling.SchedulableTask; -import org.springframework.integration.scheduling.Schedule; -import org.springframework.util.Assert; - -/** - * A {@link SchedulableTask} that combines polling and dispatching. - * - * @author Mark Fisher - */ -public class PollingDispatcherTask implements SchedulableTask, Subscribable { - - private final MessageChannel channel; - - private final Schedule schedule; - - private final SimpleDispatcher dispatcher; - - private volatile long receiveTimeout = 5000; - - private volatile int maxMessagesPerTask = 1; - - - public PollingDispatcherTask(MessageChannel channel, Schedule schedule) { - Assert.notNull(channel, "channel must not be null"); - this.channel = channel; - this.schedule = schedule; - this.dispatcher = new SimpleDispatcher(this.channel.getDispatcherPolicy()); - } - - - /** - * Set the maximum amount of time in milliseconds to wait for a message to be available. - * A negative value indicates that receive calls should block indefinitely. - */ - public void setReceiveTimeout(long receiveTimeout) { - this.receiveTimeout = receiveTimeout; - } - - /** - * Set the maximum number of messages for each retrieval attempt. - */ - public void setMaxMessagesPerTask(int maxMessagesPerTask) { - Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagePerTask' must be at least 1"); - this.maxMessagesPerTask = maxMessagesPerTask; - } - - public boolean subscribe(MessageTarget target) { - return this.dispatcher.subscribe(target); - } - - public boolean unsubscribe(MessageTarget target) { - return this.dispatcher.unsubscribe(target); - } - - public Schedule getSchedule() { - return this.schedule; - } - - public void run() { - int count = 0; - while (count < this.maxMessagesPerTask) { - Message message = (this.receiveTimeout < 0) ? - this.channel.receive() : this.channel.receive(this.receiveTimeout); - if (message == null) { - return; - } - this.dispatcher.send(message); - count++; - } - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java index 86e8fb051e..927426d290 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java @@ -19,15 +19,10 @@ package org.springframework.integration.dispatcher; import java.util.ArrayList; import java.util.Iterator; import java.util.List; -import java.util.concurrent.CopyOnWriteArrayList; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.handler.MessageHandlerNotRunningException; import org.springframework.integration.handler.MessageHandlerRejectedExecutionException; -import org.springframework.integration.message.BlockingTarget; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.MessageTarget; @@ -37,34 +32,15 @@ import org.springframework.integration.message.MessageTarget; * * @author Mark Fisher */ -public class SimpleDispatcher implements MessageDispatcher { - - protected final Log logger = LogFactory.getLog(this.getClass()); - - private final List targets = new CopyOnWriteArrayList(); +public class SimpleDispatcher extends AbstractDispatcher { protected final DispatcherPolicy dispatcherPolicy; - private volatile long sendTimeout; - public SimpleDispatcher(DispatcherPolicy dispatcherPolicy) { this.dispatcherPolicy = (dispatcherPolicy != null) ? dispatcherPolicy : new DispatcherPolicy(); } - - public void setSendTimeout(long sendTimeout) { - this.sendTimeout = sendTimeout; - } - - public boolean subscribe(MessageTarget target) { - return this.targets.add(target); - } - - public boolean unsubscribe(MessageTarget target) { - return this.targets.remove(target); - } - public boolean send(Message message) { int attempts = 0; List targetList = new ArrayList(this.targets); @@ -94,8 +70,7 @@ public class SimpleDispatcher implements MessageDispatcher { while (iter.hasNext()) { MessageTarget target = iter.next(); try { - boolean sent = (target instanceof BlockingTarget && this.sendTimeout >= 0) ? - ((BlockingTarget) target).send(message, this.sendTimeout) : target.send(message); + boolean sent = this.sendMessageToTarget(message, target); if (!this.dispatcherPolicy.isPublishSubscribe() && sent) { return true; } @@ -112,7 +87,7 @@ public class SimpleDispatcher implements MessageDispatcher { catch (MessageHandlerRejectedExecutionException e) { rejected = true; if (logger.isDebugEnabled()) { - logger.debug("target is busy, continuing with other targets if available", e); + logger.debug("target '" + target + "' is busy, continuing with other targets if available", e); } } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java index bfe450ce14..e2338de60d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java @@ -25,19 +25,22 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.BeanNameAware; -import org.springframework.beans.factory.InitializingBean; import org.springframework.context.Lifecycle; import org.springframework.integration.ConfigurationException; +import org.springframework.integration.channel.ChannelRegistry; +import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.handler.MessageHandlerNotRunningException; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageHandlingException; +import org.springframework.integration.scheduling.Schedule; +import org.springframework.integration.scheduling.Subscription; /** * Base class for {@link MessageEndpoint} implementations. * * @author Mark Fisher */ -public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware, Lifecycle, InitializingBean { +public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware, Lifecycle { protected final Log logger = LogFactory.getLog(this.getClass()); @@ -49,6 +52,16 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware private volatile boolean running; + private volatile Schedule schedule; + + private volatile EndpointTrigger trigger; + + private volatile Subscription subscription; + + private volatile String outputChannelName; + + private volatile ChannelRegistry channelRegistry; + private final Object lifecycleMonitor = new Object(); @@ -64,6 +77,60 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware this.setName(beanName); } + public void setSchedule(Schedule schedule) { + this.schedule = schedule; + } + + public Schedule getSchedule() { + return this.schedule; + } + + public void setTrigger(EndpointTrigger trigger) { + this.trigger = trigger; + } + + public EndpointTrigger getTrigger() { + return this.trigger; + } + + public Subscription getSubscription() { + return this.subscription; + } + + public void setSubscription(Subscription subscription) { + this.subscription = subscription; + } + + /** + * Set the name of the channel to which this endpoint should send reply + * messages. + */ + public void setOutputChannelName(String outputChannelName) { + this.outputChannelName = outputChannelName; + } + + public MessageChannel getOutputChannel() { + if (this.outputChannelName != null && this.channelRegistry != null) { + return this.channelRegistry.lookupChannel(this.outputChannelName); + } + return null; + } + + public String getOutputChannelName() { + return this.outputChannelName; + } + + /** + * Set the channel registry to use for looking up channels by name. + */ + public void setChannelRegistry(ChannelRegistry channelRegistry) { + this.channelRegistry = channelRegistry; + } + + protected ChannelRegistry getChannelRegistry() { + return this.channelRegistry; + } + public void setAutoStartup(boolean autoStartup) { this.autoStartup = autoStartup; } @@ -131,7 +198,7 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware } } - public final boolean invoke(Message message) { + public final boolean send(Message message) { if (message == null) { throw new IllegalArgumentException("Message must not be null."); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointTrigger.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointTrigger.java new file mode 100644 index 0000000000..a483fcf61d --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointTrigger.java @@ -0,0 +1,59 @@ +/* + * 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. + * 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.endpoint; + +import org.springframework.integration.dispatcher.BroadcastingDispatcher; +import org.springframework.integration.dispatcher.PollingDispatcher; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageSource; +import org.springframework.integration.message.PollCommand; +import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.integration.scheduling.Schedule; + +/** + * A {@link PollingDispatcher} implementation that sends poll command + * trigger messages to endpoints. + * + * @author Mark Fisher + */ +public class EndpointTrigger extends PollingDispatcher { + + /** + * Create a trigger with the specified {@link Schedule}. + */ + public EndpointTrigger(Schedule schedule) { + super(new PollCommandMessageSource(), new BroadcastingDispatcher(), schedule); + } + + /** + * Create a trigger. A {@link PollingSchedule} will be created + * with the specified interval. + */ + public EndpointTrigger(long interval) { + this(new PollingSchedule(interval)); + } + + + private static class PollCommandMessageSource implements MessageSource { + + public Message receive() { + return new GenericMessage(new PollCommand()); + } + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java index a7d4a2e476..09d298ca68 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java @@ -35,7 +35,7 @@ import org.springframework.util.StringUtils; * * @author Mark Fisher */ -public class HandlerEndpoint extends TargetEndpoint implements MessageProducingEndpoint { +public class HandlerEndpoint extends TargetEndpoint { private volatile MessageHandler handler; @@ -43,8 +43,6 @@ public class HandlerEndpoint extends TargetEndpoint implements MessageProducingE private volatile long replyTimeout = 1000; - private volatile String outputChannelName; - private volatile boolean returnAddressOverrides = false; @@ -73,14 +71,6 @@ public class HandlerEndpoint extends TargetEndpoint implements MessageProducingE this.replyTimeout = replyTimeout; } - /** - * Set the name of the channel to which this endpoint should send reply - * messages. - */ - public void setOutputChannelName(String outputChannelName) { - this.outputChannelName = outputChannelName; - } - public void setReturnAddressOverrides(boolean returnAddressOverrides) { this.returnAddressOverrides = returnAddressOverrides; } @@ -128,18 +118,6 @@ public class HandlerEndpoint extends TargetEndpoint implements MessageProducingE return null; } - public MessageChannel getOutputChannel() { - ChannelRegistry registry = this.getChannelRegistry(); - if (this.outputChannelName != null && registry != null) { - return registry.lookupChannel(this.outputChannelName); - } - return null; - } - - public String getOutputChannelName() { - return this.outputChannelName; - } - private static class HandlerInvokingTarget implements MessageTarget { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageConsumingEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageConsumingEndpoint.java deleted file mode 100644 index ce721e0633..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageConsumingEndpoint.java +++ /dev/null @@ -1,28 +0,0 @@ -/* - * 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. - * 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.endpoint; - -import org.springframework.integration.scheduling.Subscription; - -/** - * @author Mark Fisher - */ -public interface MessageConsumingEndpoint { - - Subscription getSubscription(); - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java index b7abe24fdf..8c9a3227c8 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java @@ -16,17 +16,30 @@ package org.springframework.integration.endpoint; -import org.springframework.integration.message.Message; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.channel.ChannelRegistryAware; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.scheduling.Schedule; +import org.springframework.integration.scheduling.Subscription; /** * Base interface for message endpoints. * * @author Mark Fisher */ -public interface MessageEndpoint { +public interface MessageEndpoint extends MessageTarget, ChannelRegistryAware, InitializingBean { String getName(); - boolean invoke(Message message); + Schedule getSchedule(); + + EndpointTrigger getTrigger(); + + Subscription getSubscription(); + + String getOutputChannelName(); + + MessageChannel getOutputChannel(); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageProducingEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageProducingEndpoint.java deleted file mode 100644 index d661232512..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageProducingEndpoint.java +++ /dev/null @@ -1,30 +0,0 @@ -/* - * 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. - * 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.endpoint; - -import org.springframework.integration.channel.MessageChannel; - -/** - * @author Mark Fisher - */ -public interface MessageProducingEndpoint { - - String getOutputChannelName(); - - MessageChannel getOutputChannel(); - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java index ee208a6b47..ed5b50e992 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java @@ -24,7 +24,6 @@ import org.springframework.integration.message.MessageDeliveryAware; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.PollCommand; import org.springframework.integration.message.MessageSource; -import org.springframework.integration.scheduling.Schedule; import org.springframework.util.Assert; /** @@ -39,25 +38,15 @@ public class SourceEndpoint extends AbstractEndpoint { private final SimpleDispatcher dispatcher = new SimpleDispatcher(new DispatcherPolicy()); - private volatile Schedule schedule; - public SourceEndpoint(MessageSource source, MessageChannel channel) { Assert.notNull(source, "source must not be null"); Assert.notNull(channel, "channel must not be null"); this.source = source; - this.dispatcher.subscribe(channel); + this.dispatcher.addTarget(channel); } - public void setSchedule(Schedule schedule) { - this.schedule = schedule; - } - - public Schedule getSchedule() { - return this.schedule; - } - protected boolean supports(Message message) { return (message.getPayload() instanceof PollCommand); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java index 7c1359b6e1..ce91d01062 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java @@ -16,10 +16,11 @@ package org.springframework.integration.endpoint; -import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistryAware; +import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.message.PollCommand; import org.springframework.integration.message.selector.MessageSelector; import org.springframework.integration.scheduling.Subscription; import org.springframework.util.Assert; @@ -29,8 +30,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class TargetEndpoint extends AbstractEndpoint - implements MessageTarget, MessageConsumingEndpoint, ChannelRegistryAware { +public class TargetEndpoint extends AbstractEndpoint { private volatile MessageTarget target; @@ -38,8 +38,6 @@ public class TargetEndpoint extends AbstractEndpoint private volatile MessageSelector selector; - private volatile ChannelRegistry channelRegistry; - private volatile boolean initialized; private final Object initializationMonitor = new Object(); @@ -75,24 +73,13 @@ public class TargetEndpoint extends AbstractEndpoint this.subscription = subscription; } - /** - * Set the channel registry to use for looking up channels by name. - */ - public void setChannelRegistry(ChannelRegistry channelRegistry) { - this.channelRegistry = channelRegistry; - } - - protected ChannelRegistry getChannelRegistry() { - return this.channelRegistry; - } - protected void initialize() { synchronized (this.initializationMonitor) { if (this.initialized) { return; } - if (this.target instanceof ChannelRegistryAware && this.channelRegistry != null) { - ((ChannelRegistryAware) this.target).setChannelRegistry(this.channelRegistry); + if (this.target instanceof ChannelRegistryAware && this.getChannelRegistry() != null) { + ((ChannelRegistryAware) this.target).setChannelRegistry(this.getChannelRegistry()); } this.initialized = true; } @@ -100,6 +87,23 @@ public class TargetEndpoint extends AbstractEndpoint @Override protected final boolean doInvoke(Message message) { + if (message.getPayload() instanceof PollCommand) { + MessageChannel channel = this.getSubscription().getChannel(); + if (channel == null && this.getSubscription().getChannelName() != null) { + channel = this.getChannelRegistry().lookupChannel(this.getSubscription().getChannelName()); + } + if (channel != null) { + Message receivedMessage = channel.receive(5000); + if (receivedMessage != null) { + return this.doInvoke(receivedMessage); + } + } + else if (logger.isDebugEnabled()) { + logger.debug("TargetEndpoint unable to resolve channel '" + + this.getSubscription().getChannelName() + "'"); + } + return false; + } if (this.selector != null && !this.selector.accept(message)) { return false; } @@ -111,8 +115,4 @@ public class TargetEndpoint extends AbstractEndpoint return true; } - public boolean send(Message message) { - return this.invoke(message); - } - } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/MessageBusTests.java index 0d8bb08236..20df684935 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -30,6 +30,7 @@ import org.springframework.beans.factory.BeanCreationException; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.SourceEndpoint; import org.springframework.integration.handler.MessageHandler; @@ -129,8 +130,7 @@ public class MessageBusTests { @Test public void testBothHandlersReceivePublishSubscribeMessage() throws InterruptedException { - DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(true); - QueueChannel inputChannel = new QueueChannel(10, dispatcherPolicy); + PublishSubscribeChannel inputChannel = new PublishSubscribeChannel(); QueueChannel outputChannel1 = new QueueChannel(); QueueChannel outputChannel2 = new QueueChannel(); final CountDownLatch latch = new CountDownLatch(2); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java deleted file mode 100644 index 459f45cb01..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java +++ /dev/null @@ -1,470 +0,0 @@ -/* - * 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. - * 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.bus; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertTrue; - -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ScheduledThreadPoolExecutor; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; - -import org.junit.Test; - -import org.springframework.beans.factory.InitializingBean; -import org.springframework.context.Lifecycle; -import org.springframework.integration.channel.DispatcherPolicy; -import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.config.MessageEndpointBeanPostProcessor; -import org.springframework.integration.endpoint.AbstractEndpoint; -import org.springframework.integration.endpoint.ConcurrencyPolicy; -import org.springframework.integration.endpoint.HandlerEndpoint; -import org.springframework.integration.endpoint.interceptor.ConcurrencyInterceptor; -import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.handler.MessageHandlerRejectedExecutionException; -import org.springframework.integration.handler.TestHandlers; -import org.springframework.integration.message.ErrorMessage; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageDeliveryException; -import org.springframework.integration.message.MessageTarget; -import org.springframework.integration.message.StringMessage; -import org.springframework.integration.message.selector.PayloadTypeSelector; -import org.springframework.integration.scheduling.MessagePublishingErrorHandler; -import org.springframework.integration.scheduling.SimpleTaskScheduler; - -/** - * @author Mark Fisher - */ -public class SubscriptionManagerTests { - - private SimpleTaskScheduler scheduler = new SimpleTaskScheduler(new ScheduledThreadPoolExecutor(10)); - - - @Test - public void testNonBroadcastingDispatcherSendsToExactlyOneEndpoint() throws InterruptedException { - final AtomicInteger counter1 = new AtomicInteger(); - final AtomicInteger counter2 = new AtomicInteger(); - final CountDownLatch latch = new CountDownLatch(1); - MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); - MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - QueueChannel channel = new QueueChannel(); - channel.send(new StringMessage(1, "test")); - SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - manager.addTarget(createEndpoint(handler1, true)); - manager.addTarget(createEndpoint(handler2, true)); - manager.start(); - latch.await(2000, TimeUnit.MILLISECONDS); - assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("exactly one handler should have received message", 1, counter1.get() + counter2.get()); - } - - @Test - public void testBroadcastingDispatcherSendsToAllEndpoints() throws InterruptedException { - final AtomicInteger counter1 = new AtomicInteger(); - final AtomicInteger counter2 = new AtomicInteger(); - final CountDownLatch latch = new CountDownLatch(2); - MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); - MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true)); - channel.send(new StringMessage(1, "test")); - SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - manager.addTarget(createEndpoint(handler1, true)); - manager.addTarget(createEndpoint(handler2, true)); - manager.start(); - latch.await(2000, TimeUnit.MILLISECONDS); - assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("both handlers should have received message", 2, counter1.get() + counter2.get()); - } - - @Test - public void testNonBroadcastingDispatcherSkipsInactiveExecutor() throws InterruptedException { - final AtomicInteger counter1 = new AtomicInteger(); - final AtomicInteger counter2 = new AtomicInteger(); - final AtomicInteger counter3 = new AtomicInteger(); - final CountDownLatch latch = new CountDownLatch(1); - MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); - MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); - QueueChannel channel = new QueueChannel(); - SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - MessageTarget inactiveEndpoint = createEndpoint(handler1, true); - manager.addTarget(inactiveEndpoint); - manager.addTarget(createEndpoint(handler2, true)); - manager.addTarget(createEndpoint(handler3, true)); - manager.start(); - ((Lifecycle) inactiveEndpoint).stop(); - channel.send(new StringMessage(1, "test")); - latch.await(2000, TimeUnit.MILLISECONDS); - assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("inactive handler should not have received message", 0, counter1.get()); - assertEquals("exactly one handler should have received message", 1, counter2.get() + counter3.get()); - } - - @Test - public void testBroadcastingDispatcherSkipsInactiveExecutor() throws InterruptedException { - final AtomicInteger counter1 = new AtomicInteger(); - final AtomicInteger counter2 = new AtomicInteger(); - final AtomicInteger counter3 = new AtomicInteger(); - final CountDownLatch latch = new CountDownLatch(2); - MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); - MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); - QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true)); - SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - MessageTarget inactiveEndpoint = createEndpoint(handler2, true); - manager.addTarget(createEndpoint(handler1, true)); - manager.addTarget(inactiveEndpoint); - manager.addTarget(createEndpoint(handler3, true)); - manager.start(); - ((Lifecycle) inactiveEndpoint).stop(); - channel.send(new StringMessage(1, "test")); - latch.await(2000, TimeUnit.MILLISECONDS); - assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("inactive handler should not have received message", 0, counter2.get()); - assertEquals("both active handlers should have received message", 2, counter1.get() + counter3.get()); - } - - @Test - public void testDispatcherWithNoExecutorsDoesNotFail() { - QueueChannel channel = new QueueChannel(); - channel.send(new StringMessage(1, "test")); - SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - manager.start(); - } - - @Test - public void testBroadcastingDispatcherReachesRejectionLimitAndShouldFail() throws InterruptedException { - final AtomicInteger counter1 = new AtomicInteger(); - final AtomicInteger counter3 = new AtomicInteger(); - final CountDownLatch latch = new CountDownLatch(2); - MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); - MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); - QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true)); - channel.getDispatcherPolicy().setRejectionLimit(2); - channel.getDispatcherPolicy().setRetryInterval(3); - channel.send(new StringMessage(1, "test")); - SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - manager.addTarget(createEndpoint(handler1, true)); - manager.addTarget(new MessageTarget() { - public boolean send(Message message) { - throw new MessageHandlerRejectedExecutionException(message); - } - }); - manager.addTarget(createEndpoint(handler3, true)); - QueueChannel errorChannel = new QueueChannel(); - scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel)); - manager.start(); - latch.await(2000, TimeUnit.MILLISECONDS); - assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - Message errorMessage = errorChannel.receive(1000); - assertNotNull(errorMessage); - assertTrue(errorMessage instanceof ErrorMessage); - assertEquals(MessageDeliveryException.class, ((ErrorMessage) errorMessage).getPayload().getClass()); - } - - @Test - public void testBroadcastingDispatcherReachesRejectionLimitAndShouldNotFail() throws InterruptedException { - final AtomicInteger counter1 = new AtomicInteger(); - final AtomicInteger counter2 = new AtomicInteger(); - final CountDownLatch latch = new CountDownLatch(3); - MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); - MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true)); - channel.getDispatcherPolicy().setRejectionLimit(2); - channel.getDispatcherPolicy().setRetryInterval(3); - channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false); - channel.send(new StringMessage(1, "test")); - SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - manager.addTarget(createEndpoint(handler1, false)); - manager.addTarget(createEndpoint(new MessageHandler() { - public Message handle(Message message) { - latch.countDown(); - throw new MessageHandlerRejectedExecutionException(message); - } - }, false)); - manager.addTarget(createEndpoint(handler2, false)); - manager.start(); - latch.await(2000, TimeUnit.MILLISECONDS); - assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("both non-rejecting handlers should have received message", 2, counter1.get() + counter2.get()); - } - - @Test - public void testNonBroadcastingDispatcherReachesRejectionLimitAndShouldFail() throws InterruptedException { - final CountDownLatch latch = new CountDownLatch(4); - MessageHandler handler1 = TestHandlers.rejectingCountDownHandler(latch); - MessageHandler handler2 = TestHandlers.rejectingCountDownHandler(latch); - QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(false)); - channel.send(new StringMessage(1, "test")); - SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - channel.getDispatcherPolicy().setRejectionLimit(2); - channel.getDispatcherPolicy().setRetryInterval(3); - manager.addTarget(createEndpoint(handler1, false)); - manager.addTarget(createEndpoint(handler2, false)); - QueueChannel errorChannel = new QueueChannel(); - scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel)); - manager.start(); - latch.await(2000, TimeUnit.MILLISECONDS); - assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - Message errorMessage = errorChannel.receive(500); - assertNotNull(errorMessage); - assertTrue(errorMessage instanceof ErrorMessage); - assertEquals(MessageDeliveryException.class, ((ErrorMessage) errorMessage).getPayload().getClass()); - } - - @Test - public void testNonBroadcastingDispatcherReachesRejectionLimitButShouldNotFail() throws InterruptedException { - final AtomicInteger counter1 = new AtomicInteger(); - final AtomicInteger counter2 = new AtomicInteger(); - final AtomicInteger rejectedCounter1 = new AtomicInteger(); - final AtomicInteger rejectedCounter2 = new AtomicInteger(); - final CountDownLatch latch = new CountDownLatch(4); - QueueChannel channel = new QueueChannel(); - channel.getDispatcherPolicy().setRejectionLimit(2); - channel.getDispatcherPolicy().setRetryInterval(3); - channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false); - channel.send(new StringMessage(1, "test")); - SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - manager.addTarget(createEndpoint(new MessageHandler() { - public Message handle(Message message) { - rejectedCounter1.incrementAndGet(); - latch.countDown(); - throw new MessageHandlerRejectedExecutionException(message); - } - }, false)); - manager.addTarget(createEndpoint(new MessageHandler() { - public Message handle(Message message) { - rejectedCounter2.incrementAndGet(); - latch.countDown(); - throw new MessageHandlerRejectedExecutionException(message); - } - }, false)); - manager.start(); - latch.await(2000, TimeUnit.MILLISECONDS); - assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("latch should have counted down within allotted time", 0, latch.getCount()); - assertEquals("rejecting handlers should not have received message", 0, counter1.get() + counter2.get()); - assertEquals("handler1 should have rejected two times", 2, rejectedCounter1.get()); - assertEquals("handler2 should have rejected two times", 2, rejectedCounter2.get()); - } - - @Test - public void testNonBroadcastingDispatcherWithOneEndpointSucceeding() throws InterruptedException { - final AtomicInteger counter1 = new AtomicInteger(); - final AtomicInteger counter2 = new AtomicInteger(); - final AtomicInteger counter3 = new AtomicInteger(); - final AtomicInteger rejectedCounter1 = new AtomicInteger(); - final AtomicInteger rejectedCounter2 = new AtomicInteger(); - final AtomicInteger rejectedCounter3 = new AtomicInteger(); - final CountDownLatch latch = new CountDownLatch(5); - final MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); - dispatcherPolicy.setRejectionLimit(2); - dispatcherPolicy.setRetryInterval(3); - dispatcherPolicy.setShouldFailOnRejectionLimit(false); - QueueChannel channel = new QueueChannel(25, dispatcherPolicy); - channel.send(new StringMessage(1, "test")); - SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - manager.addTarget(createEndpoint(new MessageHandler() { - public Message handle(Message message) { - rejectedCounter1.incrementAndGet(); - latch.countDown(); - throw new MessageHandlerRejectedExecutionException(message); - } - }, false)); - manager.addTarget(createEndpoint(new MessageHandler() { - public Message handle(Message message) { - if (rejectedCounter2.get() == 1) { - return handler2.handle(message); - } - rejectedCounter2.incrementAndGet(); - latch.countDown(); - throw new MessageHandlerRejectedExecutionException(message); - } - }, false)); - manager.addTarget(createEndpoint(new MessageHandler() { - public Message handle(Message message) { - rejectedCounter3.incrementAndGet(); - latch.countDown(); - throw new MessageHandlerRejectedExecutionException(message); - } - }, false)); - manager.start(); - latch.await(2000, TimeUnit.MILLISECONDS); - assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("handler1 should not have received message", 0, counter1.get()); - assertEquals("handler2 should have received message the second time", 1, counter2.get()); - assertEquals("handler3 should not have received message", 0, counter3.get()); - assertEquals("handler1 should have rejected two times", 2, rejectedCounter1.get()); - assertEquals("handler2 should have rejected one time", 1, rejectedCounter2.get()); - assertEquals("handler3 should have rejected one time", 1, rejectedCounter3.get()); - } - - @Test - public void testBroadcastingDispatcherStillRetriesRejectedExecutorAfterOtherSucceeds() throws InterruptedException { - final AtomicInteger counter1 = new AtomicInteger(); - final AtomicInteger counter2 = new AtomicInteger(); - final AtomicInteger rejectedCounter1 = new AtomicInteger(); - final AtomicInteger rejectedCounter2 = new AtomicInteger(); - final CountDownLatch latch = new CountDownLatch(8); - final MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); - final MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(true); - dispatcherPolicy.setRejectionLimit(5); - dispatcherPolicy.setRetryInterval(3); - dispatcherPolicy.setShouldFailOnRejectionLimit(false); - QueueChannel channel = new QueueChannel(25, dispatcherPolicy); - channel.send(new StringMessage(1, "test")); - SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - manager.addTarget(createEndpoint(new MessageHandler() { - public Message handle(Message message) { - if (rejectedCounter1.get() == 2) { - return handler1.handle(message); - } - rejectedCounter1.incrementAndGet(); - latch.countDown(); - throw new MessageHandlerRejectedExecutionException(message); - } - }, false)); - manager.addTarget(createEndpoint(new MessageHandler() { - public Message handle(Message message) { - if (rejectedCounter2.get() == 4) { - return handler2.handle(message); - } - rejectedCounter2.incrementAndGet(); - latch.countDown(); - throw new MessageHandlerRejectedExecutionException(message); - } - }, false)); - manager.start(); - latch.await(2000, TimeUnit.MILLISECONDS); - assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("handler1 should have received one message", 1, counter1.get()); - assertEquals("handler2 should have received one message", 1, counter2.get()); - assertEquals("handler1 should have rejected two times", 2, rejectedCounter1.get()); - assertEquals("handler2 should have rejected four times", 4, rejectedCounter2.get()); - } - - @Test - public void testTwoExecutorsWithSelectorsAndOneAccepts() throws InterruptedException { - final AtomicInteger counter1 = new AtomicInteger(); - final AtomicInteger counter2 = new AtomicInteger(); - final CountDownLatch latch = new CountDownLatch(1); - MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); - MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - QueueChannel channel = new QueueChannel(); - channel.send(new StringMessage(1, "test")); - SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1); - HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2); - endpoint1.setMessageSelector(new PayloadTypeSelector(Integer.class)); - endpoint2.setMessageSelector(new PayloadTypeSelector(String.class)); - manager.addTarget(endpoint1); - manager.addTarget(endpoint2); - manager.start(); - latch.await(2000, TimeUnit.MILLISECONDS); - assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("handler1 should not have accepted the message", 0, counter1.get()); - assertEquals("handler2 should have accepted the message", 1, counter2.get()); - } - - @Test - public void testTwoExecutorsWithSelectorsAndNeitherAccepts() throws InterruptedException { - final AtomicInteger counter1 = new AtomicInteger(); - final AtomicInteger counter2 = new AtomicInteger(); - final AtomicInteger selectorCounter1 = new AtomicInteger(); - final AtomicInteger selectorCounter2 = new AtomicInteger(); - final CountDownLatch selectorLatch = new CountDownLatch(2); - final CountDownLatch handlerLatch = new CountDownLatch(1); - MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, handlerLatch); - MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, handlerLatch); - QueueChannel channel = new QueueChannel(); - channel.send(new StringMessage(1, "test")); - SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - final HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1); - final HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2); - endpoint1.setMessageSelector(new PayloadTypeSelector(Integer.class) { - @Override - public boolean accept(Message message) { - selectorCounter1.incrementAndGet(); - selectorLatch.countDown(); - return super.accept(message); - } - }); - endpoint2.setMessageSelector(new PayloadTypeSelector(Integer.class) { - @Override - public boolean accept(Message message) { - selectorCounter2.incrementAndGet(); - selectorLatch.countDown(); - return super.accept(message); - } - }); - manager.addTarget(endpoint1); - manager.addTarget(endpoint2); - manager.start(); - selectorLatch.await(2000, TimeUnit.MILLISECONDS); - assertEquals("messages should have been dispatched within allotted time", 0, selectorLatch.getCount()); - assertEquals("handler1 should not have accepted the message", 0, counter1.get()); - assertEquals("handler2 should not have accepted the message", 0, counter2.get()); - assertEquals("executor1 should have had exactly one attempt", 1, selectorCounter1.get()); - assertEquals("executor2 should have had exactly one attempt", 1, selectorCounter2.get()); - assertEquals("handlerLatch should not have counted down", 1, handlerLatch.getCount()); - } - - @Test - public void testBroadcastingDispatcherWithSelectorsAndOneAccepts() throws InterruptedException { - final AtomicInteger counter1 = new AtomicInteger(); - final AtomicInteger counter2 = new AtomicInteger(); - final CountDownLatch latch = new CountDownLatch(1); - MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); - MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true)); - channel.send(new StringMessage(1, "test")); - SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1); - HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2); - endpoint1.setMessageSelector(new PayloadTypeSelector(Integer.class)); - endpoint2.setMessageSelector(new PayloadTypeSelector(String.class)); - manager.addTarget(endpoint1); - manager.addTarget(endpoint2); - manager.start(); - latch.await(2000, TimeUnit.MILLISECONDS); - assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); - assertEquals("endpoint1 should not have accepted the message", 0, counter1.get()); - assertEquals("endpoint2 should have accepted the message", 1, counter2.get()); - } - - - private static MessageTarget createEndpoint(MessageHandler handler, boolean asynchronous) { - MessageTarget endpoint = new HandlerEndpoint(handler); - if (asynchronous) { - MessageEndpointBeanPostProcessor postProcessor = new MessageEndpointBeanPostProcessor(); - ((AbstractEndpoint) endpoint).addInterceptor(new ConcurrencyInterceptor(new ConcurrencyPolicy(1, 1))); - endpoint = (MessageTarget) postProcessor.postProcessAfterInitialization(endpoint, "test-endpoint"); - } - try { - ((InitializingBean) endpoint).afterPropertiesSet(); - } - catch (Exception e) { - throw new RuntimeException("failed to initialize endpoint", e); - } - return endpoint; - } - -} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java index b045853ff1..cca9c19fcd 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java @@ -20,27 +20,22 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ScheduledThreadPoolExecutor; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; - import org.junit.Test; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.FatalBeanException; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.bus.SubscriptionManager; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PublishSubscribeChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.TestChannelInterceptor; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.MessagePriority; import org.springframework.integration.message.StringMessage; -import org.springframework.integration.message.MessageTarget; -import org.springframework.integration.scheduling.SimpleTaskScheduler; /** * @author Mark Fisher @@ -65,43 +60,11 @@ public class ChannelParserTests { } @Test - public void testPointToPointChannelByDefault() throws InterruptedException { + public void testQueueChannelByDefault() throws InterruptedException { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "channelParserTests.xml", this.getClass()); - MessageChannel channel = (MessageChannel) context.getBean("pointToPointChannelByDefault"); - channel.send(new StringMessage("test")); - SimpleTaskScheduler scheduler = new SimpleTaskScheduler(new ScheduledThreadPoolExecutor(1)); - SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - AtomicInteger counter = new AtomicInteger(); - CountDownLatch latch = new CountDownLatch(1); - TestTarget target1 = new TestTarget(counter, latch); - TestTarget target2 = new TestTarget(counter, latch); - manager.addTarget(target1); - manager.addTarget(target2); - manager.start(); - latch.await(500, TimeUnit.MILLISECONDS); - assertEquals(0, latch.getCount()); - assertEquals(1, counter.get()); - } - - @Test - public void testPointToPointChannelExplicitlyConfigured() throws InterruptedException { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "channelParserTests.xml", this.getClass()); - MessageChannel channel = (MessageChannel) context.getBean("pointToPointChannelExplicitlyConfigured"); - channel.send(new StringMessage("test")); - SimpleTaskScheduler scheduler = new SimpleTaskScheduler(new ScheduledThreadPoolExecutor(1)); - SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - AtomicInteger counter = new AtomicInteger(); - CountDownLatch latch = new CountDownLatch(1); - TestTarget target1 = new TestTarget(counter, latch); - TestTarget target2 = new TestTarget(counter, latch); - manager.addTarget(target1); - manager.addTarget(target2); - manager.start(); - latch.await(500, TimeUnit.MILLISECONDS); - assertEquals(0, latch.getCount()); - assertEquals(1, counter.get()); + MessageChannel channel = (MessageChannel) context.getBean("queueChannelByDefault"); + assertEquals(QueueChannel.class, channel.getClass()); } @Test @@ -109,26 +72,27 @@ public class ChannelParserTests { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "channelParserTests.xml", this.getClass()); MessageChannel channel = (MessageChannel) context.getBean("publishSubscribeChannel"); - channel.send(new StringMessage("test")); - SimpleTaskScheduler scheduler = new SimpleTaskScheduler(new ScheduledThreadPoolExecutor(1)); - SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - AtomicInteger counter = new AtomicInteger(); - CountDownLatch latch = new CountDownLatch(2); - TestTarget target1 = new TestTarget(counter, latch); - TestTarget target2 = new TestTarget(counter, latch); - manager.addTarget(target1); - manager.addTarget(target2); - manager.start(); - latch.await(500, TimeUnit.MILLISECONDS); - assertEquals(0, latch.getCount()); - assertEquals(2, counter.get()); + assertEquals(PublishSubscribeChannel.class, channel.getClass()); + } + + @Test + public void testPublishSubscribeChannelWithTaskExecutorReference() throws InterruptedException { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "channelParserTests.xml", this.getClass()); + MessageChannel channel = (MessageChannel) context.getBean("publishSubscribeChannelWithTaskExecutorRef"); + assertEquals(PublishSubscribeChannel.class, channel.getClass()); + DirectFieldAccessor accessor = new DirectFieldAccessor(channel); + accessor = new DirectFieldAccessor(accessor.getPropertyValue("dispatcher")); + Object taskExecutorProperty = accessor.getPropertyValue("taskExecutor"); + Object taskExecutorBean = context.getBean("taskExecutor"); + assertEquals(taskExecutorBean, taskExecutorProperty); } @Test public void testDefaultDispatcherPolicy() throws InterruptedException { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "channelParserTests.xml", this.getClass()); - MessageChannel channel = (MessageChannel) context.getBean("pointToPointChannelByDefault"); + MessageChannel channel = (MessageChannel) context.getBean("queueChannelByDefault"); DispatcherPolicy dispatcherPolicy = channel.getDispatcherPolicy(); assertFalse(dispatcherPolicy.isPublishSubscribe()); assertEquals(DispatcherPolicy.DEFAULT_REJECTION_LIMIT, dispatcherPolicy.getRejectionLimit()); @@ -142,7 +106,6 @@ public class ChannelParserTests { "channelParserTests.xml", this.getClass()); MessageChannel channel = (MessageChannel) context.getBean("channelWithDispatcherPolicy"); DispatcherPolicy dispatcherPolicy = channel.getDispatcherPolicy(); - assertTrue(dispatcherPolicy.isPublishSubscribe()); assertEquals(7, dispatcherPolicy.getRejectionLimit()); assertEquals(77, dispatcherPolicy.getRetryInterval()); assertFalse(dispatcherPolicy.getShouldFailOnRejectionLimit()); @@ -267,23 +230,4 @@ public class ChannelParserTests { assertTrue(threwException); } - - private static class TestTarget implements MessageTarget { - - private AtomicInteger counter; - - private CountDownLatch latch; - - TestTarget(AtomicInteger counter, CountDownLatch latch) { - this.counter = counter; - this.latch = latch; - } - - public boolean send(Message message) { - this.counter.incrementAndGet(); - this.latch.countDown(); - return true; - } - } - } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml index 45194b4dec..311e5ab148 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml @@ -9,13 +9,14 @@ - + - + - + - + @@ -27,4 +28,6 @@ + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageEndpointBeanPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageEndpointBeanPostProcessorTests.java index b970c65af6..7d5644a200 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageEndpointBeanPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/MessageEndpointBeanPostProcessorTests.java @@ -47,14 +47,13 @@ public class MessageEndpointBeanPostProcessorTests { public void testHandlerEndpointWithAdviceChain() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "messageEndpointBeanPostProcessorTests.xml", this.getClass()); - context.start(); MessageEndpoint endpoint = (MessageEndpoint) context.getBean("handlerEndpointWithAdvice"); assertTrue(AopUtils.isAopProxy(endpoint)); TestBeforeAdvice beforeAdvice = (TestBeforeAdvice) context.getBean("simpleAdvice"); TestEndpointInterceptor interceptor = (TestEndpointInterceptor) context.getBean("interceptor"); assertEquals(0, beforeAdvice.getCount()); assertEquals(0, interceptor.getCount()); - endpoint.invoke(new StringMessage("test")); + endpoint.send(new StringMessage("test")); assertEquals(1, beforeAdvice.getCount()); assertEquals(2, interceptor.getCount()); context.stop(); @@ -72,14 +71,13 @@ public class MessageEndpointBeanPostProcessorTests { public void testTargetEndpointWithAdviceChain() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "messageEndpointBeanPostProcessorTests.xml", this.getClass()); - context.start(); MessageEndpoint endpoint = (MessageEndpoint) context.getBean("targetEndpointWithAdvice"); assertTrue(AopUtils.isAopProxy(endpoint)); TestBeforeAdvice beforeAdvice = (TestBeforeAdvice) context.getBean("simpleAdvice"); TestEndpointInterceptor interceptor = (TestEndpointInterceptor) context.getBean("interceptor"); assertEquals(0, beforeAdvice.getCount()); assertEquals(0, interceptor.getCount()); - endpoint.invoke(new StringMessage("test")); + endpoint.send(new StringMessage("test")); assertEquals(1, beforeAdvice.getCount()); assertEquals(2, interceptor.getCount()); context.stop(); @@ -97,14 +95,13 @@ public class MessageEndpointBeanPostProcessorTests { public void testSourceEndpointWithAdviceChain() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "messageEndpointBeanPostProcessorTests.xml", this.getClass()); - context.start(); MessageEndpoint endpoint = (MessageEndpoint) context.getBean("sourceEndpointWithAdvice"); assertTrue(AopUtils.isAopProxy(endpoint)); TestBeforeAdvice beforeAdvice = (TestBeforeAdvice) context.getBean("simpleAdvice"); TestEndpointInterceptor interceptor = (TestEndpointInterceptor) context.getBean("interceptor"); assertEquals(0, beforeAdvice.getCount()); assertEquals(0, interceptor.getCount()); - endpoint.invoke(new CommandMessage(new PollCommand())); + endpoint.send(new CommandMessage(new PollCommand())); assertEquals(1, beforeAdvice.getCount()); assertEquals(2, interceptor.getCount()); context.stop(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/messageEndpointBeanPostProcessorTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/messageEndpointBeanPostProcessorTests.xml index 3b27bc0faf..e6297bc8ad 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/messageEndpointBeanPostProcessorTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/messageEndpointBeanPostProcessorTests.xml @@ -7,7 +7,7 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd"> - + 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 970326abf0..db877afdfd 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 @@ -40,7 +40,7 @@ public class SimpleDispatcherTests { public void testSingleMessage() throws InterruptedException { SimpleDispatcher dispatcher = new SimpleDispatcher(new DispatcherPolicy()); final CountDownLatch latch = new CountDownLatch(1); - dispatcher.subscribe(createEndpoint(TestHandlers.countDownHandler(latch))); + dispatcher.addTarget(createEndpoint(TestHandlers.countDownHandler(latch))); dispatcher.send(new StringMessage("test")); latch.await(500, TimeUnit.MILLISECONDS); assertEquals(0, latch.getCount()); @@ -52,8 +52,8 @@ public class SimpleDispatcherTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); - dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch))); - dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch))); + dispatcher.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch))); + dispatcher.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch))); dispatcher.send(new StringMessage("test")); latch.await(500, TimeUnit.MILLISECONDS); assertEquals(0, latch.getCount()); @@ -66,8 +66,8 @@ public class SimpleDispatcherTests { final CountDownLatch latch = new CountDownLatch(2); final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); - dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch))); - dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch))); + dispatcher.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch))); + dispatcher.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch))); dispatcher.send(new StringMessage("test")); latch.await(500, TimeUnit.MILLISECONDS); assertEquals(0, latch.getCount()); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java index 921d9d6bd8..9be2c154f5 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java @@ -42,7 +42,7 @@ public class SourceEndpointTests { QueueChannel channel = new QueueChannel(); SourceEndpoint endpoint = new SourceEndpoint(source, channel); endpoint.afterPropertiesSet(); - endpoint.invoke(new CommandMessage(new PollCommand())); + endpoint.send(new CommandMessage(new PollCommand())); Message message = channel.receive(1000); assertNotNull("message should not be null", message); assertEquals("testing.1", message.getPayload()); @@ -55,7 +55,7 @@ public class SourceEndpointTests { SourceEndpoint endpoint = new SourceEndpoint(source, channel); endpoint.setAutoStartup(false); endpoint.afterPropertiesSet(); - endpoint.invoke(new CommandMessage(new PollCommand())); + endpoint.send(new CommandMessage(new PollCommand())); } private static class TestSource implements MessageSource { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptorTests.java index 9621ccf5b0..067aa54c8c 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptorTests.java @@ -79,13 +79,13 @@ public class TransactionInterceptorTests { TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager"); final MessageEndpoint endpoint = (MessageEndpoint) context.getBean("required"); assertEquals(0, txManager.getCommitCount()); - endpoint.invoke(new StringMessage("test")); + endpoint.send(new StringMessage("test")); assertEquals(1, txManager.getCommitCount()); TestTransactionManager outerTxManager = new TestTransactionManager(); TransactionTemplate txTemplate = new TransactionTemplate(outerTxManager); txTemplate.execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { - return endpoint.invoke(new StringMessage("test")); + return endpoint.send(new StringMessage("test")); } }); assertEquals(1, outerTxManager.getCommitCount()); @@ -100,13 +100,13 @@ public class TransactionInterceptorTests { TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager"); final MessageEndpoint endpoint = (MessageEndpoint) context.getBean("requiresNew"); assertEquals(0, txManager.getCommitCount()); - endpoint.invoke(new StringMessage("test")); + endpoint.send(new StringMessage("test")); assertEquals(1, txManager.getCommitCount()); TestTransactionManager outerTxManager = new TestTransactionManager(); TransactionTemplate txTemplate = new TransactionTemplate(outerTxManager); txTemplate.execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { - return endpoint.invoke(new StringMessage("test")); + return endpoint.send(new StringMessage("test")); } }); assertEquals(1, outerTxManager.getCommitCount()); @@ -121,13 +121,13 @@ public class TransactionInterceptorTests { TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager"); final MessageEndpoint endpoint = (MessageEndpoint) context.getBean("supports"); assertEquals(0, txManager.getCommitCount()); - endpoint.invoke(new StringMessage("test")); + endpoint.send(new StringMessage("test")); assertEquals(0, txManager.getCommitCount()); TestTransactionManager outerTxManager = new TestTransactionManager(); TransactionTemplate txTemplate = new TransactionTemplate(outerTxManager); txTemplate.execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { - return endpoint.invoke(new StringMessage("test")); + return endpoint.send(new StringMessage("test")); } }); assertEquals(0, txManager.getCommitCount()); @@ -141,13 +141,13 @@ public class TransactionInterceptorTests { TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager"); final MessageEndpoint endpoint = (MessageEndpoint) context.getBean("notSupported"); assertEquals(0, txManager.getCommitCount()); - endpoint.invoke(new StringMessage("test")); + endpoint.send(new StringMessage("test")); assertEquals(0, txManager.getCommitCount()); TestTransactionManager outerTxManager = new TestTransactionManager(); TransactionTemplate txTemplate = new TransactionTemplate(outerTxManager); txTemplate.execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { - return endpoint.invoke(new StringMessage("test")); + return endpoint.send(new StringMessage("test")); } }); assertEquals(0, txManager.getCommitCount()); @@ -159,7 +159,7 @@ public class TransactionInterceptorTests { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "transactionInterceptorPropagationTests.xml", this.getClass()); final MessageEndpoint endpoint = (MessageEndpoint) context.getBean("mandatory"); - endpoint.invoke(new StringMessage("test")); + endpoint.send(new StringMessage("test")); } }