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 index 490475de6c..950beae69f 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2011 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. @@ -17,8 +17,11 @@ package org.springframework.integration.channel; import java.util.Comparator; +import java.util.Map; import java.util.concurrent.PriorityBlockingQueue; +import java.util.concurrent.atomic.AtomicLong; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.integration.Message; import org.springframework.integration.MessageHeaders; import org.springframework.integration.util.UpperBound; @@ -28,10 +31,15 @@ import org.springframework.integration.util.UpperBound; * The default comparator is based upon the message header's 'priority'. * * @author Mark Fisher + * @author Oleg Zhurakousky */ public class PriorityChannel extends QueueChannel { private final UpperBound upperBound; + + private final AtomicLong sequenceCounter = new AtomicLong(); + + private static final String SEQUENCE_HEADER_NAME = "__priorityChannelSequence__"; /** @@ -42,8 +50,7 @@ public class PriorityChannel extends QueueChannel { * {@link MessageHeaders#getPriority()}. */ public PriorityChannel(int capacity, Comparator> comparator) { - super(new PriorityBlockingQueue>(11, - (comparator != null) ? comparator : new MessagePriorityComparator())); + super(new PriorityBlockingQueue>(11, new SequenceFallbackComparator(comparator))); this.upperBound = new UpperBound(capacity); } @@ -74,32 +81,58 @@ public class PriorityChannel extends QueueChannel { } + @SuppressWarnings({ "rawtypes", "unchecked" }) @Override protected boolean doSend(Message message, long timeout) { if (!upperBound.tryAcquire(timeout)) { return false; } + Map innerMap = (Map) new DirectFieldAccessor(message.getHeaders()).getPropertyValue("headers"); + innerMap.put(SEQUENCE_HEADER_NAME, sequenceCounter.incrementAndGet()); return super.doSend(message, 0); } + @SuppressWarnings({ "rawtypes"}) @Override protected Message doReceive(long timeout) { Message message = super.doReceive(timeout); - if (message != null) { - upperBound.release(); - return message; - } - return null; - } - private static class MessagePriorityComparator implements Comparator> { + if (message != null) { + Map innerMap = (Map) new DirectFieldAccessor(message.getHeaders()).getPropertyValue("headers"); + innerMap.remove(SEQUENCE_HEADER_NAME); + upperBound.release(); + } + return message; + } + + private static class SequenceFallbackComparator implements Comparator> { + + private final Comparator> targetComparator; + + public SequenceFallbackComparator(Comparator> targetComparator){ + this.targetComparator = targetComparator; + } public int compare(Message message1, Message message2) { - Integer priority1 = message1.getHeaders().getPriority(); - Integer priority2 = message2.getHeaders().getPriority(); - priority1 = priority1 != null ? priority1 : 0; - priority2 = priority2 != null ? priority2 : 0; - return priority2.compareTo(priority1); + int compareResult = 0; + if (this.targetComparator != null){ + compareResult = this.targetComparator.compare(message1, message2); + } + else { + Integer priority1 = message1.getHeaders().getPriority(); + Integer priority2 = message2.getHeaders().getPriority(); + + priority1 = priority1 != null ? priority1 : 0; + priority2 = priority2 != null ? priority2 : 0; + compareResult = priority2.compareTo(priority1); + } + + if (compareResult == 0){ + Long sequence1 = (Long) message1.getHeaders().get(SEQUENCE_HEADER_NAME); + Long sequence2 = (Long) message2.getHeaders().get(SEQUENCE_HEADER_NAME); + compareResult = sequence1.compareTo(sequence2); + } + return compareResult; } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/StandardHeaderEnricherParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/StandardHeaderEnricherParser.java index 277db8ea94..a87265bdf4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/StandardHeaderEnricherParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/StandardHeaderEnricherParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2011 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. @@ -31,6 +31,7 @@ import org.springframework.util.StringUtils; * references) if provided as 'header' sub-elements. * * @author Mark Fisher + * @author Oleg Zhurakousky */ public class StandardHeaderEnricherParser extends HeaderEnricherParserSupport { 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 index 929ca1b356..9372eeca5b 100644 --- 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 @@ -50,6 +50,17 @@ public class PriorityChannelTests { channel.receive(0); assertTrue(channel.send(new GenericMessage("test5"))); } + + @Test + public void testDefaultComparatorWithTimestampFallback() throws Exception{ + PriorityChannel channel = new PriorityChannel(); + for (int i = 0; i < 1000; i++) { + channel.send(new GenericMessage(i)); + } + for (int i = 0; i < 1000; i++) { + assertEquals(i, channel.receive().getPayload()); + } + } @Test public void testDefaultComparator() { @@ -90,6 +101,85 @@ public class PriorityChannelTests { assertEquals("D", channel.receive(0).getPayload()); assertEquals("E", channel.receive(0).getPayload()); } + + @Test + public void testWithCustomComparatorAndSequence() { + PriorityChannel channel = new PriorityChannel(10, new FooHeaderComparator()); + Message message1 = MessageBuilder.withPayload(1).setHeader("foo", 1).build(); + Message message2 = MessageBuilder.withPayload(2).setHeader("foo", 1).build(); + Message message3 = MessageBuilder.withPayload(3).setHeader("foo", 1).build(); + Message message4 = MessageBuilder.withPayload(4).build(); + Message message5 = MessageBuilder.withPayload(5).setHeader("foo", 3).build(); + + Message message6 = MessageBuilder.withPayload(6).setHeader("foo", 3).build(); + Message message7 = MessageBuilder.withPayload(7).setHeader("foo", 4).build(); + Message message8 = MessageBuilder.withPayload(8).setHeader("foo", 4).build(); + + + channel.send(message1); + channel.send(message2); + channel.send(message3); + channel.send(message4); + channel.send(message5); + channel.send(message6); + channel.send(message7); + channel.send(message8); + + Object receivedOne = channel.receive(0).getPayload(); + Object receivedTwo = channel.receive(0).getPayload(); + Object receivedThree = channel.receive(0).getPayload(); + Object receivedFour = channel.receive(0).getPayload(); + Object receivedFive = channel.receive(0).getPayload(); + Object receivedSix = channel.receive(0).getPayload(); + Object receivedSeven = channel.receive(0).getPayload(); + Object receivedEight = channel.receive(0).getPayload(); + + assertEquals(7, receivedOne); + assertEquals(8, receivedTwo); + assertEquals(5, receivedThree); + assertEquals(6, receivedFour); + assertEquals(1, receivedFive); + assertEquals(2, receivedSix); + assertEquals(3, receivedSeven); + assertEquals(4, receivedEight); + } + + @Test + public void testWithDefaultComparatorAndSequence() { + PriorityChannel channel = new PriorityChannel(); + Message message1 = MessageBuilder.withPayload(1).setPriority(1).build(); + Message message2 = MessageBuilder.withPayload(2).setPriority(1).build(); + Message message3 = MessageBuilder.withPayload(3).setPriority(1).build(); + Message message4 = MessageBuilder.withPayload(4).setPriority(2).build(); + Message message5 = MessageBuilder.withPayload(5).setPriority(2).build(); + + Message message6 = MessageBuilder.withPayload(6).build(); + Message message7 = MessageBuilder.withPayload(7).build(); + + channel.send(message1); + channel.send(message2); + channel.send(message3); + channel.send(message4); + channel.send(message5); + channel.send(message6); + channel.send(message7); + + Object receivedOne = channel.receive(0).getPayload(); + Object receivedTwo = channel.receive(0).getPayload(); + Object receivedThree = channel.receive(0).getPayload(); + Object receivedFour = channel.receive(0).getPayload(); + Object receivedFive = channel.receive(0).getPayload(); + Object receivedSix = channel.receive(0).getPayload(); + Object receivedSeven = channel.receive(0).getPayload(); + + assertEquals(4, receivedOne); + assertEquals(5, receivedTwo); + assertEquals(1, receivedThree); + assertEquals(2, receivedFour); + assertEquals(3, receivedFive); + assertEquals(6, receivedSix); + assertEquals(7, receivedSeven); + } @Test public void testNullPriorityIsConsideredNormal() { @@ -206,5 +296,15 @@ public class PriorityChannelTests { return s1.compareTo(s2); } } + + public static class FooHeaderComparator implements Comparator> { + public int compare(Message message1, Message message2) { + Integer foo1 = (Integer) message1.getHeaders().get("foo"); + Integer foo2 = (Integer) message2.getHeaders().get("foo"); + foo1 = foo1 != null ? foo1 : 0; + foo2 = foo2 != null ? foo2 : 0; + return foo2.compareTo(foo1); + } + } }