package org.redisson;

import static org.assertj.core.api.Assertions.assertThat;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.junit.Test;
import org.redisson.api.RBlockingFairQueue;

public class RedissonBlockingFairQueueTest extends BaseTest {

    @Test
    public void testPollTimeout() throws InterruptedException {
        int size = 100;
        RBlockingFairQueue<String> queue = redisson.getBlockingFairQueue("test");

        CountDownLatch latch = new CountDownLatch(size);
        List<Thread> threads = new ArrayList<Thread>();
        for (int i = 0; i < size; i++) {
            final int j = i;
            Thread t = new Thread() {
                public void run() {
                    try {
                        String value = queue.poll(1, TimeUnit.SECONDS);
                        assertThat(value).isEqualTo("" + j);
                        latch.countDown();
                    } catch (InterruptedException e) {
                    }
                };
            };
            
            threads.add(t);
        }
        
        for (Thread thread : threads) {
            thread.start();
            thread.join(5);
        }
        
        for (int i = 0; i < size; i++) {
            queue.add("" + i);
        }
        
        assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
    }
    
}

@@ -208,16 +208,91 @@
 meUnit;%0A
+import java.util.concurrent.atomic.AtomicInteger;%0A%0Aimport org.junit.Assert;
 %0Aimport 
@@ -350,16 +350,56 @@
 irQueue;
+%0Aimport org.redisson.api.RBlockingQueue;
 %0A%0Apublic
@@ -489,19 +489,16 @@
 test
-PollTimeout
+Fairness
 () t
@@ -548,16 +548,18 @@
 ze = 100
+00
 ;%0A      
@@ -565,28 +565,24 @@
    RBlocking
-Fair
 Queue%3CString
@@ -607,28 +607,24 @@
 .getBlocking
-Fair
 Queue(%22test%22
@@ -696,136 +696,118 @@
     
-List%3CThread%3E threads = new ArrayList%3CThread%3E();%0A        for (int i = 0; i %3C size; i++) %7B%0A            final int j = i;%0A  
+AtomicInteger t1Counter = new AtomicInteger();%0A        AtomicInteger t2Counter = new AtomicInteger();%0A
         
   Th
@@ -806,18 +806,16 @@
     
-  
 Thread t
  = n
@@ -810,16 +810,17 @@
 Thread t
+1
  = new T
@@ -825,24 +825,34 @@
  Thread(
+%22test-thread1%22
 ) %7B%0A
-    
         
@@ -871,24 +871,55 @@
 oid run() %7B%0A
+                while (true) %7B%0A
             
@@ -963,21 +963,17 @@
  String 
-value
+a
  = queue
@@ -1028,44 +1028,85 @@
     
-assertThat(value).isEqualTo(%22%22 + j);
+if (a == null) %7B%0A                            break;%0A                        %7D
 %0A   
@@ -1141,24 +1141,77 @@
 ountDown();%0A
+                        t1Counter.incrementAndGet();%0A
             
@@ -1284,33 +1284,32 @@
                %7D
-;
 %0A            %7D;%0A
@@ -1312,28 +1312,27 @@
  %7D;%0A        
-    
+%7D;%0A
 %0A           
@@ -1332,27 +1332,111 @@
     
-    threads.add(t);
+Thread t2 = new Thread(%22test-thread1%22) %7B%0A            public void run() %7B%0A                while (true) %7B
 %0A   
@@ -1436,34 +1436,34 @@
 ) %7B%0A        
-%7D%0A
+  
         
 %0A        for
@@ -1442,32 +1442,39 @@
                 
+  try %7B
 %0A        for (Th
@@ -1470,94 +1470,415 @@
     
-for (Thread thread : threads) %7B%0A            thread.start();%0A            thread.join(5)
+                String a = queue.poll(1, TimeUnit.SECONDS);%0A                        if (a == null) %7B%0A                            break;%0A                        %7D%0A                        Thread.sleep(5);%0A                        latch.countDown();%0A                        t2Counter.incrementAndGet();%0A                    %7D catch (InterruptedException e) %7B%0A                    %7D%0A                %7D%0A            %7D
 ;%0A  
@@ -1876,32 +1876,33 @@
     %7D;%0A        %7D
+;
 %0A        %0A      
@@ -1977,24 +1977,25 @@
       %7D%0A
+%0A
         
 %0A       
@@ -1990,69 +1990,256 @@
     
-%0A        assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(
+t1.start();%0A        t2.start();%0A        %0A        t2.join();%0A        t1.join();%0A%0A        assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();%0A        System.out.println(%22t1: %22 + t1Counter.get());%0A        System.out.println(%22t2: %22 + t2Counter.get()
 );%0A 
@@ -2250,8 +2250,9 @@
 %0A    %0A%7D%0A
+%0A

Test fixed