Author: davsclaus
Date: Sat Feb 28 16:26:00 2009
New Revision: 748883
URL: http://svn.apache.org/viewvc?rev=748883&view=rev
Log:
Better handling of interrupt while sleeping/loopingwhile consuming to better
support shutdown.
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerCommitRenameAbsolutePathStrategyTest.java
- copied, changed from r748781,
camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerCommitRenameStrategyTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DeadLetterChannelRedeliveringWhileShutdownTest.java
- copied, changed from r748781,
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DeadLetterChannelOnRedeliveryTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/ServiceStatus.java
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/strategy/FileLockExclusiveReadLockStrategy.java
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/strategy/GenericFileRenameExclusiveReadLockStrategy.java
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultExchange.java
camel/trunk/camel-core/src/main/java/org/apache/camel/model/RouteType.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DeadLetterChannel.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RedeliveryPolicy.java
camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DeadLetterChannelOnRedeliveryTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/RouteMEPOptionalOutTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/ServiceStatus.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/ServiceStatus.java?rev=748883&r1=748882&r2=748883&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/ServiceStatus.java
(original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/ServiceStatus.java
Sat Feb 28 16:26:00 2009
@@ -1,5 +1,4 @@
/**
- *
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
@@ -7,7 +6,7 @@
* (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
+ * 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,
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java?rev=748883&r1=748882&r2=748883&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
Sat Feb 28 16:26:00 2009
@@ -94,7 +94,8 @@
if (total > 0 && log.isDebugEnabled()) {
log.debug("Total " + total + " files to consume");
}
- for (int index = 0; index < total; index++) {
+ for (int index = 0; index < total && isRunAllowed(); index++) {
+ // only loop if we are started (allowed to run)
GenericFileExchange<T> exchange = exchanges.get(index);
// add current index and total as headers
exchange.getIn().setHeader(Exchange.FILE_BATCH_INDEX, index);
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/strategy/FileLockExclusiveReadLockStrategy.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/strategy/FileLockExclusiveReadLockStrategy.java?rev=748883&r1=748882&r2=748883&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/strategy/FileLockExclusiveReadLockStrategy.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/strategy/FileLockExclusiveReadLockStrategy.java
Sat Feb 28 16:26:00 2009
@@ -84,7 +84,11 @@
exclusive = true;
} else {
- sleep();
+ boolean interrupted = sleep();
+ if (interrupted) {
+ // we were interrputed while sleeping, we are likely
being shutdown so return false
+ return false;
+ }
}
}
} catch (IOException e) {
@@ -97,7 +101,11 @@
if (LOG.isDebugEnabled()) {
LOG.debug("Cannot acquire read lock. Will try again.", e);
}
- sleep();
+ boolean interrupted = sleep();
+ if (interrupted) {
+ // we were interrputed while sleeping, we are likely being
shutdown so return false
+ return false;
+ }
}
return true;
@@ -122,12 +130,14 @@
}
}
- private void sleep() {
+ private boolean sleep() {
LOG.trace("Exclusive read lock not granted. Sleeping for 1000
millis.");
try {
Thread.sleep(1000);
+ return true;
} catch (InterruptedException e) {
- // ignore
+ LOG.debug("Sleep interrupted while waiting for exclusive read
lock");
+ return false;
}
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/strategy/GenericFileRenameExclusiveReadLockStrategy.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/strategy/GenericFileRenameExclusiveReadLockStrategy.java?rev=748883&r1=748882&r2=748883&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/strategy/GenericFileRenameExclusiveReadLockStrategy.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/strategy/GenericFileRenameExclusiveReadLockStrategy.java
Sat Feb 28 16:26:00 2009
@@ -68,7 +68,11 @@
// rename it back so we can read it
operations.renameFile(newFile.getAbsoluteFileName(),
file.getAbsoluteFileName());
} else {
- sleep();
+ boolean interrupted = sleep();
+ if (interrupted) {
+ // we were interrputed while sleeping, we are likely being
shutdown so return false
+ return false;
+ }
}
}
@@ -80,12 +84,14 @@
// noop
}
- private void sleep() {
+ private boolean sleep() {
LOG.trace("Exclusive read lock not granted. Sleeping for 1000
millis.");
try {
Thread.sleep(1000);
+ return true;
} catch (InterruptedException e) {
- // ignore
+ LOG.debug("Sleep interrupted while waiting for exclusive read
lock");
+ return false;
}
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java?rev=748883&r1=748882&r2=748883&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
Sat Feb 28 16:26:00 2009
@@ -62,9 +62,7 @@
try {
exchange = queue.poll(1000, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
- if (LOG.isTraceEnabled()) {
- LOG.trace("Interupted: " + e, e);
- }
+ LOG.debug("Sleep interrupted, are we stopping? " +
(isStopping() || isStopped()));
continue;
}
if (exchange != null) {
@@ -82,9 +80,7 @@
try {
queue.put(exchange);
} catch (InterruptedException e) {
- if (LOG.isTraceEnabled()) {
- LOG.trace("Interupted: " + e, e);
- }
+ LOG.debug("Sleep interrupted, are we stopping? " +
(isStopping() || isStopped()));
}
}
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java?rev=748883&r1=748882&r2=748883&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
Sat Feb 28 16:26:00 2009
@@ -38,8 +38,8 @@
import org.apache.camel.Routes;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.Service;
-import org.apache.camel.TypeConverter;
import org.apache.camel.ServiceStatus;
+import org.apache.camel.TypeConverter;
import org.apache.camel.builder.ErrorHandlerBuilder;
import org.apache.camel.impl.converter.DefaultTypeConverter;
import org.apache.camel.management.InstrumentationLifecycleStrategy;
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultExchange.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultExchange.java?rev=748883&r1=748882&r2=748883&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultExchange.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultExchange.java
Sat Feb 28 16:26:00 2009
@@ -26,10 +26,9 @@
import org.apache.camel.ExchangeProperty;
import org.apache.camel.Message;
import org.apache.camel.RuntimeCamelException;
-import org.apache.camel.TypeConverter;
import org.apache.camel.spi.UnitOfWork;
-import org.apache.camel.util.UuidGenerator;
import org.apache.camel.util.ExchangeHelper;
+import org.apache.camel.util.UuidGenerator;
import static org.apache.camel.util.ObjectHelper.wrapRuntimeCamelException;
/**
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/model/RouteType.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/RouteType.java?rev=748883&r1=748882&r2=748883&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/model/RouteType.java
(original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/model/RouteType.java
Sat Feb 28 16:26:00 2009
@@ -94,8 +94,7 @@
ServiceStatus status = getStatus();
if (status == null) {
return true;
- }
- else {
+ } else {
return status.isStartable();
}
}
@@ -104,8 +103,7 @@
ServiceStatus status = getStatus();
if (status == null) {
return false;
- }
- else {
+ } else {
return status.isStoppable();
}
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DeadLetterChannel.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DeadLetterChannel.java?rev=748883&r1=748882&r2=748883&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DeadLetterChannel.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DeadLetterChannel.java
Sat Feb 28 16:26:00 2009
@@ -137,6 +137,9 @@
while (true) {
// we can't keep retrying if the route is being shutdown.
if (!isRunAllowed()) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Rejected execution as we are not started for
exchange: " + exchange);
+ }
if (exchange.getException() == null) {
exchange.setException(new RejectedExecutionException());
}
@@ -175,7 +178,13 @@
MessageHelper.resetStreamCache(exchange.getIn());
// wait until we should redeliver
- data.redeliveryDelay =
data.currentRedeliveryPolicy.sleep(data.redeliveryDelay,
data.redeliveryCounter);
+ try {
+ data.redeliveryDelay =
data.currentRedeliveryPolicy.sleep(data.redeliveryDelay,
data.redeliveryCounter);
+ } catch (InterruptedException e) {
+ LOG.debug("Sleep interrupted, are we stopping? " +
(isStopping() || isStopped()));
+ // continue from top
+ continue;
+ }
// letting onRedeliver be executed
deliverToRedeliveryProcessor(exchange, callback, data);
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RedeliveryPolicy.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RedeliveryPolicy.java?rev=748883&r1=748882&r2=748883&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RedeliveryPolicy.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RedeliveryPolicy.java
Sat Feb 28 16:26:00 2009
@@ -138,21 +138,16 @@
* @param redeliveryDelay previous redelivery delay
* @param redeliveryCounter number of previous redelivery attempts
* @return the calculate delay
+ * @throws InterruptedException is thrown if the sleep is interruped
likely because of shutdown
*/
- public long sleep(long redeliveryDelay, int redeliveryCounter) {
+ public long sleep(long redeliveryDelay, int redeliveryCounter) throws
InterruptedException {
redeliveryDelay = calculateRedeliveryDelay(redeliveryDelay,
redeliveryCounter);
if (redeliveryDelay > 0) {
if (LOG.isDebugEnabled()) {
LOG.debug("Sleeping for: " + redeliveryDelay + " millis until
attempting redelivery");
}
- try {
- Thread.sleep(redeliveryDelay);
- } catch (InterruptedException e) {
- if (LOG.isTraceEnabled()) {
- LOG.trace("Thread interrupted: " + e, e);
- }
- }
+ Thread.sleep(redeliveryDelay);
}
return redeliveryDelay;
}
@@ -202,8 +197,6 @@
for (String group : groups) {
long delay = Long.valueOf(ObjectHelper.after(group, ":"));
int count = Integer.valueOf(ObjectHelper.before(group, ":"));
-
-
if (count > redeliveryCounter) {
break;
} else {
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java?rev=748883&r1=748882&r2=748883&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
Sat Feb 28 16:26:00 2009
@@ -19,6 +19,7 @@
import java.util.HashMap;
import java.util.Map;
+import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
@@ -30,7 +31,6 @@
import org.apache.camel.NoSuchHeaderException;
import org.apache.camel.NoSuchPropertyException;
import org.apache.camel.NoTypeConversionAvailableException;
-import org.apache.camel.CamelContext;
import org.apache.camel.TypeConverter;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
Copied:
camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerCommitRenameAbsolutePathStrategyTest.java
(from r748781,
camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerCommitRenameStrategyTest.java)
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerCommitRenameAbsolutePathStrategyTest.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerCommitRenameAbsolutePathStrategyTest.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerCommitRenameStrategyTest.java&r1=748781&r2=748883&rev=748883&view=diff
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerCommitRenameStrategyTest.java
(original)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerCommitRenameAbsolutePathStrategyTest.java
Sat Feb 28 16:26:00 2009
@@ -17,27 +17,33 @@
package org.apache.camel.component.file;
import java.io.File;
-import java.io.FileWriter;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.converter.IOConverter;
/**
- * Unit test for the FileRenameStrategy using move options
+ * Unit test for the FileRenameStrategy using move options with absolute paths
*/
-public class FileConsumerCommitRenameStrategyTest extends ContextTestSupport {
+public class FileConsumerCommitRenameAbsolutePathStrategyTest extends
ContextTestSupport {
+
+ private String base;
@Override
protected void setUp() throws Exception {
deleteDirectory("target/done");
deleteDirectory("target/reports");
+ // use current dir as base as aboslute path
+ base = new File(".").getPath();
super.setUp();
}
- public void testRenameSuccess() throws Exception {
+ // TODO:
+ public void TestRenameSuccess() throws Exception {
+ }
+
+ public void xxxTestRenameSuccess() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:report");
mock.expectedBodiesReceived("Hello Paris");
mock.expectedFileExists("./target/done/paris.txt", "Hello Paris");
@@ -47,38 +53,11 @@
mock.assertIsSatisfied();
}
- public void testRenameFileExists() throws Exception {
- // create a file in done to let there be a duplicate file
- File file = new File("target/done");
- file.mkdirs();
- FileWriter fw = new FileWriter("./target/done/london.txt");
- try {
- fw.write("I was there once in London");
- fw.flush();
- } finally {
- fw.close();
- }
-
- MockEndpoint mock = getMockEndpoint("mock:report");
- mock.expectedBodiesReceived("Hello London");
-
- template.sendBodyAndHeader("file:target/reports", "Hello London",
Exchange.FILE_NAME, "london.txt");
-
- mock.assertIsSatisfied();
-
- // sleep to let the file consumer do its renaming
- Thread.sleep(100);
-
- // content of file should be Hello London
- String content = IOConverter.toString(new
File("./target/done/london.txt"));
- assertEquals("The file should have been renamed replacing any existing
files", "Hello London", content);
- }
-
protected RouteBuilder createRouteBuilder() throws Exception {
return new RouteBuilder() {
public void configure() throws Exception {
-
from("file://target/reports?moveExpression=../done/${file:name}&consumer.delay=5000").to("mock:report");
+ from("file://target/reports?moveExpression=" + base +
"/done/${file:name}&consumer.delay=5000").to("mock:report");
}
};
}
-}
+}
\ No newline at end of file
Modified:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DeadLetterChannelOnRedeliveryTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DeadLetterChannelOnRedeliveryTest.java?rev=748883&r1=748882&r2=748883&view=diff
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DeadLetterChannelOnRedeliveryTest.java
(original)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DeadLetterChannelOnRedeliveryTest.java
Sat Feb 28 16:26:00 2009
@@ -38,7 +38,7 @@
assertMockEndpointsSatisfied();
}
- public void xxxtestOnExceptionAlterMessageWithHeadersBeforeRedelivery()
throws Exception {
+ public void testOnExceptionAlterMessageWithHeadersBeforeRedelivery()
throws Exception {
MockEndpoint mock = getMockEndpoint("mock:result");
mock.expectedBodiesReceived("Hello World123");
mock.expectedHeaderReceived("foo", "123");
Copied:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DeadLetterChannelRedeliveringWhileShutdownTest.java
(from r748781,
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DeadLetterChannelOnRedeliveryTest.java)
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DeadLetterChannelRedeliveringWhileShutdownTest.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DeadLetterChannelRedeliveringWhileShutdownTest.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DeadLetterChannelOnRedeliveryTest.java&r1=748781&r2=748883&rev=748883&view=diff
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DeadLetterChannelOnRedeliveryTest.java
(original)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DeadLetterChannelRedeliveringWhileShutdownTest.java
Sat Feb 28 16:26:00 2009
@@ -23,35 +23,24 @@
import org.apache.camel.component.mock.MockEndpoint;
/**
- * Unit test for testing possibility to modify exchange before redelivering
+ * Unit test for shutting down whil DLC is sleeping in a redelivery.
*/
-public class DeadLetterChannelOnRedeliveryTest extends ContextTestSupport {
+public class DeadLetterChannelRedeliveringWhileShutdownTest extends
ContextTestSupport {
- static int counter;
+ public void testDLCRedelivery() throws Exception {
+ long start = System.currentTimeMillis();
- public void testOnExceptionAlterMessageBeforeRedelivery() throws Exception
{
MockEndpoint mock = getMockEndpoint("mock:result");
- mock.expectedBodiesReceived("Hello World123");
+ mock.expectedBodiesReceived("Hello World");
+ // send a message that causes redeliveries
+ template.sendBody("seda:damm", "Damm World");
template.sendBody("direct:start", "Hello World");
assertMockEndpointsSatisfied();
- }
-
- public void xxxtestOnExceptionAlterMessageWithHeadersBeforeRedelivery()
throws Exception {
- MockEndpoint mock = getMockEndpoint("mock:result");
- mock.expectedBodiesReceived("Hello World123");
- mock.expectedHeaderReceived("foo", "123");
-
- template.sendBodyAndHeader("direct:start", "Hello World", "foo",
"123");
-
- assertMockEndpointsSatisfied();
- }
- @Override
- protected void setUp() throws Exception {
- super.setUp();
- counter = 0;
+ long delta = System.currentTimeMillis() - start;
+ assertTrue("Should be faster than: " + delta, delta < 2500);
}
@Override
@@ -59,47 +48,16 @@
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- // START SNIPPET: e1
- // we configure our Dead Letter Channel to invoke
- // MyRedeliveryProcessor before a redelivery is
- // attempted. This allows us to alter the message before
- errorHandler(deadLetterChannel("mock:error")
- .onRedelivery(new MyRedeliverPrcessor())
- // setting delay to zero is just to make unit teting
faster
- .delay(0L));
- // END SNIPPET: e1
+
errorHandler(deadLetterChannel("mock:error").maximumRedeliveries(2).delay(5000));
+ from("direct:start").delay(500).to("mock:result");
- from("direct:start").process(new Processor() {
+ from("seda:damm").process(new Processor() {
public void process(Exchange exchange) throws Exception {
- // force some error so Camel will do redelivery
- if (++counter <= 3) {
- throw new IllegalArgumentException("Forced by unit
test");
- }
+ throw new IllegalArgumentException("Damm");
}
- }).to("mock:result");
-
+ });
}
};
}
-
- // START SNIPPET: e2
- // This is our processor that is executed before every redelivery attempt
- // here we can do what we want in the java code, such as altering the
message
- public class MyRedeliverPrcessor implements Processor {
-
- public void process(Exchange exchange) throws Exception {
- // the message is being redelivered so we can alter it
-
- // we just append the redelivery counter to the body
- // you can of course do all kind of stuff instead
- String body = exchange.getIn().getBody(String.class);
- int count = exchange.getIn().getHeader("CamelRedeliveryCounter",
Integer.class);
-
- exchange.getIn().setBody(body + count);
- }
- }
- // END SNIPPET: e2
-
-
}
\ No newline at end of file
Modified:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/RouteMEPOptionalOutTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/RouteMEPOptionalOutTest.java?rev=748883&r1=748882&r2=748883&view=diff
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/RouteMEPOptionalOutTest.java
(original)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/RouteMEPOptionalOutTest.java
Sat Feb 28 16:26:00 2009
@@ -17,9 +17,9 @@
package org.apache.camel.processor;
import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.Processor;
-import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
/**
