You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by wt...@apache.org on 2009/04/16 21:57:45 UTC
svn commit: r765731 - in /camel/branches/camel-1.x: ./
camel-core/src/main/java/org/apache/camel/processor/StreamResequencer.java
camel-core/src/test/java/org/apache/camel/processor/StreamResequencerTest.java
Author: wtam
Date: Thu Apr 16 19:57:45 2009
New Revision: 765731
URL: http://svn.apache.org/viewvc?rev=765731&view=rev
Log:
Merged revisions 765729 via svnmerge from
https://svn.apache.org/repos/asf/camel/trunk
........
r765729 | wtam | 2009-04-16 15:44:16 -0400 (Thu, 16 Apr 2009) | 1 line
[CAMEL-1510] BatchProcessor interrupt has side effects (submitted on behalf of Martin Krasser)
........
Modified:
camel/branches/camel-1.x/ (props changed)
camel/branches/camel-1.x/camel-core/src/main/java/org/apache/camel/processor/StreamResequencer.java
camel/branches/camel-1.x/camel-core/src/test/java/org/apache/camel/processor/StreamResequencerTest.java
Propchange: camel/branches/camel-1.x/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Thu Apr 16 19:57:45 2009
@@ -1 +1 @@
-/camel/trunk:736980,739733,739904,740251,740295,740306,740596,740663,741848,742231,742705,742739,742854,742856,742898,742906,743613,743762,743773,743920,743959-743960,744123,745105,745367,745541,745751,745826,745978,746269,746872,746895,746962,747258,747678-747704,748392,748436,748821,749563-749564,749574,749628-749629,749936,749956,750017,750334,750396,750761,750796,752068,752117,752418,752751-752755,752764-752773,752956,753087,753101,753175,755136,755487,756313,756348,756870,756939,757636,757693,757743,757865,758539,758563,758600,758617,758692,758990,759362,759453,759887,759931,760003,760890,760909,760937,761194,761536,761583,761607,762047,762633,762650,762935,763095,763484,763551,765154,765686
+/camel/trunk:736980,739733,739904,740251,740295,740306,740596,740663,741848,742231,742705,742739,742854,742856,742898,742906,743613,743762,743773,743920,743959-743960,744123,745105,745367,745541,745751,745826,745978,746269,746872,746895,746962,747258,747678-747704,748392,748436,748821,749563-749564,749574,749628-749629,749936,749956,750017,750334,750396,750761,750796,752068,752117,752418,752751-752755,752764-752773,752956,753087,753101,753175,755136,755487,756313,756348,756870,756939,757636,757693,757743,757865,758539,758563,758600,758617,758692,758990,759362,759453,759887,759931,760003,760890,760909,760937,761194,761536,761583,761607,762047,762633,762650,762935,763095,763484,763551,765154,765686,765729
Propchange: camel/branches/camel-1.x/
------------------------------------------------------------------------------
Binary property 'svnmerge-integrated' - no diff available.
Modified: camel/branches/camel-1.x/camel-core/src/main/java/org/apache/camel/processor/StreamResequencer.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-1.x/camel-core/src/main/java/org/apache/camel/processor/StreamResequencer.java?rev=765731&r1=765730&r2=765731&view=diff
==============================================================================
--- camel/branches/camel-1.x/camel-core/src/main/java/org/apache/camel/processor/StreamResequencer.java (original)
+++ camel/branches/camel-1.x/camel-core/src/main/java/org/apache/camel/processor/StreamResequencer.java Thu Apr 16 19:57:45 2009
@@ -16,6 +16,11 @@
*/
package org.apache.camel.processor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.impl.LoggingExceptionHandler;
@@ -170,7 +175,8 @@
private class Delivery extends Thread {
- private volatile boolean cancelRequested;
+ private Lock deliveryRequestLock = new ReentrantLock();
+ private Condition deliveryRequestCondition = deliveryRequestLock.newCondition();
public Delivery() {
super("Delivery Thread");
@@ -180,11 +186,14 @@
public void run() {
while (true) {
try {
- Thread.sleep(DELIVERY_ATTEMPT_INTERVAL);
- } catch (InterruptedException e) {
- if (cancelRequested) {
- return;
+ deliveryRequestLock.lock();
+ try {
+ deliveryRequestCondition.await(DELIVERY_ATTEMPT_INTERVAL, TimeUnit.MILLISECONDS);
+ } finally {
+ deliveryRequestLock.unlock();
}
+ } catch (InterruptedException e) {
+ break;
}
try {
engine.deliver();
@@ -195,12 +204,16 @@
}
public void cancel() {
- cancelRequested = true;
interrupt();
}
public void request() {
- interrupt();
+ deliveryRequestLock.lock();
+ try {
+ deliveryRequestCondition.signal();
+ } finally {
+ deliveryRequestLock.unlock();
+ }
}
}
Modified: camel/branches/camel-1.x/camel-core/src/test/java/org/apache/camel/processor/StreamResequencerTest.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-1.x/camel-core/src/test/java/org/apache/camel/processor/StreamResequencerTest.java?rev=765731&r1=765730&r2=765731&view=diff
==============================================================================
--- camel/branches/camel-1.x/camel-core/src/test/java/org/apache/camel/processor/StreamResequencerTest.java (original)
+++ camel/branches/camel-1.x/camel-core/src/test/java/org/apache/camel/processor/StreamResequencerTest.java Thu Apr 16 19:57:45 2009
@@ -22,6 +22,7 @@
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
+import org.apache.camel.ProducerTemplate;
import org.apache.camel.Route;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
@@ -54,6 +55,22 @@
resultEndpoint.assertIsSatisfied();
}
+ public void testMultithreaded() throws Exception {
+ int numMessages = 500;
+ Thread t1 = new Sender(context.createProducerTemplate(), 0, numMessages, 2);
+ Thread t2 = new Sender(context.createProducerTemplate(), 1, numMessages + 1, 2);
+ Object[] bodies = new Object[numMessages];
+ for (int i = 0; i < numMessages; i++) {
+ bodies[i] = "msg" + i;
+ }
+ resultEndpoint.expectedBodiesReceived(bodies);
+ t1.start();
+ t2.start();
+ t1.join();
+ t2.join();
+ resultEndpoint.assertIsSatisfied();
+ }
+
@Override
protected void setUp() throws Exception {
super.setUp();
@@ -111,5 +128,34 @@
assertIsInstanceOf(StreamResequencer.class, outputProcessor);
}
+
+ private static class Sender extends Thread {
+
+ ProducerTemplate template;
+
+ int start;
+ int end;
+ int increment;
+
+ public Sender(ProducerTemplate template, int start, int end, int increment) {
+ this.template = template;
+ this.start = start;
+ this.end = end;
+ this.increment = increment;
+ }
+
+ @Override
+ public void run() {
+ for (int i = start; i < end; i += increment) {
+ try {
+ Thread.sleep(2);
+ } catch (InterruptedException e) {
+ // ignore
+ }
+ template.sendBodyAndHeader("direct:start", "msg" + i, "seqnum", Long.valueOf(i));
+ }
+ }
+
+ }
}