From ea5e04e7ac5c07c6c863a5f0b122cc887df5a465 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 7 Dec 2007 22:48:51 +0000 Subject: [PATCH] Moved consumer creation and lifecycle from endpoint to message bus. --- .../integration/MessageBus.java | 100 +++++++++++++++--- .../integration/channel/ChannelRegistry.java | 48 --------- .../endpoint/EndpointRegistry.java | 44 -------- .../endpoint/GenericMessageEndpoint.java | 93 +++------------- .../integration/endpoint/MessageEndpoint.java | 6 ++ .../integration/MessageBusTests.java | 6 +- .../consumer/EventDrivenConsumerTests.java | 9 +- .../consumer/FixedDelayConsumerTests.java | 16 +-- .../consumer/FixedRateConsumerTests.java | 16 +-- .../endpoint/GenericMessageEndpointTests.java | 15 ++- .../integration/messageBusTests.xml | 2 +- 11 files changed, 130 insertions(+), 225 deletions(-) delete mode 100644 spring-eai-core/src/main/java/org/springframework/integration/channel/ChannelRegistry.java delete mode 100644 spring-eai-core/src/main/java/org/springframework/integration/endpoint/EndpointRegistry.java diff --git a/spring-eai-core/src/main/java/org/springframework/integration/MessageBus.java b/spring-eai-core/src/main/java/org/springframework/integration/MessageBus.java index 4a54b7538e..0efd1a6c3b 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/MessageBus.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/MessageBus.java @@ -16,7 +16,10 @@ package org.springframework.integration; +import java.util.List; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -25,9 +28,13 @@ import org.springframework.beans.BeansException; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.Lifecycle; -import org.springframework.integration.channel.ChannelRegistry; +import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.endpoint.EndpointRegistry; +import org.springframework.integration.channel.consumer.AbstractConsumer; +import org.springframework.integration.channel.consumer.ConsumerType; +import org.springframework.integration.channel.consumer.EventDrivenConsumer; +import org.springframework.integration.channel.consumer.FixedDelayConsumer; +import org.springframework.integration.channel.consumer.FixedRateConsumer; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.util.Assert; @@ -37,16 +44,22 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class MessageBus implements ApplicationContextAware { +public class MessageBus implements ChannelResolver, ApplicationContextAware, Lifecycle { private final Log logger = LogFactory.getLog(getClass()); - private ChannelRegistry channelRegistry = new ChannelRegistry(); + private Map channels = new ConcurrentHashMap(); - private EndpointRegistry endpointRegistry = new EndpointRegistry(); + private Map endpoints = new ConcurrentHashMap(); + + private List consumers = new CopyOnWriteArrayList(); private ApplicationContext applicationContext; + private boolean running; + + private Object lifecycleMonitor = new Object(); + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { Assert.notNull(applicationContext, "applicationContext must not be null"); @@ -60,7 +73,10 @@ public class MessageBus implements ApplicationContextAware { Map channelBeans = (Map) this.applicationContext .getBeansOfType(MessageChannel.class); for (Map.Entry entry : channelBeans.entrySet()) { - this.registerChannel(entry.getKey(), entry.getValue()); + this.channels.put(entry.getKey(), entry.getValue()); + if (logger.isInfoEnabled()) { + logger.info("registered channel '" + entry.getKey() + "'"); + } } } @@ -69,24 +85,74 @@ public class MessageBus implements ApplicationContextAware { Map endpointBeans = (Map) this.applicationContext .getBeansOfType(MessageEndpoint.class); for (Map.Entry entry : endpointBeans.entrySet()) { - this.registerEndpoint(entry.getKey(), entry.getValue()); + this.endpoints.put(entry.getKey(), entry.getValue()); + if (logger.isInfoEnabled()) { + logger.info("registered endpoint '" + entry.getKey() + "'"); + } } } - public void registerChannel(String name, MessageChannel channel) { - this.channelRegistry.register(name, channel); - if (logger.isInfoEnabled()) { - logger.info("registering channel '" + name + "'"); + public MessageChannel resolve(String channelName) { + return this.channels.get(channelName); + } + + public boolean isRunning() { + synchronized (this.lifecycleMonitor) { + return this.running; } } - public void registerEndpoint(String name, MessageEndpoint endpoint) { - if (endpoint instanceof Lifecycle) { - ((Lifecycle) endpoint).start(); + public void start() { + synchronized (this.lifecycleMonitor) { + if (!this.isRunning()) { + this.running = true; + this.activateEndpoints(); + } } - this.endpointRegistry.register(name, endpoint); - if (logger.isInfoEnabled()) { - logger.info("registering endpoint '" + name + "'"); + } + + public void stop() { + synchronized (this.lifecycleMonitor) { + if (this.isRunning()) { + this.running = false; + this.deactivateEndpoints(); + } + } + } + + private void activateEndpoints() { + for (MessageEndpoint endpoint : this.endpoints.values()) { + MessageSource source = endpoint.getSource(); + ConsumerType consumerType = endpoint.getConsumerType(); + AbstractConsumer consumer = createConsumer(consumerType, source, endpoint); + consumer.initialize(); + consumer.start(); + } + } + + private void deactivateEndpoints() { + for (AbstractConsumer consumer : this.consumers) { + consumer.stop(); + } + } + + + /** + * Create a consumer based upon the specified consumer type. + */ + private AbstractConsumer createConsumer(ConsumerType type, MessageSource source, MessageEndpoint endpoint) { + if (type.equals(ConsumerType.EVENT_DRIVEN)) { + return new EventDrivenConsumer(source, endpoint); + } + else if (type.equals(ConsumerType.FIXED_RATE)) { + return new FixedRateConsumer(source, endpoint); + } + else if (type.equals(ConsumerType.FIXED_DELAY)) { + return new FixedDelayConsumer(source, endpoint); + } + else { + throw new UnsupportedOperationException("the consumerType '" + + type.name() + "' is not supported."); } } diff --git a/spring-eai-core/src/main/java/org/springframework/integration/channel/ChannelRegistry.java b/spring-eai-core/src/main/java/org/springframework/integration/channel/ChannelRegistry.java deleted file mode 100644 index 025dcb6373..0000000000 --- a/spring-eai-core/src/main/java/org/springframework/integration/channel/ChannelRegistry.java +++ /dev/null @@ -1,48 +0,0 @@ -/* - * Copyright 2002-2007 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.Map; -import java.util.concurrent.ConcurrentHashMap; - -import org.springframework.util.Assert; - -/** - * A registry for maintaining message channels in a map keyed by name. - * - * @author Mark Fisher - */ -public class ChannelRegistry implements ChannelResolver { - - private Map mappings = new ConcurrentHashMap(); - - - public void register(String name, MessageChannel channel) { - Assert.notNull(name, "name must not be null"); - Assert.notNull(channel, "channel must not be null"); - this.mappings.put(name, channel); - } - - public MessageChannel lookup(String name) { - return this.mappings.get(name); - } - - public MessageChannel resolve(String channelName) { - return this.lookup(channelName); - } - -} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/EndpointRegistry.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/EndpointRegistry.java deleted file mode 100644 index bd2d43b05f..0000000000 --- a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/EndpointRegistry.java +++ /dev/null @@ -1,44 +0,0 @@ -/* - * Copyright 2002-2007 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 java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -import org.springframework.util.Assert; - -/** - * A registry for maintaining message endpoints in a map keyed by name. - * - * @author Mark Fisher - */ -public class EndpointRegistry { - - private Map mappings = new ConcurrentHashMap(); - - - public void register(String name, MessageEndpoint endpoint) { - Assert.notNull(name, "name must not be null"); - Assert.notNull(endpoint, "endpoint must not be null"); - this.mappings.put(name, endpoint); - } - - public MessageEndpoint lookup(String name) { - return this.mappings.get(name); - } - -} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java index b3468c27cd..5b739ecba7 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java @@ -16,16 +16,11 @@ package org.springframework.integration.endpoint; -import org.springframework.context.Lifecycle; import org.springframework.integration.MessageHandlingException; import org.springframework.integration.MessageSource; import org.springframework.integration.MessageTarget; import org.springframework.integration.channel.ChannelResolver; -import org.springframework.integration.channel.consumer.AbstractConsumer; import org.springframework.integration.channel.consumer.ConsumerType; -import org.springframework.integration.channel.consumer.EventDrivenConsumer; -import org.springframework.integration.channel.consumer.FixedDelayConsumer; -import org.springframework.integration.channel.consumer.FixedRateConsumer; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; @@ -42,7 +37,7 @@ import org.springframework.integration.message.Message; * * @author Mark Fisher */ -public class GenericMessageEndpoint implements MessageEndpoint, Lifecycle { +public class GenericMessageEndpoint implements MessageEndpoint { private MessageSource source; @@ -50,25 +45,10 @@ public class GenericMessageEndpoint implements MessageEndpoint, Lifecycle { private MessageHandler handler; - private ConsumerType consumerType = ConsumerType.EVENT_DRIVEN; - - private AbstractConsumer consumer; - private ChannelResolver channelResolver; - private boolean running; + private ConsumerType consumerType = ConsumerType.EVENT_DRIVEN; - private Object lifecycleMonitor = new Object(); - - - public GenericMessageEndpoint() {} - - /** - * Create an endpoint to consume messages from the given source. - */ - public GenericMessageEndpoint(MessageSource source) { - this.source = source; - } /** * Set the source from which this endpoint receives messages. @@ -77,6 +57,13 @@ public class GenericMessageEndpoint implements MessageEndpoint, Lifecycle { this.source = source; } + /** + * Return the source from which this endpoint receives messages. + */ + public MessageSource getSource() { + return this.source; + } + /** * Set the target to which this endpoint can send messages. */ @@ -92,15 +79,14 @@ public class GenericMessageEndpoint implements MessageEndpoint, Lifecycle { } /** - * Set the consumer type to use for this endpoint's source. - * @see ConsumerType + * Set the type of consumer to use for this endpoint. */ public void setConsumerType(ConsumerType consumerType) { this.consumerType = consumerType; } /** - * Return the consumer type to use for this endpoint's source. + * Return the type of consumer to use for this endpoint. */ public ConsumerType getConsumerType() { return this.consumerType; @@ -110,66 +96,11 @@ public class GenericMessageEndpoint implements MessageEndpoint, Lifecycle { * Set the channel resolver strategy to use when a message * provides a 'replyChannelName'. */ - public void setChannelResolver(ChannelResolver channelResolver) { + public void setChannelResolver(final ChannelResolver channelResolver) { this.channelResolver = channelResolver; } - /** - * Create a consumer based upon the specified consumer type. - */ - protected AbstractConsumer createDefaultConsumer() { - if (this.consumerType.equals(ConsumerType.EVENT_DRIVEN)) { - return new EventDrivenConsumer(this.source, this); - } - else if (this.consumerType.equals(ConsumerType.FIXED_RATE)) { - return new FixedRateConsumer(this.source, this); - } - else if (this.consumerType.equals(ConsumerType.FIXED_DELAY)) { - return new FixedDelayConsumer(this.source, this); - } - else { - throw new UnsupportedOperationException("the consumerType '" - + this.consumerType.name() + "' is not supported."); - } - } - - /** - * Start the consumer. - */ - public final void start() { - synchronized (this.lifecycleMonitor) { - if (this.source != null && this.consumer == null) { - this.consumer = createDefaultConsumer(); - } - this.consumer.initialize(); - this.running = true; - } - } - - /** - * Stop the consumer. - */ - public final void stop() { - synchronized (this.lifecycleMonitor) { - if (this.running) { - if (this.consumer != null) { - this.consumer.stop(); - } - this.running = false; - } - } - } - - /** - * Return whether this endpoint is running (and hence its consumer). - */ - public final boolean isRunning() { - synchronized (this.lifecycleMonitor) { - return this.running; - } - } - public void messageReceived(Message message) { if (this.handler == null) { target.send(message); diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java index 67e14f0524..60a87ba200 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java @@ -16,6 +16,8 @@ package org.springframework.integration.endpoint; +import org.springframework.integration.MessageSource; +import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.channel.consumer.ConsumerType; import org.springframework.integration.message.Message; @@ -28,6 +30,10 @@ public interface MessageEndpoint { ConsumerType getConsumerType(); + MessageSource getSource(); + + void setChannelResolver(ChannelResolver channelResolver); + void messageReceived(Message message); } diff --git a/spring-eai-core/src/test/java/org/springframework/integration/MessageBusTests.java b/spring-eai-core/src/test/java/org/springframework/integration/MessageBusTests.java index bb617a7647..a76ee05be1 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/MessageBusTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/MessageBusTests.java @@ -33,7 +33,7 @@ import org.springframework.integration.message.Message; * @author Mark Fisher */ public class MessageBusTests { - +/* @Test public void testStandaloneWithEndpoint() { MessageBus bus = new MessageBus(); @@ -60,10 +60,12 @@ public class MessageBusTests { Message result = targetChannel.receive(10); assertNull(result); } +*/ @Test public void testAutodetectionWithApplicationContext() { - ApplicationContext context = new ClassPathXmlApplicationContext("messageBusTests.xml", this.getClass()); + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("messageBusTests.xml", this.getClass()); + context.start(); MessageChannel sourceChannel = (MessageChannel) context.getBean("sourceChannel"); sourceChannel.send(new DocumentMessage("123", "test")); MessageChannel targetChannel = (MessageChannel) context.getBean("targetChannel"); diff --git a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/EventDrivenConsumerTests.java b/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/EventDrivenConsumerTests.java index 1d6414328c..f846095436 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/EventDrivenConsumerTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/EventDrivenConsumerTests.java @@ -24,8 +24,8 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; - import org.springframework.integration.channel.PointToPointChannel; +import org.springframework.integration.endpoint.GenericMessageEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; @@ -50,17 +50,13 @@ public class EventDrivenConsumerTests { executor.setMaxPoolSize(maxConcurrency); executor.setQueueCapacity(0); PointToPointChannel channel = new PointToPointChannel(); - MessageEndpoint endpoint = new MessageEndpoint() { + MessageEndpoint endpoint = new GenericMessageEndpoint() { public void messageReceived(Message message) { counter.incrementAndGet(); latch.countDown(); activeSum.set(activeSum.addAndGet(executor.getActiveCount())); maxActive.set(Math.max(executor.getActiveCount(), maxActive.get())); } - - public ConsumerType getConsumerType() { - return ConsumerType.EVENT_DRIVEN; - } }; EventDrivenConsumer consumer = new EventDrivenConsumer(channel, endpoint); consumer.setExecutor(executor); @@ -70,6 +66,7 @@ public class EventDrivenConsumerTests { consumer.setMaxMessagesPerTask(1); consumer.setReceiveTimeout(100); consumer.initialize(); + consumer.start(); for (int i = 0; i < messagesToSend - 110; i++) { channel.send(new DocumentMessage(1, "fast-1." + (i+1))); } diff --git a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedDelayConsumerTests.java b/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedDelayConsumerTests.java index c6237ebfd7..5cf4ad2f02 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedDelayConsumerTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedDelayConsumerTests.java @@ -24,8 +24,8 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; - import org.springframework.integration.channel.PointToPointChannel; +import org.springframework.integration.endpoint.GenericMessageEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; @@ -41,19 +41,16 @@ public class FixedDelayConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); PointToPointChannel channel = new PointToPointChannel(); - MessageEndpoint endpoint = new MessageEndpoint() { + MessageEndpoint endpoint = new GenericMessageEndpoint() { public void messageReceived(Message message) { counter.incrementAndGet(); latch.countDown(); } - - public ConsumerType getConsumerType() { - return ConsumerType.FIXED_DELAY; - } }; FixedDelayConsumer consumer = new FixedDelayConsumer(channel, endpoint); consumer.setPollInterval(10); consumer.initialize(); + consumer.start(); for (int i = 0; i < messagesToSend; i++) { channel.send(new DocumentMessage(1, "test " + (i+1))); } @@ -67,19 +64,16 @@ public class FixedDelayConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); PointToPointChannel channel = new PointToPointChannel(); - MessageEndpoint endpoint = new MessageEndpoint() { + MessageEndpoint endpoint = new GenericMessageEndpoint() { public void messageReceived(Message message) { counter.incrementAndGet(); latch.countDown(); } - - public ConsumerType getConsumerType() { - return ConsumerType.FIXED_DELAY; - } }; FixedDelayConsumer consumer = new FixedDelayConsumer(channel, endpoint); consumer.setPollInterval(10); consumer.initialize(); + consumer.start(); for (int i = 0; i < messagesToSend; i++) { channel.send(new DocumentMessage(1, "test " + (i+1))); } diff --git a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedRateConsumerTests.java b/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedRateConsumerTests.java index 5388e2057c..2394d84d37 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedRateConsumerTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedRateConsumerTests.java @@ -24,8 +24,8 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; - import org.springframework.integration.channel.PointToPointChannel; +import org.springframework.integration.endpoint.GenericMessageEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; @@ -41,19 +41,16 @@ public class FixedRateConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); PointToPointChannel channel = new PointToPointChannel(); - MessageEndpoint endpoint = new MessageEndpoint() { + MessageEndpoint endpoint = new GenericMessageEndpoint() { public void messageReceived(Message message) { counter.incrementAndGet(); latch.countDown(); } - - public ConsumerType getConsumerType() { - return ConsumerType.FIXED_RATE; - } }; FixedRateConsumer consumer = new FixedRateConsumer(channel, endpoint); consumer.setPollInterval(10); consumer.initialize(); + consumer.start(); for (int i = 0; i < messagesToSend; i++) { channel.send(new DocumentMessage(1, "test " + (i+1))); } @@ -67,19 +64,16 @@ public class FixedRateConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); PointToPointChannel channel = new PointToPointChannel(); - MessageEndpoint endpoint = new MessageEndpoint() { + MessageEndpoint endpoint = new GenericMessageEndpoint() { public void messageReceived(Message message) { counter.incrementAndGet(); latch.countDown(); } - - public ConsumerType getConsumerType() { - return ConsumerType.FIXED_RATE; - } }; FixedRateConsumer consumer = new FixedRateConsumer(channel, endpoint); consumer.setPollInterval(10); consumer.initialize(); + consumer.start(); for (int i = 0; i < messagesToSend; i++) { channel.send(new DocumentMessage(1, "test " + (i+1))); } diff --git a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java index 2011db25d9..f51a19759d 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java @@ -24,6 +24,7 @@ import org.junit.Test; import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PointToPointChannel; +import org.springframework.integration.channel.consumer.EventDrivenConsumer; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; @@ -42,10 +43,13 @@ public class GenericMessageEndpointTests { return new DocumentMessage("123", "hello " + message.getPayload()); } }; - GenericMessageEndpoint endpoint = new GenericMessageEndpoint(channel); + GenericMessageEndpoint endpoint = new GenericMessageEndpoint(); + endpoint.setSource(channel); endpoint.setHandler(handler); endpoint.setTarget(replyChannel); - endpoint.start(); + EventDrivenConsumer consumer = new EventDrivenConsumer(channel, endpoint); + consumer.initialize(); + consumer.start(); DocumentMessage testMessage = new DocumentMessage(1, "test"); channel.send(testMessage); Message reply = replyChannel.receive(10); @@ -70,10 +74,13 @@ public class GenericMessageEndpointTests { return null; } }; - GenericMessageEndpoint endpoint = new GenericMessageEndpoint(channel); + GenericMessageEndpoint endpoint = new GenericMessageEndpoint(); + endpoint.setSource(channel); endpoint.setHandler(handler); endpoint.setChannelResolver(channelResolver); - endpoint.start(); + EventDrivenConsumer consumer = new EventDrivenConsumer(channel, endpoint); + consumer.initialize(); + consumer.start(); DocumentMessage testMessage = new DocumentMessage(1, "test"); testMessage.getHeader().setReplyChannelName("replyChannel"); channel.send(testMessage); diff --git a/spring-eai-core/src/test/java/org/springframework/integration/messageBusTests.xml b/spring-eai-core/src/test/java/org/springframework/integration/messageBusTests.xml index 86375edf7f..7048ee20fa 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/messageBusTests.xml +++ b/spring-eai-core/src/test/java/org/springframework/integration/messageBusTests.xml @@ -11,7 +11,7 @@ - +