You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2013/01/05 08:37:46 UTC
svn commit: r1429218 - in /camel/trunk/camel-core/src:
main/java/org/apache/camel/processor/RouteContextProcessor.java
test/java/org/apache/camel/processor/RouteContextProcessorTest.java
Author: davsclaus
Date: Sat Jan 5 07:37:45 2013
New Revision: 1429218
URL: http://svn.apache.org/viewvc?rev=1429218&view=rev
Log:
CAMEL-5925: Fixed potential NPE in RouteContextProcessor. Thanks to Taariq Levack for the patch.
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/RouteContextProcessorTest.java (with props)
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RouteContextProcessor.java
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RouteContextProcessor.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RouteContextProcessor.java?rev=1429218&r1=1429217&r2=1429218&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RouteContextProcessor.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/RouteContextProcessor.java Sat Jan 5 07:37:45 2013
@@ -27,7 +27,7 @@ import org.apache.camel.spi.UnitOfWork;
* This ensures that the {@link Exchange} have details under which route its being currently processed.
*/
public class RouteContextProcessor extends DelegateAsyncProcessor {
-
+
private final RouteContext routeContext;
public RouteContextProcessor(RouteContext routeContext, Processor processor) {
@@ -38,17 +38,17 @@ public class RouteContextProcessor exten
@Override
protected boolean processNext(final Exchange exchange, final AsyncCallback callback) {
// push the current route context
- if (exchange.getUnitOfWork() != null) {
- exchange.getUnitOfWork().pushRouteContext(routeContext);
+ final UnitOfWork unitOfWork = exchange.getUnitOfWork();
+ if (unitOfWork != null) {
+ unitOfWork.pushRouteContext(routeContext);
}
boolean sync = processor.process(exchange, new AsyncCallback() {
public void done(boolean doneSync) {
try {
- UnitOfWork uow = exchange.getUnitOfWork();
// pop the route context we just used
- if (uow != null) {
- uow.popRouteContext();
+ if (unitOfWork != null) {
+ unitOfWork.popRouteContext();
}
} catch (Exception e) {
exchange.setException(e);
Added: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/RouteContextProcessorTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/RouteContextProcessorTest.java?rev=1429218&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/RouteContextProcessorTest.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/RouteContextProcessorTest.java Sat Jan 5 07:37:45 2013
@@ -0,0 +1,98 @@
+/**
+ * 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.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (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
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.processor;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+
+/**
+ * @version
+ */
+public class RouteContextProcessorTest extends ContextTestSupport {
+
+ // Number of concurrent processing threads
+ public static final int CONCURRENCY = 10;
+
+ // Additional resequencer time-out above theoretical time-out
+ public static final long SAFETY_TIMEOUT = 100;
+
+ // Additional resequencer capacity above theoretical capacity
+ public static final int SAFETY_CAPACITY = 10;
+
+ // Resequencer time-out
+ public static final long TIMEOUT = SAFETY_TIMEOUT
+ + (RandomSleepProcessor.MAX_PROCESS_TIME - RandomSleepProcessor.MIN_PROCESS_TIME);
+
+ // Resequencer capacity
+ public static final int CAPACITY = SAFETY_CAPACITY
+ + (int) (CONCURRENCY * TIMEOUT / RandomSleepProcessor.MIN_PROCESS_TIME);
+
+ private static final int NUMBER_OF_MESSAGES = 10000;
+
+ public void testForkAndJoin() throws InterruptedException {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(NUMBER_OF_MESSAGES);
+
+ ProducerTemplate template = context.createProducerTemplate();
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
+ template.sendBodyAndHeader("seda:fork", "Test Message: " + i,
+ "seqnum", new Long(i));
+ }
+
+ long expectedTime = NUMBER_OF_MESSAGES
+ * (RandomSleepProcessor.MAX_PROCESS_TIME + RandomSleepProcessor.MIN_PROCESS_TIME)
+ / 2 / CONCURRENCY + TIMEOUT;
+ Thread.sleep(expectedTime);
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ Processor myProcessor = new RandomSleepProcessor();
+ from("seda:fork?concurrentConsumers=" + CONCURRENCY).process(
+ myProcessor).to("seda:join");
+ from("seda:join").resequence(header("seqnum")).stream()
+ .capacity(CAPACITY).timeout(TIMEOUT).to("mock:result");
+ }
+ };
+ }
+
+ /**
+ * Simulation processor that sleeps a random time between MIN_PROCESS_TIME
+ * and MAX_PROCESS_TIME milliseconds.
+ */
+ public static class RandomSleepProcessor implements Processor {
+ public static final long MIN_PROCESS_TIME = 5;
+ public static final long MAX_PROCESS_TIME = 50;
+
+ @Override
+ public void process(Exchange arg0) throws Exception {
+ long processTime = (long) (MIN_PROCESS_TIME + Math.random()
+ * (MAX_PROCESS_TIME - MIN_PROCESS_TIME));
+ Thread.sleep(processTime);
+ }
+ }
+}
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/RouteContextProcessorTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/RouteContextProcessorTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date