Merge pull request #167 from spoletaev/rblockingQueueFix

Bugfix for RedissonBlockingQueue#drainTo method with maxElements

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);

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() {