/****************************************************************
 * 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.james.transport.camel;

import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.PollingConsumer;
import org.apache.camel.Processor;
import org.apache.camel.impl.DefaultEndpoint;
import org.apache.camel.impl.ScheduledPollConsumer;

/**
 * Consumer which polls a JMS Endpoint with a selector which only selects messages with JAMES_NEXT_DELIVERY header
 * value is smaller then the current time in milliseconds.
 * 
 * This implementation should work on every JMS Queue
 * 
 * 
 * 
 *
 */
public class JMSSelectorPollingConsumer extends ScheduledPollConsumer{

    private CamelContext context;
	private String uri;
    
    public JMSSelectorPollingConsumer(DefaultEndpoint endpoint, Processor processor, CamelContext context) {
        super(endpoint, processor);
        this.context = context;
    }
  
    /**
     * Set the EndpointUri to poll on
     * 
     * @param uri
     */
    public void setEndpointUri(String uri) {
    	this.uri = uri;
    }
    
    @Override
    protected void poll() throws Exception {
      
        StringBuffer consumerUri = new StringBuffer();
        consumerUri.append(uri);
        consumerUri.append("?");
        consumerUri.append("transacted=true");
        consumerUri.append("&");
        consumerUri.append("selector=");
        consumerUri.append(JamesCamelConstants.JAMES_NEXT_DELIVERY);
        consumerUri.append(" < ");
        consumerUri.append(System.currentTimeMillis());
        
        PollingConsumer consumer = context.getEndpoint(consumerUri.toString()).createPollingConsumer();
        consumer.start();
        // process every exchange which is ready. If no exchange is left break the loop
        while(true) {
            Exchange ex = consumer.receiveNoWait();
            if (ex != null) {
                getProcessor().process(ex);
            } else {
                consumer.stop();
                break;
            }
            
        }
    }

}

@@ -2806,24 +2806,42 @@
 er.start();%0A
+        try %7B%0A    
         // p
@@ -2924,16 +2924,20 @@
         
+   
+ 
 while(tr
@@ -2938,24 +2938,28 @@
 ile(true) %7B%0A
+    
             
@@ -3010,16 +3010,20 @@
         
+    
 if (ex !
@@ -3048,16 +3048,20 @@
         
+  
+  
 getProce
@@ -3092,16 +3092,20 @@
         
+    
 %7D else %7B
@@ -3125,23 +3125,17 @@
     
-consumer.stop()
+    break
 ;%0A  
@@ -3140,38 +3140,33 @@
                 
-break;
+%7D
 %0A            %7D%0A 
@@ -3168,28 +3168,129 @@
   %7D%0A        
-    
+%7D finally %7B%0A            // be sure to stop the consumer, even on exception..%0A            consumer.stop();
 %0A        %7D%0A 

Make sure we shutdown the consumer , even on exception. Thx to Claus (Camel Dev Team) for the pointer..