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