Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Added PriorityChannel (INT-115).

Browse Source
This commit is contained in:
Mark Fisher
2008-02-26 04:11:10 +00:00
parent 51d02ee8ad
commit 82cb7bcb62
10 changed files with 349 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java
View File

@@ -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);
/**

103
spring-integration-core/src/main/java/org/springframework/integration/channel/BaseBlockingQueueChannel.java Normal file
View File

@@ -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<Message<?>> queue;
/**
* Create a channel with the specified queue and dispatcher policy.
*/
public BaseBlockingQueueChannel(BlockingQueue<Message<?>> 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<Message<?>> clear() {
List<Message<?>> clearedMessages = new ArrayList<Message<?>>();
this.queue.drainTo(clearedMessages);
return clearedMessages;
}
public List<Message<?>> purge(MessageSelector selector) {
if (selector == null) {
return this.clear();
}
List<Message<?>> purgedMessages = new ArrayList<Message<?>>();
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;
}
}

3
spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java
View File

@@ -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.
*/

106
spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java Normal file
View File

@@ -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<Message<?>> comparator) {
super(new PriorityBlockingQueue<Message<?>>(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<Message<?>> {
public int compare(Message<?> message1, Message<?> message2) {
Integer priority1 = message1.getHeader().getPriority();
Integer priority2 = message2.getHeader().getPriority();
return priority1.compareTo(priority2);
}
}
}

86
spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java
View File

@@ -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<Message<?>> 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<Message<?>>(capacity);
}
else {
this.queue = new SynchronousQueue<Message<?>>(true);
}
super((capacity > 0) ? new LinkedBlockingQueue<Message<?>>(capacity) :
new SynchronousQueue<Message<?>>(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<Message<?>> clear() {
List<Message<?>> clearedMessages = new ArrayList<Message<?>>();
this.queue.drainTo(clearedMessages);
return clearedMessages;
}
public List<Message<?>> purge(MessageSelector selector) {
if (selector == null) {
return this.clear();
}
List<Message<?>> purgedMessages = new ArrayList<Message<?>>();
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;
}
}

10
spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java
View File

@@ -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<String, Object> attributes = new ConcurrentHashMap<String, Object>();
@@ -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);
}

12
spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java
View File

@@ -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));

103
spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.java Normal file
View File

@@ -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<Message<?>> {
public int compare(Message<?> message1, Message<?> message2) {
String s1 = (String) message1.getPayload();
String s2 = (String) message2.getPayload();
return s1.compareTo(s2);
}
}
}

6
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetrieverTests.java
View File

@@ -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<Message<?>> 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<Message<?>> 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<Message<?>> results = retriever.retrieveMessages();
assertTrue(results.isEmpty());

16
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java
View File

@@ -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);