camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject svn commit: r1328817 - in /camel/branches/camel-2.9.x: ./ camel-core/src/main/java/org/apache/camel/component/file/ camel-core/src/test/java/org/apache/camel/component/file/ components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/ com...
Date Sun, 22 Apr 2012 08:43:03 GMT
Author: davsclaus
Date: Sun Apr 22 08:43:02 2012
New Revision: 1328817

URL: http://svn.apache.org/viewvc?rev=1328817&view=rev
Log:
CAMEL-5202: Added option eagerLimitMaxMessagesPerPoll to file/ftp endpoints.

Added:
    camel/branches/camel-2.9.x/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeNotEagerMaxMessagesPerPollTest.java
      - copied unchanged from r1328814, camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeNotEagerMaxMessagesPerPollTest.java
    camel/branches/camel-2.9.x/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/FtpConsumerNotEagerMaxMessagesPerPollTest.java
      - copied unchanged from r1328814, camel/trunk/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/FtpConsumerNotEagerMaxMessagesPerPollTest.java
Modified:
    camel/branches/camel-2.9.x/   (props changed)
    camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/component/file/FileEndpoint.java
    camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
    camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/component/file/GenericFileEndpoint.java
    camel/branches/camel-2.9.x/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFileEndpoint.java

Propchange: camel/branches/camel-2.9.x/
------------------------------------------------------------------------------
  Merged /camel/trunk:r1328814

Propchange: camel/branches/camel-2.9.x/
------------------------------------------------------------------------------
Binary property 'svnmerge-integrated' - no diff available.

Modified: camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/component/file/FileEndpoint.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/component/file/FileEndpoint.java?rev=1328817&r1=1328816&r2=1328817&view=diff
==============================================================================
--- camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/component/file/FileEndpoint.java
(original)
+++ camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/component/file/FileEndpoint.java
Sun Apr 22 08:43:02 2012
@@ -84,6 +84,7 @@ public class FileEndpoint extends Generi
 
         // set max messages per poll
         result.setMaxMessagesPerPoll(getMaxMessagesPerPoll());
+        result.setEagerLimitMaxMessagesPerPoll(isEagerMaxMessagesPerPoll());
 
         configureConsumer(result);
         return result;

Modified: camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java?rev=1328817&r1=1328816&r2=1328817&view=diff
==============================================================================
--- camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
(original)
+++ camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
Sun Apr 22 08:43:02 2012
@@ -18,6 +18,7 @@ package org.apache.camel.component.file;
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Deque;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Queue;
@@ -49,6 +50,7 @@ public abstract class GenericFileConsume
     protected volatile ShutdownRunningTask shutdownRunningTask;
     protected volatile int pendingExchanges;
     protected Processor customProcessor;
+    protected boolean eagerLimitMaxMessagesPerPoll = true;
 
     public GenericFileConsumer(GenericFileEndpoint<T> endpoint, Processor processor,
GenericFileOperations<T> operations) {
         super(endpoint, processor);
@@ -75,9 +77,18 @@ public abstract class GenericFileConsume
         this.customProcessor = processor;
     }
 
+    public boolean isEagerLimitMaxMessagesPerPoll() {
+        return eagerLimitMaxMessagesPerPoll;
+    }
+
+    public void setEagerLimitMaxMessagesPerPoll(boolean eagerLimitMaxMessagesPerPoll) {
+        this.eagerLimitMaxMessagesPerPoll = eagerLimitMaxMessagesPerPoll;
+    }
+
     /**
      * Poll for files
      */
+    @SuppressWarnings("unchecked")
     protected int poll() throws Exception {
         // must reset for each poll
         fileExpressionResult = null;
@@ -114,6 +125,7 @@ public abstract class GenericFileConsume
         }
 
         // sort using build in sorters so we can use expressions
+        // use a linked list so we can deque the exchanges
         LinkedList<Exchange> exchanges = new LinkedList<Exchange>();
         for (GenericFile<T> file : files) {
             Exchange exchange = endpoint.createExchange(file);
@@ -126,13 +138,24 @@ public abstract class GenericFileConsume
             Collections.sort(exchanges, endpoint.getSortBy());
         }
 
