svn commit: r1202148 - in /camel/trunk/camel-core/src: main/java/org/apache/camel/component/seda/SedaConsumer.java test/java/org/apache/camel/component/seda/SedaConcurrentConsumersNPEIssueTest.java

[da...@apache.org]

Author: davsclaus
Date: Tue Nov 15 12:01:24 2011
New Revision: 1202148

URL: http://svn.apache.org/viewvc?rev=1202148&view=rev
Log:
CAMEL-4650: Fixed NPE in seda consumer and check for multiple consumers supported if multicasting.

Added:
    camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaConcurrentConsumersNPEIssueTest.java
      - copied, changed from r1202071, camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaConcurrentConsumersTest.java
Modified:
    camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java

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=1202148&r1=1202147&r2=1202148&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 Tue Nov 15 12:01:24 2011
@@ -37,6 +37,7 @@ import org.apache.camel.spi.ShutdownAwar
 import org.apache.camel.support.ServiceSupport;
 import org.apache.camel.util.AsyncProcessorConverterHelper;
 import org.apache.camel.util.AsyncProcessorHelper;
+import org.apache.camel.util.ObjectHelper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -199,12 +200,18 @@ public class SedaConsumer extends Servic
         // if there are multiple consumers then multicast to them
         if (size > 1) {
 
+            // validate multiple consumers has been enabled
+            if (!endpoint.isMultipleConsumersSupported()) {
+                throw new IllegalStateException("Multiple consumers for the same endpoint is not allowed: " + endpoint);
+            }
+
             if (LOG.isDebugEnabled()) {
                 LOG.debug("Multicasting to {} consumers for Exchange: {}", endpoint.getConsumers().size(), exchange);
             }
            
             // use a multicast processor to process it
             MulticastProcessor mp = endpoint.getConsumerMulticastProcessor();
+            ObjectHelper.notNull(mp, "ConsumerMulticastProcessor", this);
 
             // and use the asynchronous routing engine to support it
             AsyncProcessorHelper.process(mp, exchange, new AsyncCallback() {

Copied: camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaConcurrentConsumersNPEIssueTest.java (from r1202071, camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaConcurrentConsumersTest.java)
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaConcurrentConsumersNPEIssueTest.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaConcurrentConsumersNPEIssueTest.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaConcurrentConsumersTest.java&r1=1202071&r2=1202148&rev=1202148&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaConcurrentConsumersTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/component/seda/SedaConcurrentConsumersNPEIssueTest.java Tue Nov 15 12:01:24 2011
@@ -21,15 +21,24 @@ import org.apache.camel.builder.RouteBui
 import org.apache.camel.component.mock.MockEndpoint;
 
 /**
- * @version 
+ * @version
  */
-public class SedaConcurrentConsumersTest extends ContextTestSupport {
+public class SedaConcurrentConsumersNPEIssueTest extends ContextTestSupport {
 
     public void testSendToSeda() throws Exception {
         MockEndpoint mock = getMockEndpoint("mock:result");
         mock.expectedBodiesReceived("Hello World");
 
-        template.sendBody("seda:foo?concurrentConsumers=5", "Hello World");
+        template.sendBody("seda:foo", "Hello World");
+
+        assertMockEndpointsSatisfied();
+
+        // now start the first route, which should cause a failure due multiple consumers is not allowed
+        mock.reset();
+        mock.expectedMessageCount(0);
+
+        context.startRoute("first");
+        template.sendBody("seda:foo", "Bye World");
 
         assertMockEndpointsSatisfied();
     }
@@ -39,7 +48,9 @@ public class SedaConcurrentConsumersTest
         return new RouteBuilder() {
             @Override
             public void configure() throws Exception {
-                from("seda:foo?concurrentConsumers=5").to("mock:result");
+                from("seda:foo?concurrentConsumers=5").routeId("first").noAutoStartup().to("mock:result");
+
+                from("seda:foo?concurrentConsumers=5").routeId("second").to("mock:result");
             }
         };
     }