camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bo...@apache.org
Subject svn commit: r1358452 - in /camel/trunk: camel-core/src/main/java/org/apache/camel/model/ camel-core/src/main/java/org/apache/camel/processor/ camel-core/src/test/java/org/apache/camel/processor/enricher/ components/camel-spring/src/test/resources/org/a...
Date Fri, 06 Jul 2012 22:13:52 GMT
Author: boday
Date: Fri Jul  6 22:13:51 2012
New Revision: 1358452

URL: http://svn.apache.org/viewvc?rev=1358452&view=rev
Log:
CAMEL-3211 added basic "pollMultiple" support to the pollEnrich EIP

Modified:
    camel/trunk/camel-core/src/main/java/org/apache/camel/model/PollEnrichDefinition.java
    camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java
    camel/trunk/camel-core/src/main/java/org/apache/camel/processor/PollEnricher.java
    camel/trunk/camel-core/src/test/java/org/apache/camel/processor/enricher/PollEnricherTest.java
    camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/pollEnricher.xml
    camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/pollEnricherRef.xml

Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/model/PollEnrichDefinition.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/PollEnrichDefinition.java?rev=1358452&r1=1358451&r2=1358452&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/model/PollEnrichDefinition.java
(original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/model/PollEnrichDefinition.java
Fri Jul  6 22:13:51 2012
@@ -48,14 +48,17 @@ public class PollEnrichDefinition extend
     private String aggregationStrategyRef;
     @XmlTransient
     private AggregationStrategy aggregationStrategy;
+    @XmlAttribute
+    private Boolean pollMultiple;
 
     public PollEnrichDefinition() {
     }
 
-    public PollEnrichDefinition(AggregationStrategy aggregationStrategy, String resourceUri,
long timeout) {
+    public PollEnrichDefinition(AggregationStrategy aggregationStrategy, String resourceUri,
long timeout, Boolean pollMultiple) {
         this.aggregationStrategy = aggregationStrategy;
         this.resourceUri = resourceUri;
         this.timeout = timeout;
+        this.pollMultiple = pollMultiple;
     }
 
     @Override
@@ -93,10 +96,10 @@ public class PollEnrichDefinition extend
 
         PollEnricher enricher;
         if (timeout != null) {
-            enricher = new PollEnricher(null, endpoint.createPollingConsumer(), timeout);
+            enricher = new PollEnricher(null, endpoint.createPollingConsumer(), timeout,
pollMultiple);
         } else {
             // if no timeout then we should block, and there use a negative timeout
-            enricher = new PollEnricher(null, endpoint.createPollingConsumer(), -1);
+            enricher = new PollEnricher(null, endpoint.createPollingConsumer(), -1, pollMultiple);
         }
 
         if (aggregationStrategyRef != null) {
@@ -150,4 +153,12 @@ public class PollEnrichDefinition extend
     public void setAggregationStrategy(AggregationStrategy aggregationStrategy) {
         this.aggregationStrategy = aggregationStrategy;
     }
+
+    public Boolean isPollMultiple() {
+        return pollMultiple;
+    }
+
+    public void setPollMultiple(Boolean pollMultiple) {
+        this.pollMultiple = pollMultiple;
+    }
 }
\ No newline at end of file

Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java?rev=1358452&r1=1358451&r2=1358452&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java Fri
Jul  6 22:13:51 2012
@@ -2823,7 +2823,52 @@ public abstract class ProcessorDefinitio
      */
     @SuppressWarnings("unchecked")
     public Type pollEnrich(String resourceUri) {
-        addOutput(new PollEnrichDefinition(null, resourceUri, -1));
+        addOutput(new PollEnrichDefinition(null, resourceUri, -1, false));
+        return (Type) this;
+    }
+
+    /**
+     * The <a href="http://camel.apache.org/content-enricher.html">Content Enricher
EIP</a>
+     * enriches an exchange with additional data obtained from a <code>resourceUri</code>
+     * using a {@link org.apache.camel.PollingConsumer} to poll the endpoint.
+     * <p/>
+     * The difference between this and {@link #enrich(String)} is that this uses a consumer
+     * to obtain the additional data, where as enrich uses a producer.
+     * <p/>
+     * This method will not wait for data to become available, use the method with an explicit
timeout
+     * if you want to wait for data for a period of time from the resourceUri.
+     *
+     * @param resourceUri           URI of resource endpoint for obtaining additional data.
+     * @param pollMultiple           if enabled will poll for all Exchanges available on
the endpoint
+     * @return the builder
+     * @see org.apache.camel.processor.PollEnricher
+     */
+    @SuppressWarnings("unchecked")
+    public Type pollEnrich(String resourceUri, boolean pollMultiple) {
+        addOutput(new PollEnrichDefinition(null, resourceUri, 0, pollMultiple));
+        return (Type) this;
+    }
+
+    /**
+     * The <a href="http://camel.apache.org/content-enricher.html">Content Enricher
EIP</a>
+     * enriches an exchange with additional data obtained from a <code>resourceUri</code>
+     * using a {@link org.apache.camel.PollingConsumer} to poll the endpoint.
+     * <p/>
+     * The difference between this and {@link #enrich(String)} is that this uses a consumer
+     * to obtain the additional data, where as enrich uses a producer.
+     * <p/>
+     * This method will <tt>block</tt> until data is available, use the method
with timeout if you do not
+     * want to risk waiting a long time before data is available from the resourceUri.
+     *
+     * @param resourceUri           URI of resource endpoint for obtaining additional data.
+     * @param timeout               timeout in millis to wait at most for data to be available.
+     * @param pollMultiple           if enabled will poll for all Exchanges available on
the endpoint
+     * @return the builder
+     * @see org.apache.camel.processor.PollEnricher
+     */
+    @SuppressWarnings("unchecked")
+    public Type pollEnrich(String resourceUri, long timeout, boolean pollMultiple) {
+        addOutput(new PollEnrichDefinition(null, resourceUri, timeout, pollMultiple));
         return (Type) this;
     }
 
@@ -2845,7 +2890,7 @@ public abstract class ProcessorDefinitio
      */
     @SuppressWarnings("unchecked")
     public Type pollEnrich(String resourceUri, AggregationStrategy aggregationStrategy) {
-        addOutput(new PollEnrichDefinition(aggregationStrategy, resourceUri, -1));
+        addOutput(new PollEnrichDefinition(aggregationStrategy, resourceUri, -1, false));
         return (Type) this;
     }
 
@@ -2869,7 +2914,7 @@ public abstract class ProcessorDefinitio
      */
     @SuppressWarnings("unchecked")
     public Type pollEnrich(String resourceUri, long timeout, AggregationStrategy aggregationStrategy)
{
-        addOutput(new PollEnrichDefinition(aggregationStrategy, resourceUri, timeout));
+        addOutput(new PollEnrichDefinition(aggregationStrategy, resourceUri, timeout, false));
         return (Type) this;
     }
 
@@ -2892,7 +2937,7 @@ public abstract class ProcessorDefinitio
      */
     @SuppressWarnings("unchecked")
     public Type pollEnrich(String resourceUri, long timeout) {
-        addOutput(new PollEnrichDefinition(null, resourceUri, timeout));
+        addOutput(new PollEnrichDefinition(null, resourceUri, timeout, false));
         return (Type) this;
     }
 

Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/PollEnricher.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/PollEnricher.java?rev=1358452&r1=1358451&r2=1358452&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/PollEnricher.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/PollEnricher.java Fri
Jul  6 22:13:51 2012
@@ -16,6 +16,9 @@
  */
 package org.apache.camel.processor;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import org.apache.camel.CamelExchangeException;
 import org.apache.camel.Exchange;
 import org.apache.camel.PollingConsumer;
@@ -47,6 +50,7 @@ public class PollEnricher extends Servic
     private AggregationStrategy aggregationStrategy;
     private PollingConsumer consumer;
     private long timeout;
+    private Boolean pollMultiple;
 
     /**
      * Creates a new {@link PollEnricher}. The default aggregation strategy is to
@@ -57,7 +61,7 @@ public class PollEnricher extends Servic
      * @param consumer consumer to resource endpoint.
      */
     public PollEnricher(PollingConsumer consumer) {
-        this(defaultAggregationStrategy(), consumer, 0);
+        this(defaultAggregationStrategy(), consumer, 0, false);
     }
 
     /**
@@ -74,6 +78,21 @@ public class PollEnricher extends Servic
     }
 
     /**
+     * Creates a new {@link PollEnricher}.
+     *
+     * @param aggregationStrategy  aggregation strategy to aggregate input data and additional
data.
+     * @param consumer consumer to resource endpoint.
+     * @param timeout timeout in millis
+     * @param pollMultiple enabled building a List of multiple exchanges
+     */
+    public PollEnricher(AggregationStrategy aggregationStrategy, PollingConsumer consumer,
long timeout, Boolean pollMultiple) {
+        this.aggregationStrategy = aggregationStrategy;
+        this.consumer = consumer;
+        this.timeout = timeout;
+        this.pollMultiple = pollMultiple;
+    }
+
+    /**
      * Sets the aggregation strategy for this poll enricher.
      *
      * @param aggregationStrategy the aggregationStrategy to set
@@ -100,6 +119,10 @@ public class PollEnricher extends Servic
         this.timeout = timeout;
     }
 
+    public void setPollMultiple(Boolean value) {
+        this.pollMultiple = value;
+    }
+    
     /**
      * Enriches the input data (<code>exchange</code>) by first obtaining
      * additional data from an endpoint represented by an endpoint
@@ -113,47 +136,70 @@ public class PollEnricher extends Servic
      * @param exchange input data.
      */
     public void process(Exchange exchange) throws Exception {
+
         preCheckPoll(exchange);
 
-        Exchange resourceExchange;
-        if (timeout < 0) {
-            LOG.debug("Consumer receive: {}", consumer);
-            resourceExchange = consumer.receive();
-        } else if (timeout == 0) {
-            LOG.debug("Consumer receiveNoWait: {}", consumer);
-            resourceExchange = consumer.receiveNoWait();
-        } else {
-            LOG.debug("Consumer receive with timeout: {} ms. {}", timeout, consumer);
-            resourceExchange = consumer.receive(timeout);
-        }
+        if (pollMultiple != null && pollMultiple) {
 
-        if (resourceExchange == null) {
-            LOG.debug("Consumer received no exchange");
-        } else {
-            LOG.debug("Consumer received: {}", resourceExchange);
-        }
+            List<Exchange> exchangeList = new ArrayList<Exchange>();
+            Exchange receivedExchange;
+            while (true) {
+                if (timeout == 0) {
+                    LOG.debug("Polling Consumer receiveNoWait: {}", consumer);
+                    receivedExchange = consumer.receiveNoWait();
+                } else {
+                    LOG.debug("Polling Consumer receive with timeout: {} ms. {}", timeout,
consumer);
+                    receivedExchange = consumer.receive(timeout);
+                }
 
-        if (resourceExchange != null && resourceExchange.isFailed()) {
-            // copy resource exchange onto original exchange (preserving pattern)
-            copyResultsPreservePattern(exchange, resourceExchange);
+                if (receivedExchange == null) {
+                    break;
+                }
+                exchangeList.add(receivedExchange);
+            }
+            exchange.getIn().setBody(exchangeList);
         } else {
-            prepareResult(exchange);
-
-            // prepare the exchanges for aggregation
-            ExchangeHelper.prepareAggregation(exchange, resourceExchange);
-            // must catch any exception from aggregation
-            Exchange aggregatedExchange;
-            try {
-                aggregatedExchange = aggregationStrategy.aggregate(exchange, resourceExchange);
-            } catch (Throwable e) {
-                throw new CamelExchangeException("Error occurred during aggregation", exchange,
e);
+        
+            Exchange resourceExchange;
+            if (timeout < 0) {
+                LOG.debug("Consumer receive: {}", consumer);
+                resourceExchange = consumer.receive();
+            } else if (timeout == 0) {
+                LOG.debug("Consumer receiveNoWait: {}", consumer);
+                resourceExchange = consumer.receiveNoWait();
+            } else {
+                LOG.debug("Consumer receive with timeout: {} ms. {}", timeout, consumer);
+                resourceExchange = consumer.receive(timeout);
+            }
+    
+            if (resourceExchange == null) {
+                LOG.debug("Consumer received no exchange");
+            } else {
+                LOG.debug("Consumer received: {}", resourceExchange);
             }
-            if (aggregatedExchange != null) {
-                // copy aggregation result onto original exchange (preserving pattern)
-                copyResultsPreservePattern(exchange, aggregatedExchange);
-                // handover any synchronization
-                if (resourceExchange != null) {
-                    resourceExchange.handoverCompletions(exchange);
+
+            if (resourceExchange != null && resourceExchange.isFailed()) {
+                // copy resource exchange onto original exchange (preserving pattern)
+                copyResultsPreservePattern(exchange, resourceExchange);
+            } else {
+                prepareResult(exchange);
+    
+                // prepare the exchanges for aggregation
+                ExchangeHelper.prepareAggregation(exchange, resourceExchange);
+                // must catch any exception from aggregation
+                Exchange aggregatedExchange;
+                try {
+                    aggregatedExchange = aggregationStrategy.aggregate(exchange, resourceExchange);
+                } catch (Throwable e) {
+                    throw new CamelExchangeException("Error occurred during aggregation",
exchange, e);
+                }
+                if (aggregatedExchange != null) {
+                    // copy aggregation result onto original exchange (preserving pattern)
+                    copyResultsPreservePattern(exchange, aggregatedExchange);
+                    // handover any synchronization
+                    if (resourceExchange != null) {
+                        resourceExchange.handoverCompletions(exchange);
+                    }
                 }
             }
         }

Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/enricher/PollEnricherTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/enricher/PollEnricherTest.java?rev=1358452&r1=1358451&r2=1358452&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/enricher/PollEnricherTest.java
(original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/enricher/PollEnricherTest.java
Fri Jul  6 22:13:51 2012
@@ -16,6 +16,9 @@
  */
 package org.apache.camel.processor.enricher;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import org.apache.camel.ContextTestSupport;
 import org.apache.camel.Exchange;
 import org.apache.camel.ExchangePattern;
@@ -116,6 +119,41 @@ public class PollEnricherTest extends Co
         assertNull(exchange.getException());
     }
 
+    public void testPollEnrichMultipleDefaultNoWait() throws InterruptedException {
+
+        mock.expectedMessageCount(1);
+        template.sendBody("seda:foo5", "msg1");
+        template.sendBody("seda:foo5", "msg2");
+        template.sendBody("seda:enricher-test-5", "test");
+        Thread.sleep(100);
+        template.sendBody("seda:foo5", "msg3");
+        template.sendBody("seda:foo5", "msg4");
+
+        List<Exchange> polledExchanges = mock.getExchanges().get(0).getIn().getBody(List.class);
+        assertEquals(2, polledExchanges.size());
+
+        mock.expectedHeaderReceived(Exchange.TO_ENDPOINT, "seda://foo5");
+        mock.assertIsSatisfied(0);
+    }
+
+    public void testPollEnrichMultipleExplicitTimeout() throws InterruptedException {
+
+        mock.expectedMessageCount(1);
+        template.sendBody("seda:foo6", "msg1");
+        template.sendBody("seda:foo6", "msg2");
+        template.sendBody("seda:enricher-test-6", "test");
+        template.sendBody("seda:foo6", "msg3");
+        template.sendBody("seda:foo6", "msg4");
+
+        Thread.sleep(500);
+
+        List<Exchange> polledExchanges = mock.getExchanges().get(0).getIn().getBody(List.class);
+        assertEquals(4, polledExchanges.size());
+
+        mock.expectedHeaderReceived(Exchange.TO_ENDPOINT, "seda://foo6");
+        mock.assertIsSatisfied(0);
+    }
+
     protected RouteBuilder createRouteBuilder() {
         return new RouteBuilder() {
             public void configure() {
@@ -141,6 +179,18 @@ public class PollEnricherTest extends Co
 
                 from("direct:enricher-test-4")
                     .pollEnrich("seda:foo4", aggregationStrategy);
+
+                // -------------------------------------------------------------
+                //  Poll Multiple routes
+                // -------------------------------------------------------------
+
+                from("seda:enricher-test-5")
+                    .pollEnrich("seda:foo5", true)
+                    .to("mock:mock");
+
+                from("seda:enricher-test-6")
+                    .pollEnrich("seda:foo6", 200, true)
+                    .to("mock:mock");
             }
         };
     }

Modified: camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/pollEnricher.xml
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/pollEnricher.xml?rev=1358452&r1=1358451&r2=1358452&view=diff
==============================================================================
--- camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/pollEnricher.xml
(original)
+++ camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/pollEnricher.xml
Fri Jul  6 22:13:51 2012
@@ -48,6 +48,19 @@
             <pollEnrich uri="seda:foo4" strategyRef="sampleAggregator"/>
             <to uri="mock:mock"/>
         </route>
+
+        <route>
+            <from uri="seda:enricher-test-5"/>
+            <pollEnrich uri="seda:foo5" pollMultiple="true"/>
+            <to uri="mock:mock"/>
+        </route>
+
+        <route>
+            <from uri="seda:enricher-test-6"/>
+            <pollEnrich uri="seda:foo6" timeout="200" pollMultiple="true"/>
+            <to uri="mock:mock"/>
+        </route>
+
     </camelContext>
 
     <bean id="sampleAggregator" class="org.apache.camel.processor.enricher.SampleAggregator"/>

Modified: camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/pollEnricherRef.xml
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/pollEnricherRef.xml?rev=1358452&r1=1358451&r2=1358452&view=diff
==============================================================================
--- camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/pollEnricherRef.xml
(original)
+++ camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/pollEnricherRef.xml
Fri Jul  6 22:13:51 2012
@@ -28,6 +28,8 @@
         <endpoint id="foo2" uri="seda:foo2"/>
         <endpoint id="foo3" uri="seda:foo3"/>
         <endpoint id="foo4" uri="seda:foo4"/>
+        <endpoint id="foo5" uri="seda:foo5"/>
+        <endpoint id="foo6" uri="seda:foo6"/>
 
         <!-- START SNIPPET: e1 -->
         <route>
@@ -54,6 +56,19 @@
             <pollEnrich ref="foo4" strategyRef="sampleAggregator"/>
             <to uri="mock:mock"/>
         </route>
+
+        <route>
+            <from uri="seda:enricher-test-5"/>
+            <pollEnrich ref="foo5" pollMultiple="true"/>
+            <to uri="mock:mock"/>
+        </route>
+
+        <route>
+            <from uri="seda:enricher-test-6"/>
+            <pollEnrich ref="foo6" timeout="200" pollMultiple="true"/>
+            <to uri="mock:mock"/>
+        </route>
+
     </camelContext>
 
     <bean id="sampleAggregator" class="org.apache.camel.processor.enricher.SampleAggregator"/>



Mime
View raw message