From 740684fb67ece3821ff54e5d4bd2e341918d4e2b Mon Sep 17 00:00:00 2001
From: Sergey Poletaev <s.poletaev@devinotele.com>
Date: Thu, 28 May 2015 11:32:29 +0300
Subject: [PATCH] Bugfix for RedissonBlockingQueue#drainTo method with
 maxElements

---
 .../org/redisson/RedissonBlockingQueue.java    |  4 ++--
 .../redisson/RedissonBlockingQueueTest.java    | 18 ++++++++++++++++++
 2 files changed, 20 insertions(+), 2 deletions(-)

diff --git a/src/main/java/org/redisson/RedissonBlockingQueue.java b/src/main/java/org/redisson/RedissonBlockingQueue.java
index c8840a71f..665e8a66a 100644
--- a/src/main/java/org/redisson/RedissonBlockingQueue.java
+++ b/src/main/java/org/redisson/RedissonBlockingQueue.java
@@ -130,8 +130,8 @@ public class RedissonBlockingQueue<V> extends RedissonQueue<V> implements RBlock
                         return Collections.emptyList();
                     }
                     conn.multi();
-                    conn.lrange(getName(), 0, len);
-                    conn.ltrim(getName(), 0, len);
+                    conn.lrange(getName(), 0, len - 1);
+                    conn.ltrim(getName(), len, -1);
                     List<Object> res = conn.exec();
                     if (res.size() == 2) {
                         List<V> items = (List<V>) res.get(0);
diff --git a/src/test/java/org/redisson/RedissonBlockingQueueTest.java b/src/test/java/org/redisson/RedissonBlockingQueueTest.java
index 3c39d7490..455541695 100644
--- a/src/test/java/org/redisson/RedissonBlockingQueueTest.java
+++ b/src/test/java/org/redisson/RedissonBlockingQueueTest.java
@@ -101,6 +101,24 @@ public class RedissonBlockingQueueTest extends BaseTest {
         queue.remove();
     }
 
+    @Test
+    public void testDrainTo() {
+        RBlockingQueue<Integer> queue = redisson.getBlockingQueue("queue");
+        for (int i = 0 ; i < 100; i++) {
+            queue.offer(i);
+        }
+        Assert.assertEquals(100, queue.size());
+        Set<Integer> batch = new HashSet<Integer>();
+        int count = queue.drainTo(batch, 10);
+        Assert.assertEquals(10, count);
+        Assert.assertEquals(10, batch.size());
+        Assert.assertEquals(90, queue.size());
+        queue.drainTo(batch, 10);
+        queue.drainTo(batch, 20);
+        queue.drainTo(batch, 60);
+        Assert.assertEquals(0, queue.size());
+    }
+
     @Test
     public void testBlockingQueue() {