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

INT-2508 Priority Channel FIFO

Browse Source 
Fix a concurrency issue with the PriorityChannel by
introducing a private MessageWrapper; an implementation
of Message, to maintain the contract with the Comparator.

The MessageWrapper contains the original Message plus
an incremented sequence number, used as a tie-breaker,
to maintain FIFO semantics within priority.

Previously, the sequence number was added to, and removed
from, the headers of the original Message using a
DirectFieldAccessor.

This could cause concurrent modification
exceptions when it was removed.

INT-2508 polishing based on PR comments

INT-2508 Polish Whitespace
This commit is contained in:
Oleg Zhurakousky
2012-05-08 15:05:01 -04:00
committed by Gary Russell
parent 9235efe16b
commit 7d0dcf0805
2 changed files with 59 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2012 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.
@@ -16,12 +16,6 @@
package org.springframework.integration.channel;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.Comparator;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
@@ -35,6 +29,12 @@ import org.springframework.integration.Message;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.support.MessageBuilder;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
/**
* @author Mark Fisher
*/
@@ -82,6 +82,27 @@ public class PriorityChannelTests {
assertEquals("test:-99", channel.receive(0).getPayload());
}
// although this test has no assertions it results in ConcurrentModificationException
// if executed before changes for INT-2508
@Test
public void testPriorityChannelWithConcurrentModification() throws Exception{
final PriorityChannel channel = new PriorityChannel();
final Message<String> message = new GenericMessage<String>("hello");
for (int i = 0; i < 1000; i++) {
channel.send(message);
new Thread(new Runnable() {
public void run() {
channel.receive();
}
}).start();
new Thread(new Runnable() {
public void run() {
message.getHeaders().toString();
}
}).start();
}
}
@Test
public void testCustomComparator() {
PriorityChannel channel = new PriorityChannel(5, new StringPayloadComparator());