diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java index b9bb695ec2..facef65a6f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java @@ -19,9 +19,6 @@ package org.springframework.integration.endpoint; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; -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; @@ -47,8 +44,6 @@ import org.springframework.util.Assert; */ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryAware, InitializingBean, BeanNameAware { - private final Log logger = LogFactory.getLog(this.getClass()); - private String name; private MessageHandler handler; @@ -172,6 +167,9 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA if (this.isRunning()) { return; } + if (!initialized) { + this.afterPropertiesSet(); + } if (this.handler instanceof Lifecycle) { ((Lifecycle) handler).start(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java index b02b4f9c87..53508631d0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java @@ -26,9 +26,8 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; import org.springframework.integration.channel.SimpleChannel; -import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.message.Message; +import org.springframework.integration.handler.TestHandlers; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.scheduling.PollingSchedule; @@ -39,34 +38,24 @@ public class FixedDelayConsumerTests { @Test public void testAllSentMessagesAreReceivedWithinTimeLimit() throws Exception { - int messagesToSend = 20; + int messagesToSend = 10; final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); SimpleChannel channel = new SimpleChannel(); - MessageHandler handler = new MessageHandler() { - public Message handle(Message message) { - counter.incrementAndGet(); - latch.countDown(); - return null; - } - }; + MessageHandler handler = TestHandlers.countingCountDownHandler(counter, latch); MessageBus bus = new MessageBus(); bus.initialize(); bus.registerChannel("testChannel", channel); - PollingSchedule schedule = new PollingSchedule(10); + PollingSchedule schedule = new PollingSchedule(5); schedule.setFixedRate(false); - ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(); - concurrencyPolicy.setCoreSize(1); - concurrencyPolicy.setMaxSize(1); - Subscription subscription = new Subscription(); + Subscription subscription = new Subscription(channel); subscription.setSchedule(schedule); - subscription.setChannelName("testChannel"); - bus.registerHandler("testHandler", handler, subscription, concurrencyPolicy); + bus.registerHandler("testHandler", handler, subscription); bus.start(); for (int i = 0; i < messagesToSend; i++) { channel.send(new GenericMessage(1, "test " + (i+1))); } - latch.await(3000, TimeUnit.MILLISECONDS); + latch.await(100, TimeUnit.MILLISECONDS); assertEquals(messagesToSend, counter.get()); } @@ -76,30 +65,22 @@ public class FixedDelayConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); SimpleChannel channel = new SimpleChannel(); - MessageHandler handler = new MessageHandler() { - public Message handle(Message message) { - counter.incrementAndGet(); - latch.countDown(); - return null; - } - }; + MessageHandler handler = TestHandlers.countingCountDownHandler(counter, latch); MessageBus bus = new MessageBus(); bus.initialize(); bus.registerChannel("testChannel", channel); PollingSchedule schedule = new PollingSchedule(10); schedule.setFixedRate(false); Subscription subscription = new Subscription(channel); - subscription.setChannelName("testChannel"); subscription.setSchedule(schedule); bus.registerHandler("testHandler", handler, subscription); for (int i = 0; i < messagesToSend; i++) { channel.send(new GenericMessage(1, "test " + (i+1))); } bus.start(); - latch.await(80, TimeUnit.MILLISECONDS); + latch.await(100, TimeUnit.MILLISECONDS); int count = counter.get(); - assertTrue("received " + count + ", expected less than 11", counter.get() < 11); - assertTrue("received " + count + ", expected more than 5", counter.get() > 5); + assertTrue("received " + count + ", expected less than 20", counter.get() < 20); bus.stop(); }