/**
 * Copyright 2014 Netflix, Inc.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package rx.internal.util;

import static org.junit.Assert.assertEquals;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.Test;

import rx.Producer;
import rx.Scheduler;
import rx.exceptions.MissingBackpressureException;
import rx.functions.Action0;
import rx.observers.TestSubscriber;
import rx.schedulers.Schedulers;

public class RxRingBufferWithoutUnsafeTest extends RxRingBufferBase {

    @Override
    protected RxRingBuffer createRingBuffer() {
        return new RxRingBuffer();
    }

    /**
     * Single producer, 2 consumers. The request() ensures it gets scheduled back on the same Producer thread.
     */
    @Test
    public void testConcurrency() throws InterruptedException {
        final RxRingBuffer b = createRingBuffer();
        final CountDownLatch emitLatch = new CountDownLatch(255);
        final CountDownLatch drainLatch = new CountDownLatch(2);

        final Scheduler.Worker w1 = Schedulers.newThread().createWorker();
        Scheduler.Worker w2 = Schedulers.newThread().createWorker();
        Scheduler.Worker w3 = Schedulers.newThread().createWorker();

        final AtomicInteger emit = new AtomicInteger();
        final AtomicInteger poll = new AtomicInteger();
        final AtomicInteger backpressureExceptions = new AtomicInteger();

        final Producer p = new Producer() {

//            AtomicInteger c = new AtomicInteger();

            @Override
            public void request(final long n) {
                //                System.out.println("request[" + c.incrementAndGet() + "]: " + n + "  Thread: " + Thread.currentThread());
                w1.schedule(new Action0() {

                    @Override
                    public void call() {
                        if (emitLatch.getCount() == 0) {
                            return;
                        }
                        for (int i = 0; i < n; i++) {
                            try {
                                b.onNext("one");
                                emit.incrementAndGet();
                            } catch (MissingBackpressureException e) {
                                System.out.println("BackpressureException => item: " + i + "  requested: " + n + " emit: " + emit.get() + "  poll: " + poll.get());
                                backpressureExceptions.incrementAndGet();
                            }
                        }
                        // we'll release after n batches
                        emitLatch.countDown();
                    }

                });
            }

        };
        final TestSubscriber<String> ts = new TestSubscriber<String>();
        w1.schedule(new Action0() {

            @Override
            public void call() {
                ts.requestMore(RxRingBuffer.SIZE);
                ts.setProducer(p);
            }

        });

        Action0 drainer = new Action0() {

            @Override
            public void call() {
                int emitted = 0;
                while (true) {
                    Object o = b.poll();
                    if (o != null) {
                        emitted++;
                        poll.incrementAndGet();
                    } else {
                        if (emitted > 0) {
                            ts.requestMore(emitted);
                            emitted = 0;
                        } else {
                            if (emitLatch.getCount() == 0) {
                                // this works with SynchronizedQueue, if changing to a non-blocking Queue
                                // then this will likely need to change like the SpmcTest version
                                drainLatch.countDown();
                                return;
                            }
                        }
                    }
                }

            }

        };

        w2.schedule(drainer);
        w3.schedule(drainer);

        emitLatch.await();
        drainLatch.await();

        w2.unsubscribe();
        w3.unsubscribe();
        w1.unsubscribe(); // put this one last as unsubscribing from it can cause Exceptions to be throw in w2/w3

        System.out.println("emit: " + emit.get() + " poll: " + poll.get());
        assertEquals(0, backpressureExceptions.get());
        assertEquals(emit.get(), poll.get());
    }
}

@@ -663,16 +663,36 @@
 quals;%0A%0A
+import java.util.*;%0A
 import j
@@ -816,37 +816,9 @@
  rx.
-Producer;%0Aimport rx.Scheduler
+*
 ;%0Aim
@@ -1136,24 +1136,210 @@
 r();%0A    %7D%0A%0A
+    @Test(timeout = 20000)%0A    public void testConcurrencyLoop() throws InterruptedException %7B%0A        for (int i = 0; i %3C 50; i++) %7B%0A            testConcurrency();%0A        %7D%0A    %7D%0A    %0A
     /**%0A    
@@ -1462,16 +1462,33 @@
    @Test
+(timeout = 10000)
 %0A    pub
@@ -1659,12 +1659,38 @@
 tch(
-255)
+127);%0A        int drainers = 3
 ;%0A  
@@ -1748,17 +1748,24 @@
 wnLatch(
-2
+drainers
 );%0A%0A    
@@ -1835,32 +1835,37 @@
 rker();%0A        
+List%3C
 Scheduler.Worker
@@ -1868,118 +1868,58 @@
 rker
- w2 = Schedulers.newThread().createWorker();%0A        Scheduler.Worker w3 = Schedulers.newThread().create
+%3E drainerWorkers = new ArrayList%3CScheduler.
 Worker
+%3E
 ();%0A
@@ -4173,32 +4173,252 @@
        %7D else %7B%0A
+                            try %7B%0A                                Thread.sleep(1);%0A                            %7D catch (InterruptedException ex) %7B%0A                                // ignored%0A                            %7D%0A
                 
@@ -4450,32 +4450,47 @@
 .getCount() == 0
+ && b.isEmpty()
 ) %7B%0A            
@@ -4913,27 +4913,114 @@
     
-w2.schedule(drain
+for (int i = 0; i %3C drainers; i++) %7B%0A            Scheduler.Worker w = Schedulers.newThread().createWork
 er
+(
 );%0A 
@@ -5026,18 +5026,21 @@
         
+  
+  
 w
-3
 .schedul
@@ -5050,16 +5050,61 @@
 rainer);
+%0A            drainerWorkers.add(w);%0A        %7D
 %0A%0A      
@@ -5165,34 +5165,62 @@
     
-w2.unsubscribe();%0A
+for (Scheduler.Worker w : drainerWorkers) %7B%0A  
         
 w3.u
@@ -5215,18 +5215,19 @@
         
+  
 w
-3
 .unsubsc
@@ -5230,24 +5230,34 @@
 ubscribe();%0A
+        %7D%0A
         w1.u

Fix the drainer to check if the queue is empty before quitting.