poll npe fixed. #181

src/main/java/org/redisson/RedissonBlockingQueue.java

@@ -28,6 +28,7 @@ import org.redisson.async.SyncOperation;
 import org.redisson.connection.ConnectionManager;
 import org.redisson.core.RBlockingQueue;
 
+import com.lambdaworks.redis.KeyValue;
 import com.lambdaworks.redis.RedisConnection;
 import org.redisson.core.RScript;
 
@@ -71,7 +72,11 @@ public class RedissonBlockingQueue<V> extends RedissonQueue<V> implements RBlock
         return connectionManager.write(getName(), new SyncInterruptedOperation<V, V>() {
             @Override
             public V execute(RedisConnection<Object, V> conn) throws InterruptedException {
-                return conn.blpop(unit.toSeconds(timeout), getName()).value;
+                KeyValue<Object, V> val = conn.blpop(unit.toSeconds(timeout), getName());
+                if (val != null) {
+                    return val.value;
+                }
+                return null;
             }
         });
     }

src/test/java/org/redisson/RedissonBlockingQueueTest.java

@@ -15,6 +15,14 @@ import org.redisson.core.*;
 
 public class RedissonBlockingQueueTest extends BaseTest {
 
+    @Test
+    public void testPoll() throws InterruptedException {
+        RBlockingQueue<Integer> queue1 = redisson.getBlockingQueue("queue1");
+        queue1.put(1);
+        Assert.assertEquals((Integer)1, queue1.poll(2, TimeUnit.SECONDS));
+        Assert.assertNull(queue1.poll(2, TimeUnit.SECONDS));
+    }
+
     @Test
     public void testPollLastAndOfferFirstTo() throws InterruptedException {
         RBlockingQueue<Integer> queue1 = redisson.getBlockingQueue("queue1");
@@ -30,7 +38,7 @@ public class RedissonBlockingQueueTest extends BaseTest {
         queue1.pollLastAndOfferFirstTo(queue2, 10, TimeUnit.SECONDS);
         MatcherAssert.assertThat(queue2, Matchers.contains(3, 4, 5, 6));
     }
-    
+
     @Test
     public void testAddOfferOrigin() {
         Queue<Integer> queue = new LinkedList<Integer>();
@@ -129,7 +137,7 @@ public class RedissonBlockingQueueTest extends BaseTest {
         final AtomicInteger counter = new AtomicInteger();
         int total = 100;
         for (int i = 0; i < total; i++) {
-            // runnable won't be executed in any particular order, and hence, int value as well. 
+            // runnable won't be executed in any particular order, and hence, int value as well.
             executor.submit(new Runnable() {
                 @Override
                 public void run() {