camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acosent...@apache.org
Subject [1/2] camel git commit: CAMEL-8992 Camel-Hazelcast: add drainTo operation to hazelcast queue producer
Date Tue, 23 Jun 2015 15:15:43 GMT
Repository: camel
Updated Branches:
  refs/heads/master 1aea80a26 -> b62f330c6


CAMEL-8992 Camel-Hazelcast: add drainTo operation to hazelcast queue producer


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/3f8b9b1d
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/3f8b9b1d
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/3f8b9b1d

Branch: refs/heads/master
Commit: 3f8b9b1d461c17f1b72803e7cfbfdb992b658adb
Parents: 1aea80a
Author: Andrea Cosentino <ancosen@gmail.com>
Authored: Tue Jun 23 17:14:32 2015 +0200
Committer: Andrea Cosentino <ancosen@gmail.com>
Committed: Tue Jun 23 17:14:32 2015 +0200

----------------------------------------------------------------------
 .../component/hazelcast/HazelcastConstants.java |  4 ++-
 .../hazelcast/queue/HazelcastQueueProducer.java | 26 ++++++++++++++++++++
 .../hazelcast/HazelcastQueueProducerTest.java   | 20 +++++++++++++++
 3 files changed, 49 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/3f8b9b1d/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastConstants.java
----------------------------------------------------------------------
diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastConstants.java
b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastConstants.java
index a146d46..6c42706 100644
--- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastConstants.java
+++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastConstants.java
@@ -41,6 +41,7 @@ public final class HazelcastConstants {
     public static final String TTL_UNIT = "CamelHazelcastObjectTtlUnit";
     public static final String QUERY = "CamelHazelcastQuery";
     public static final String EXPECTED_VALUE = "CamelHazelcastExpectedValue";
+    public static final String DRAIN_TO_COLLECTION = "CamelHazelcastDrainToCollection";
 
     /*
      * outgoing header properties
@@ -89,9 +90,10 @@ public final class HazelcastConstants {
     public static final int PEEK_OPERATION = 33;
     public static final int POLL_OPERATION = 34;
     public static final int REMAINING_CAPACITY_OPERATION = 35;
+    public static final int DRAIN_TO_OPERATION = 36;
 
     // topic
-    public static final int PUBLISH_OPERATION = 36;
+    public static final int PUBLISH_OPERATION = 37;
 
     /*
      * header values

http://git-wip-us.apache.org/repos/asf/camel/blob/3f8b9b1d/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/queue/HazelcastQueueProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/queue/HazelcastQueueProducer.java
b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/queue/HazelcastQueueProducer.java
index c814e3c..671eb60 100644
--- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/queue/HazelcastQueueProducer.java
+++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/queue/HazelcastQueueProducer.java
@@ -16,6 +16,9 @@
  */
 package org.apache.camel.component.hazelcast.queue;
 
+import java.util.Collection;
+import java.util.Map;
+
 import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.core.IQueue;
 
@@ -24,6 +27,7 @@ import org.apache.camel.component.hazelcast.HazelcastComponentHelper;
 import org.apache.camel.component.hazelcast.HazelcastConstants;
 import org.apache.camel.component.hazelcast.HazelcastDefaultEndpoint;
 import org.apache.camel.component.hazelcast.HazelcastDefaultProducer;
+import org.apache.camel.util.ObjectHelper;
 
 /**
  *
@@ -38,6 +42,15 @@ public class HazelcastQueueProducer extends HazelcastDefaultProducer {
     }
 
     public void process(Exchange exchange) throws Exception {
+        
+        Map<String, Object> headers = exchange.getIn().getHeaders();
+
+        // get header parameters
+        Object draintToCollection = null;
+        
+        if (headers.containsKey(HazelcastConstants.DRAIN_TO_COLLECTION)) {
+            draintToCollection = headers.get(HazelcastConstants.DRAIN_TO_COLLECTION);
+        }
 
         final int operation = lookupOperationNumber(exchange);
 
@@ -73,6 +86,14 @@ public class HazelcastQueueProducer extends HazelcastDefaultProducer {
             this.remainingCapacity(exchange);
             break;
             
+        case HazelcastConstants.DRAIN_TO_OPERATION:
+            if (ObjectHelper.isNotEmpty(draintToCollection)) {
+                this.drainTo((Collection) draintToCollection, exchange);
+            } else {
+                throw new IllegalArgumentException("Drain to collection header must be specified");
+            }
+            break;
+            
         default:
             throw new IllegalArgumentException(String.format("The value '%s' is not allowed
for parameter '%s' on the QUEUE cache.", operation, HazelcastConstants.OPERATION));
         }
@@ -117,4 +138,9 @@ public class HazelcastQueueProducer extends HazelcastDefaultProducer {
     private void remainingCapacity(Exchange exchange) {
         exchange.getOut().setBody(this.queue.remainingCapacity());
     }
+    
+    private void drainTo(Collection c, Exchange exchange) {
+        exchange.getOut().setBody(this.queue.drainTo(c));
+        exchange.getOut().setHeader(HazelcastConstants.DRAIN_TO_COLLECTION, c);
+    }
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/camel/blob/3f8b9b1d/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastQueueProducerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastQueueProducerTest.java
b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastQueueProducerTest.java
index 3807e98..24a4a78 100644
--- a/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastQueueProducerTest.java
+++ b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastQueueProducerTest.java
@@ -16,8 +16,14 @@
  */
 package org.apache.camel.component.hazelcast;
 
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
 import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.core.IQueue;
+
 import org.apache.camel.CamelExecutionException;
 import org.apache.camel.builder.RouteBuilder;
 import org.junit.After;
@@ -122,6 +128,17 @@ public class HazelcastQueueProducerTest extends HazelcastCamelTestSupport
{
         verify(queue).remainingCapacity();
         assertEquals(10, answer);
     }
+    
+    @Test
+    public void drainTo() throws InterruptedException {
+        Map<String, Object> headers = new HashMap<String, Object>();
+        Collection l = new ArrayList<>();
+        headers.put(HazelcastConstants.DRAIN_TO_COLLECTION, l);
+        when(queue.drainTo(l)).thenReturn(10);
+        int answer = template.requestBodyAndHeaders("direct:drainTo", "test", headers, Integer.class);
+        verify(queue).drainTo(l);
+        assertEquals(10, answer);
+    }
 
     @Override
     protected RouteBuilder createRouteBuilder() throws Exception {
@@ -148,6 +165,9 @@ public class HazelcastQueueProducerTest extends HazelcastCamelTestSupport
{
                 from("direct:remainingCapacity").setHeader(HazelcastConstants.OPERATION,
constant(HazelcastConstants.REMAINING_CAPACITY_OPERATION)).to(
                         String.format("hazelcast:%sbar", HazelcastConstants.QUEUE_PREFIX));
                 
+                from("direct:drainTo").setHeader(HazelcastConstants.OPERATION, constant(HazelcastConstants.DRAIN_TO_OPERATION)).to(
+                        String.format("hazelcast:%sbar", HazelcastConstants.QUEUE_PREFIX));
+                
                 from("direct:putWithOperationNumber").toF(String.format("hazelcast:%sbar?operation=%s",
HazelcastConstants.QUEUE_PREFIX, HazelcastConstants.PUT_OPERATION));
 
                 from("direct:putWithOperationName").toF(String.format("hazelcast:%sbar?operation=put",
HazelcastConstants.QUEUE_PREFIX));


Mime
View raw message