package com.github.davidmoten.rx.operators;

import static org.junit.Assert.assertEquals;
import static rx.Observable.range;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.junit.Test;

import rx.Observable;
import rx.Observable.OnSubscribe;
import rx.Subscriber;
import rx.functions.Action1;
import rx.observers.TestSubscriber;

import com.github.davidmoten.rx.Transformers;

public class OperatorBufferEmissionsTest {
    @Test
    public void testSyncDrainerDeliversRequestedFromBackpressureEnabledSource() {
        final AtomicInteger count = new AtomicInteger(0);
        createSource()
        // buffer
                .compose(Transformers.<Integer> bufferEmissions())
                // subscribe
                .subscribe(new Subscriber<Integer>() {

                    @Override
                    public void onStart() {
                        request(200);
                    }

                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {
                    }

                    @Override
                    public void onNext(Integer t) {
                        count.incrementAndGet();
                    }
                });
        assertEquals(200, count.get());
    }

    private static Observable<Integer> createSource() {
        return Observable.create(new OnSubscribe<Integer>() {

            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                for (int i = 0; i < 1000; i++)
                    subscriber.onNext(i);
                subscriber.onCompleted();
            }

        });
    }

    @Test
    public void testUpstreamRequests() {
        final AtomicLong requests = new AtomicLong();
        TestSubscriber<Integer> ts = new TestSubscriber<Integer>(100);
        Observable.range(1, 1000)
        //
                .doOnRequest(new Action1<Long>() {

                    @Override
                    public void call(Long n) {
                        requests.addAndGet(n);
                    }
                }) // buffer
                .compose(Transformers.<Integer> bufferEmissions())
                //
                .subscribe(ts);
        assertEquals(100, requests.get());
    }

    @Test
    public void testRequestOverflow() {
        final TestSubscriber<Integer> ts = new TestSubscriber<Integer>(Long.MAX_VALUE - 2);
        Observable.range(1, 10)
        //
                .doOnNext(new Action1<Integer>() {

                    @Override
                    public void call(Integer n) {
                        ts.requestMore(5);
                    }
                }) // buffer
                .compose(Transformers.<Integer> bufferEmissions())
                //
                .subscribe(ts);
        ts.assertCompleted();
        assertEquals(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10), ts.getOnNextEvents());
    }

    @Test
    public void testWithMaxValueRequests() {
        List<Integer> list = range(1, 3).compose(Transformers.<Integer> bufferEmissions()).toList()
                .toBlocking().single();
        assertEquals(Arrays.asList(1, 2, 3), list);
    }

    @Test
    public void testUnsubscribeAfterOne() {
        TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {

            @Override
            public void onNext(Integer t) {
                super.onNext(t);
                unsubscribe();
            }

        };
        range(1, 3).compose(Transformers.<Integer> bufferEmissions()).subscribe(ts);
        ts.assertUnsubscribed();
        ts.assertValues(1);
    }

    @Test
    public void test() {
        TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {

            @Override
            public void onNext(Integer t) {
                super.onNext(t);
                throw new RuntimeException("boo");
            }

        };
        range(1, 3).compose(Transformers.<Integer> bufferEmissions()).subscribe(ts);
        ts.assertUnsubscribed();
        ts.assertValues(1);
        ts.assertError(RuntimeException.class);
    }
}

@@ -3797,16 +3797,21 @@
 oid test
+Error
 () %7B%0A   
@@ -3865,35 +3865,33 @@
 riber%3CInteger%3E()
- %7B%0A
+;
 %0A            @Ov
@@ -3887,164 +3887,82 @@
     
-    @Override%0A            public void onNext(Integer t) %7B%0A                super.onNext(t);%0A                throw new RuntimeException(%22boo%22);%0A            %7D%0A
+range(1, 3).concatWith(Observable.%3CInteger%3E error(new RuntimeException()))
 %0A   
@@ -3962,35 +3962,32 @@
 )))%0A        
-%7D;%0A
         
 range(1, 3).
@@ -3970,35 +3970,24 @@
             
-range(1, 3)
 .compose(Tra
@@ -4065,32 +4065,32 @@
 Unsubscribed();%0A
-
         ts.asser
@@ -4094,24 +4094,30 @@
 sertValues(1
+, 2, 3
 );%0A        t

add error test for DrainerSync