package com.github.davidmoten.rx;

import static com.github.davidmoten.rx.OperationToOperator.toOperator;
import static java.util.Arrays.asList;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

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

import org.junit.Test;

import rx.Observable;
import rx.functions.Action1;
import rx.functions.Functions;
import rx.subjects.PublishSubject;

public class OperationToOperatorTest {

    private static final int AWAIT_SECONDS = 10;

    @Test
    public void testUnsubscribeFromAsynchronousSource() throws InterruptedException {

        UnsubscribeDetector<Long> detector = UnsubscribeDetector.detect();
        Observable
        // every 100ms
                .interval(100, TimeUnit.MILLISECONDS)
                // detect unsubscribe
                .lift(detector)
                // use toOperator
                .lift(toOperator(Functions.<Observable<Long>> identity())).take(1).first()
                // block and get result
                .toBlockingObservable().single();
        // wait for expected unsubscription
        assertTrue(detector.latch().await(AWAIT_SECONDS, TimeUnit.SECONDS));

    }

    @Test
    public void testUnsubscribeFromSynchronousSource() throws InterruptedException {
        UnsubscribeDetector<Integer> detector = UnsubscribeDetector.detect();
        PublishSubject<Integer> subject = PublishSubject.create();
        subject
        // detect unsubscribe
        .lift(detector)
        // use toOperator
                .lift(toOperator(Functions.<Observable<Integer>> identity()))
                // get first only
                .take(1)
                // subscribe and ignore events
                .subscribe();
        subject.onNext(1);
        //should have unsubscribed because of take(1)
        assertTrue(detector.latch().await(AWAIT_SECONDS, TimeUnit.SECONDS));
    }

    @Test
    public void testMultipleNonSimultaeousSubscriptions() {
        Observable<Integer> sequence = Observable.range(1, 3).lift(
                toOperator(Functions.<Observable<Integer>> identity()));
        assertEquals(asList(1, 2, 3), sequence.toList().toBlockingObservable().single());
        assertEquals(asList(1, 2, 3), sequence.toList().toBlockingObservable().single());
    }

    @Test
    public void testMultipleSimultaneousSubscriptions() {
        PublishSubject<Integer> subject = PublishSubject.create();
        Recorder recorder1 = new Recorder();
        Recorder recorder2 = new Recorder();
        subject.subscribe(recorder1);
        subject.subscribe(recorder2);
        subject.onNext(1);
        assertEquals(asList(1), recorder1.list());
        assertEquals(asList(1), recorder2.list());
        subject.onNext(2);
        assertEquals(asList(1, 2), recorder1.list());
        assertEquals(asList(1, 2), recorder2.list());
        subject.onCompleted();
    }

    private static class Recorder implements Action1<Integer> {

        private final List<Integer> list = new ArrayList<Integer>();

        @Override
        public void call(Integer t) {
            list.add(t);
        }

        List<Integer> list() {
            return list;
        }

    }

}

@@ -183,16 +183,60 @@
 Equals;%0A
+import static org.junit.Assert.assertFalse;%0A
 import s
@@ -421,25 +421,16 @@
  rx.
-functions.Action1
+Observer
 ;%0Aim
@@ -2956,102 +2956,1213 @@
     
-subject.onCompleted();%0A    %7D%0A%0A    private static class Recorder implements Action1%3CInteger%3E %7B%0A
+assertFalse(recorder1.isCompleted());%0A        assertFalse(recorder2.isCompleted());%0A        subject.onCompleted();%0A        assertTrue(recorder1.isCompleted());%0A        assertTrue(recorder2.isCompleted());%0A    %7D%0A    %0A    @Test%0A    public void testErrorsPassedThroughToOperator() %7B%0A        PublishSubject%3CInteger%3E subject = PublishSubject.create();%0A        Recorder recorder1 = new Recorder();%0A        Recorder recorder2 = new Recorder();%0A        subject.subscribe(recorder1);%0A        subject.subscribe(recorder2);%0A        subject.onNext(1);%0A        assertEquals(asList(1), recorder1.list());%0A        assertEquals(asList(1), recorder2.list());%0A        subject.onNext(2);%0A        assertEquals(asList(1, 2), recorder1.list());%0A        assertEquals(asList(1, 2), recorder2.list());%0A        Exception e = new Exception(%22boo%22);%0A        assertTrue(recorder1.errors().isEmpty());%0A        assertTrue(recorder2.errors().isEmpty());%0A        subject.onError(e);%0A        assertEquals(asList(e),recorder1.errors());%0A        assertEquals(asList(e),recorder2.errors());%0A    %7D%0A%0A    %0A    private static class Recorder implements Observer%3CInteger%3E %7B%0A%0A        private final List%3CThrowable%3E errors = new ArrayList%3CThrowable%3E();
 %0A   
@@ -4198,17 +4198,16 @@
 er%3E list
- 
 = new Ar
@@ -4230,61 +4230,432 @@
 ();%0A
-%0A        @Override%0A        public void call(Integer t
+        private boolean completed= false;%0A        %0A        boolean isCompleted() %7B%0A            return completed;%0A        %7D%0A%0A        List%3CThrowable%3E errors() %7B%0A            return errors;%0A        %7D%0A        %0A        List%3CInteger%3E list() %7B%0A            return list;%0A        %7D%0A%0A%0A        @Override%0A        public void onCompleted() %7B%0A            completed=true;%0A        %7D%0A%0A%0A        @Override%0A        public void onError(Throwable e
 ) %7B%0A
@@ -4670,18 +4670,20 @@
     
-list
+errors
 .add(
-t
+e
 );%0A 
@@ -4688,32 +4688,33 @@
 %0A        %7D%0A%0A
+%0A
         
 List%3CInteger
@@ -4701,35 +4701,62 @@
         
-List%3CInteger%3E list(
+@Override%0A        public void onNext(Integer t
 ) %7B%0A    
@@ -4759,35 +4759,35 @@
             
-return list
+list.add(t)
 ;%0A        %7D%0A

add unit tests