From 82cb7bcb623fb1728dd89db499532718933b5aad Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 26 Feb 2008 04:11:10 +0000 Subject: [PATCH] Added PriorityChannel (INT-115). --- .../channel/AbstractMessageChannel.java | 5 +- .../channel/BaseBlockingQueueChannel.java | 103 +++++++++++++++++ .../integration/channel/MessageChannel.java | 3 + .../integration/channel/PriorityChannel.java | 106 ++++++++++++++++++ .../integration/channel/SimpleChannel.java | 86 +------------- .../integration/message/MessageHeader.java | 10 ++ .../integration/bus/MessageBusTests.java | 12 +- .../channel/PriorityChannelTests.java | 103 +++++++++++++++++ .../ChannelPollingMessageRetrieverTests.java | 6 +- .../DefaultMessageDispatcherTests.java | 16 +-- 10 files changed, 349 insertions(+), 101 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/BaseBlockingQueueChannel.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index 51d6d92d7c..d86e4d5c1c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -39,7 +39,6 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanName private final Log logger = LogFactory.getLog(this.getClass()); - private volatile String name; private final ChannelInterceptorList interceptors = new ChannelInterceptorList(); @@ -182,7 +181,7 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanName * value indicates that the method should block until either the message is * accepted or the blocking thread is interrupted. */ - protected abstract boolean doSend(Message message, long timeout); + protected abstract boolean doSend(Message message, long timeout); /** * Subclasses must implement this method. A non-negative timeout indicates @@ -191,7 +190,7 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanName * indicates that the method should block until either a message is * available or the blocking thread is interrupted. */ - protected abstract Message doReceive(long timeout); + protected abstract Message doReceive(long timeout); /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/BaseBlockingQueueChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/BaseBlockingQueueChannel.java new file mode 100644 index 0000000000..81ef4bf3c4 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/BaseBlockingQueueChannel.java @@ -0,0 +1,103 @@ +/* + * 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.ArrayList; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.TimeUnit; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.selector.MessageSelector; +import org.springframework.util.Assert; + +/** + * Base implementation for message channels that delegate to a {@link BlockingQueue}. + * + * @author Mark Fisher + */ +public class BaseBlockingQueueChannel extends AbstractMessageChannel { + + private final BlockingQueue> queue; + + + /** + * Create a channel with the specified queue and dispatcher policy. + */ + public BaseBlockingQueueChannel(BlockingQueue> queue, DispatcherPolicy dispatcherPolicy) { + super((dispatcherPolicy != null) ? dispatcherPolicy : new DispatcherPolicy()); + Assert.notNull(queue, "'queue' must not be null"); + this.queue = queue; + } + + + protected boolean doSend(Message message, long timeout) { + Assert.notNull(message, "'message' must not be null"); + try { + if (timeout > 0) { + return this.queue.offer(message, timeout, TimeUnit.MILLISECONDS); + } + if (timeout == 0) { + return this.queue.offer(message); + } + queue.put(message); + return true; + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + } + + protected Message doReceive(long timeout) { + try { + if (timeout > 0) { + return queue.poll(timeout, TimeUnit.MILLISECONDS); + } + if (timeout == 0) { + return queue.poll(); + } + return queue.take(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return null; + } + } + + public List> clear() { + List> clearedMessages = new ArrayList>(); + this.queue.drainTo(clearedMessages); + return clearedMessages; + } + + public List> purge(MessageSelector selector) { + if (selector == null) { + return this.clear(); + } + List> purgedMessages = new ArrayList>(); + Object[] array = this.queue.toArray(); + for (Object o : array) { + Message message = (Message) o; + if (!selector.accept(message) && this.queue.remove(message)) { + purgedMessages.add(message); + } + } + return purgedMessages; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java index 61a2ef35f0..233535739f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java @@ -28,6 +28,9 @@ import org.springframework.integration.message.selector.MessageSelector; */ public interface MessageChannel { + static final int DEFAULT_CAPACITY = 100; + + /** * Return the name of this channel. */ diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java new file mode 100644 index 0000000000..85bdcc86a1 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java @@ -0,0 +1,106 @@ +/* + * 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.Comparator; +import java.util.concurrent.PriorityBlockingQueue; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageHeader; + +/** + * A message channel that prioritizes messages based on a {@link Comparator}. + * The default comparator is based upon the message header's 'priority'. + * + * @author Mark Fisher + */ +public class PriorityChannel extends BaseBlockingQueueChannel { + + private final Semaphore semaphore; + + + /** + * Create a channel with the specified queue capacity and dispatcher policy. + * Priority will be based upon the provided {@link Comparator}. + */ + public PriorityChannel(int capacity, DispatcherPolicy dispatcherPolicy, Comparator> comparator) { + super(new PriorityBlockingQueue>(capacity, comparator), dispatcherPolicy); + this.semaphore = new Semaphore(capacity, true); + } + + /** + * Create a channel with the specified queue capacity and dispatcher policy. + * Priority will be based upon the value of {@link MessageHeader#getPriority()}. + */ + public PriorityChannel(int capacity, DispatcherPolicy dispatcherPolicy) { + this(capacity, dispatcherPolicy, new MessagePriorityComparator()); + } + + /** + * Create a channel with the specified queue capacity and default dispatcher + * policy. Priority will be based on the value of {@link MessageHeader#getPriority()}. + */ + public PriorityChannel(int capacity) { + this(capacity, null, new MessagePriorityComparator()); + } + + /** + * Create a channel with the default queue capacity and dispatcher policy. + * Priority will be based on the value of {@link MessageHeader#getPriority()}. + */ + public PriorityChannel() { + this(DEFAULT_CAPACITY, null, new MessagePriorityComparator()); + } + + + @Override + protected boolean doSend(Message message, long timeout) { + try { + if (!this.semaphore.tryAcquire(timeout, TimeUnit.MILLISECONDS)) { + return false; + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + return super.doSend(message, 0); + } + + @Override + protected Message doReceive(long timeout) { + Message message = super.doReceive(timeout); + if (message != null) { + this.semaphore.release(); + return message; + } + return null; + } + + + private static class MessagePriorityComparator implements Comparator> { + + public int compare(Message message1, Message message2) { + Integer priority1 = message1.getHeader().getPriority(); + Integer priority2 = message2.getHeader().getPriority(); + return priority1.compareTo(priority2); + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java index 3c0db258c6..9ce0232575 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java @@ -16,16 +16,10 @@ package org.springframework.integration.channel; -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.SynchronousQueue; -import java.util.concurrent.TimeUnit; import org.springframework.integration.message.Message; -import org.springframework.integration.message.selector.MessageSelector; -import org.springframework.util.Assert; /** * Simple implementation of a message channel. Each {@link Message} is @@ -34,25 +28,14 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class SimpleChannel extends AbstractMessageChannel { - - public static final int DEFAULT_CAPACITY = 100; - - - private BlockingQueue> queue; - +public class SimpleChannel extends BaseBlockingQueueChannel { /** * Create a channel with the specified queue capacity and dispatcher policy. */ public SimpleChannel(int capacity, DispatcherPolicy dispatcherPolicy) { - super((dispatcherPolicy != null) ? dispatcherPolicy : new DispatcherPolicy()); - if (capacity > 0) { - this.queue = new LinkedBlockingQueue>(capacity); - } - else { - this.queue = new SynchronousQueue>(true); - } + super((capacity > 0) ? new LinkedBlockingQueue>(capacity) : + new SynchronousQueue>(true), dispatcherPolicy); } /** @@ -62,13 +45,6 @@ public class SimpleChannel extends AbstractMessageChannel { this(capacity, null); } - /** - * Create a channel with the default queue capacity and the specified dispatcher policy. - */ - public SimpleChannel(DispatcherPolicy dispatcherPolicy) { - this(DEFAULT_CAPACITY, dispatcherPolicy); - } - /** * Create a channel with the default queue capacity. */ @@ -76,60 +52,4 @@ public class SimpleChannel extends AbstractMessageChannel { this(DEFAULT_CAPACITY, null); } - - protected boolean doSend(Message message, long timeout) { - Assert.notNull(message, "'message' must not be null"); - try { - if (timeout > 0) { - return this.queue.offer(message, timeout, TimeUnit.MILLISECONDS); - } - if (timeout == 0) { - return this.queue.offer(message); - } - queue.put(message); - return true; - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return false; - } - } - - protected Message doReceive(long timeout) { - try { - if (timeout > 0) { - return queue.poll(timeout, TimeUnit.MILLISECONDS); - } - if (timeout == 0) { - return queue.poll(); - } - return queue.take(); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return null; - } - } - - public List> clear() { - List> clearedMessages = new ArrayList>(); - this.queue.drainTo(clearedMessages); - return clearedMessages; - } - - public List> purge(MessageSelector selector) { - if (selector == null) { - return this.clear(); - } - List> purgedMessages = new ArrayList>(); - Object[] array = this.queue.toArray(); - for (Object o : array) { - Message message = (Message) o; - if (!selector.accept(message) && this.queue.remove(message)) { - purgedMessages.add(message); - } - } - return purgedMessages; - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java index 6843575e42..020effcb8e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java @@ -46,6 +46,8 @@ public class MessageHeader { private volatile int sequenceSize = 1; + private volatile int priority = 0; + private final Properties properties = new Properties(); private final ConcurrentMap attributes = new ConcurrentHashMap(); @@ -106,6 +108,14 @@ public class MessageHeader { this.sequenceSize = sequenceSize; } + public int getPriority() { + return this.priority; + } + + public void setPriority(int priority) { + this.priority = priority; + } + public String getProperty(String key) { return this.properties.getProperty(key); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java index bf1c2fcd7e..c1d0bf6061 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -132,20 +132,23 @@ public class MessageBusTests { } @Test - public void testBothHandlersReceivePublishSubscribeMessage() { + public void testBothHandlersReceivePublishSubscribeMessage() throws InterruptedException { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(true); - SimpleChannel inputChannel = new SimpleChannel(dispatcherPolicy); + SimpleChannel inputChannel = new SimpleChannel(10, dispatcherPolicy); SimpleChannel outputChannel1 = new SimpleChannel(); SimpleChannel outputChannel2 = new SimpleChannel(); + final CountDownLatch latch = new CountDownLatch(2); MessageHandler handler1 = new MessageHandler() { public Message handle(Message message) { message.getHeader().setReturnAddress("output1"); + latch.countDown(); return message; } }; MessageHandler handler2 = new MessageHandler() { public Message handle(Message message) { message.getHeader().setReturnAddress("output2"); + latch.countDown(); return message; } }; @@ -157,8 +160,9 @@ public class MessageBusTests { bus.registerHandler("handler2", handler2, new Subscription(inputChannel)); bus.start(); inputChannel.send(new StringMessage(1, "testing")); - Message message1 = outputChannel1.receive(500); - Message message2 = outputChannel2.receive(500); + latch.await(500, TimeUnit.MILLISECONDS); + Message message1 = outputChannel1.receive(100); + Message message2 = outputChannel2.receive(100); bus.stop(); assertTrue("both handlers should have received and replied to the message", (message1 != null && message2 != null)); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.java new file mode 100644 index 0000000000..94f1bef8f8 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.java @@ -0,0 +1,103 @@ +/* + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.util.Comparator; + +import org.junit.Test; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class PriorityChannelTests { + + @Test + public void testCapacityEnforced() { + PriorityChannel channel = new PriorityChannel(3); + assertTrue(channel.send(new StringMessage("test1"), 0)); + assertTrue(channel.send(new StringMessage("test2"), 0)); + assertTrue(channel.send(new StringMessage("test3"), 0)); + assertFalse(channel.send(new StringMessage("test4"), 0)); + channel.receive(0); + assertTrue(channel.send(new StringMessage("test5"))); + } + + @Test + public void testDefaultComparator() { + PriorityChannel channel = new PriorityChannel(5); + Message priority1 = createPriorityMessage(1); + Message priority2 = createPriorityMessage(2); + Message priority3 = createPriorityMessage(3); + Message priority4 = createPriorityMessage(4); + Message priority5 = createPriorityMessage(5); + channel.send(priority4); + channel.send(priority3); + channel.send(priority5); + channel.send(priority1); + channel.send(priority2); + assertEquals("test-1", channel.receive(0).getPayload()); + assertEquals("test-2", channel.receive(0).getPayload()); + assertEquals("test-3", channel.receive(0).getPayload()); + assertEquals("test-4", channel.receive(0).getPayload()); + assertEquals("test-5", channel.receive(0).getPayload()); + } + + @Test + public void testCustomComparator() { + PriorityChannel channel = new PriorityChannel(5, null, new StringPayloadComparator()); + Message messageA = new StringMessage("A"); + Message messageB = new StringMessage("B"); + Message messageC = new StringMessage("C"); + Message messageD = new StringMessage("D"); + Message messageE = new StringMessage("E"); + channel.send(messageC); + channel.send(messageA); + channel.send(messageE); + channel.send(messageD); + channel.send(messageB); + assertEquals("A", channel.receive(0).getPayload()); + assertEquals("B", channel.receive(0).getPayload()); + assertEquals("C", channel.receive(0).getPayload()); + assertEquals("D", channel.receive(0).getPayload()); + assertEquals("E", channel.receive(0).getPayload()); + } + + + private static Message createPriorityMessage(int priority) { + Message message = new StringMessage("test-" + priority); + message.getHeader().setPriority(priority); + return message; + } + + + private static class StringPayloadComparator implements Comparator> { + + public int compare(Message message1, Message message2) { + String s1 = (String) message1.getPayload(); + String s2 = (String) message2.getPayload(); + return s1.compareTo(s2); + } + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetrieverTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetrieverTests.java index 94ca5bb3e0..cb33a3561d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetrieverTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetrieverTests.java @@ -39,7 +39,7 @@ public class ChannelPollingMessageRetrieverTests { public void testSingleMessagePerRetrieval() { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setReceiveTimeout(0); - MessageChannel channel = new SimpleChannel(dispatcherPolicy); + MessageChannel channel = new SimpleChannel(5, dispatcherPolicy); ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); Collection> results = retriever.retrieveMessages(); assertTrue(results.isEmpty()); @@ -58,7 +58,7 @@ public class ChannelPollingMessageRetrieverTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(2); - MessageChannel channel = new SimpleChannel(dispatcherPolicy); + MessageChannel channel = new SimpleChannel(5, dispatcherPolicy); ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); Collection> results = retriever.retrieveMessages(); assertTrue(results.isEmpty()); @@ -80,7 +80,7 @@ public class ChannelPollingMessageRetrieverTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(1); - MessageChannel channel = new SimpleChannel(dispatcherPolicy); + MessageChannel channel = new SimpleChannel(5, dispatcherPolicy); ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); Collection> results = retriever.retrieveMessages(); assertTrue(results.isEmpty()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java index 5ce08027e8..a1c5a1de93 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java @@ -81,7 +81,7 @@ public class DefaultMessageDispatcherTests { final CountDownLatch latch = new CountDownLatch(2); MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true)); + SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(true)); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor())); @@ -125,7 +125,7 @@ public class DefaultMessageDispatcherTests { MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); - SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true)); + SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(true)); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); ConcurrentHandler inactiveHandler = new ConcurrentHandler(handler2, createExecutor()); @@ -157,7 +157,7 @@ public class DefaultMessageDispatcherTests { MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); - SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true)); + SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(true)); channel.getDispatcherPolicy().setRejectionLimit(2); channel.getDispatcherPolicy().setRetryInterval(3); channel.send(new StringMessage(1, "test")); @@ -188,7 +188,7 @@ public class DefaultMessageDispatcherTests { final CountDownLatch latch = new CountDownLatch(3); MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true)); + SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(true)); channel.getDispatcherPolicy().setRejectionLimit(2); channel.getDispatcherPolicy().setRetryInterval(3); channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false); @@ -213,7 +213,7 @@ public class DefaultMessageDispatcherTests { final CountDownLatch latch = new CountDownLatch(4); MessageHandler handler1 = TestHandlers.rejectingCountDownHandler(latch); MessageHandler handler2 = TestHandlers.rejectingCountDownHandler(latch); - SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(false)); + SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(false)); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); channel.getDispatcherPolicy().setRejectionLimit(2); @@ -287,7 +287,7 @@ public class DefaultMessageDispatcherTests { dispatcherPolicy.setRejectionLimit(2); dispatcherPolicy.setRetryInterval(3); dispatcherPolicy.setShouldFailOnRejectionLimit(false); - SimpleChannel channel = new SimpleChannel(dispatcherPolicy); + SimpleChannel channel = new SimpleChannel(25, dispatcherPolicy); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()) { @@ -341,7 +341,7 @@ public class DefaultMessageDispatcherTests { dispatcherPolicy.setRejectionLimit(5); dispatcherPolicy.setRetryInterval(3); dispatcherPolicy.setShouldFailOnRejectionLimit(false); - SimpleChannel channel = new SimpleChannel(dispatcherPolicy); + SimpleChannel channel = new SimpleChannel(25, dispatcherPolicy); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()) { @@ -451,7 +451,7 @@ public class DefaultMessageDispatcherTests { final CountDownLatch latch = new CountDownLatch(1); MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true)); + SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(true)); channel.send(new StringMessage(1, "test")); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler); DefaultMessageEndpoint endpoint1 = new DefaultMessageEndpoint(handler1);