+        // use a queue for the exchanges
+        Deque<Exchange> q = exchanges;
+
+        // we are not eager limiting, but we have configured a limit, so cut the list of
files
+        if (!eagerLimitMaxMessagesPerPoll && maxMessagesPerPoll > 0) {
+            if (files.size() > maxMessagesPerPoll) {
+                log.debug("Limiting maximum messages to poll at {} files as there was more
messages in this poll.", maxMessagesPerPoll);
+                // must first remove excessive files from the in progress repository
+                removeExcessiveInProgressFiles(q, maxMessagesPerPoll);
+            }
+        }
+
         // consume files one by one
         int total = exchanges.size();
         if (total > 0) {
             log.debug("Total {} files to consume", total);
         }
 
-        Queue<Exchange> q = exchanges;
         int polledMessages = processBatch(CastUtils.cast(q));
 
         postPollCheck();
@@ -176,15 +199,22 @@ public abstract class GenericFileConsume
             }
         }
 
+        // drain any in progress files as we are done with this batch
+        removeExcessiveInProgressFiles((Deque) exchanges, 0);
+
+        return total;
+    }
+
+    @SuppressWarnings("unchecked")
+    protected void removeExcessiveInProgressFiles(Deque exchanges, int limit) {
         // remove the file from the in progress list in case the batch was limited by max
messages per poll
-        while (exchanges.size() > 0) {
-            Exchange exchange = (Exchange) exchanges.poll();
+        while (exchanges.size() > limit) {
+            // must remove last
+            Exchange exchange = (Exchange) exchanges.removeLast();
             GenericFile<T> file = (GenericFile<T>) exchange.getProperty(FileComponent.FILE_EXCHANGE_FILE);
             String key = file.getAbsoluteFilePath();
             endpoint.getInProgressRepository().remove(key);
         }
-
-        return total;
     }
 
     public boolean deferShutdown(ShutdownRunningTask shutdownRunningTask) {
@@ -242,6 +272,11 @@ public abstract class GenericFileConsume
      * @return <tt>true</tt> to continue, <tt>false</tt> to stop
due hitting maxMessagesPerPoll limit
      */
     public boolean canPollMoreFiles(List<?> fileList) {
+        // at this point we should not limit if we are not eager
+        if (!eagerLimitMaxMessagesPerPoll) {
+            return true;
+        }
+
         if (maxMessagesPerPoll <= 0) {
             // no limitation
             return true;

Modified: camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/component/file/GenericFileEndpoint.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/component/file/GenericFileEndpoint.java?rev=1328817&r1=1328816&r2=1328817&view=diff
==============================================================================
--- camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/component/file/GenericFileEndpoint.java
(original)
+++ camel/branches/camel-2.9.x/camel-core/src/main/java/org/apache/camel/component/file/GenericFileEndpoint.java
Sun Apr 22 08:43:02 2012
@@ -71,6 +71,7 @@ public abstract class GenericFileEndpoin
     protected boolean delete;
     protected boolean flatten;
     protected int maxMessagesPerPoll;
+    protected boolean eagerMaxMessagesPerPoll = true;
     protected int maxDepth = Integer.MAX_VALUE;
     protected int minDepth;
     protected String tempPrefix;
@@ -556,6 +557,14 @@ public abstract class GenericFileEndpoin
         this.maxMessagesPerPoll = maxMessagesPerPoll;
     }
 
+    public boolean isEagerMaxMessagesPerPoll() {
+        return eagerMaxMessagesPerPoll;
+    }
+
+    public void setEagerMaxMessagesPerPoll(boolean eagerMaxMessagesPerPoll) {
+        this.eagerMaxMessagesPerPoll = eagerMaxMessagesPerPoll;
+    }
+
     public int getMaxDepth() {
         return maxDepth;
     }
@@ -565,7 +574,6 @@ public abstract class GenericFileEndpoin
     }
 
     public int getMinDepth() {
-
         return minDepth;
     }
 

Modified: camel/branches/camel-2.9.x/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFileEndpoint.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.9.x/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFileEndpoint.java?rev=1328817&r1=1328816&r2=1328817&view=diff
==============================================================================
--- camel/branches/camel-2.9.x/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFileEndpoint.java
(original)
+++ camel/branches/camel-2.9.x/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFileEndpoint.java
Sun Apr 22 08:43:02 2012
@@ -92,6 +92,7 @@ public abstract class RemoteFileEndpoint
 
         // set max messages per poll
         consumer.setMaxMessagesPerPoll(getMaxMessagesPerPoll());
+        consumer.setEagerLimitMaxMessagesPerPoll(isEagerMaxMessagesPerPoll());
 
         configureConsumer(consumer);
         return consumer;



Mime
View raw message