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

INT-2031, INT-2112 Added support to fall back on FIFO semantics to the PriorityChannel if PRIORITY header is not provided, changed type of PRIORITY header to be Long instead of Integer

Browse Source 
INT-2031, INT-2112 fixed one more broken test

INT-2031 changed the priority type back to Integer

INT-2031 added SequenceFallbackComparator

INT-2031 changed the name of the comparator from delegatingComparator to targetComparator
This commit is contained in:
Oleg Zhurakousky
2011-09-07 14:33:49 -04:00
committed by Mark Fisher
parent 90061778a4
commit 2930ad0065
3 changed files with 150 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java
View File

@@ -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<Message<?>> comparator) {
super(new PriorityBlockingQueue<Message<?>>(11,
(comparator != null) ? comparator : new MessagePriorityComparator()));
super(new PriorityBlockingQueue<Message<?>>(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<Message<?>> {
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<Message<?>> {
private final Comparator<Message<?>> targetComparator;
public SequenceFallbackComparator(Comparator<Message<?>> 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;
}
}

3
spring-integration-core/src/main/java/org/springframework/integration/config/xml/StandardHeaderEnricherParser.java
View File

@@ -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 {

100
spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.java
View File

@@ -50,6 +50,17 @@ public class PriorityChannelTests {
channel.receive(0);
assertTrue(channel.send(new GenericMessage<String>("test5")));
}
@Test
public void testDefaultComparatorWithTimestampFallback() throws Exception{
PriorityChannel channel = new PriorityChannel();
for (int i = 0; i < 1000; i++) {
channel.send(new GenericMessage<Integer>(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<Message<?>> {
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);
}
}
}