package com.github.davidmoten.rx;

import static org.junit.Assert.assertTrue;

import java.util.concurrent.TimeUnit;

import org.junit.Test;

import rx.Observable;
import rx.functions.Functions;

public class OperationToOperatorTest {

    @Test
    public void testUnsubscribeFromAsynchronousSource() throws InterruptedException {

        UnsubscribeDetector<Long> detector = UnsubscribeDetector.detect();
        Observable.interval(100, TimeUnit.MILLISECONDS).lift(detector)
                .lift(new OperationToOperator<Long, Long>(Functions.<Observable<Long>> identity())).take(1).first()
                .toBlockingObservable().single();
        assertTrue(detector.latch().await(1, TimeUnit.SECONDS));

    }

    @Test
    public void testMultipleSubscriptions() {

    }

}

@@ -46,37 +46,243 @@
 tic 
-org.junit.Assert.assertTrue;%0A
+com.github.davidmoten.rx.OperationToOperator.toOperator;%0Aimport static java.util.Arrays.asList;%0Aimport static org.junit.Assert.assertEquals;%0Aimport static org.junit.Assert.assertTrue;%0A%0Aimport java.util.ArrayList;%0Aimport java.util.List;
 %0Aimp
@@ -363,16 +363,45 @@
 rvable;%0A
+import rx.functions.Action1;%0A
 import r
@@ -422,16 +422,51 @@
 nctions;
+%0Aimport rx.subjects.PublishSubject;
 %0A%0Apublic
@@ -495,24 +495,74 @@
 atorTest %7B%0A%0A
+    private static final int AWAIT_SECONDS = 10;%0A%0A
     @Test%0A  
@@ -743,118 +743,222 @@
 able
-.interval(100, TimeUnit.MILLISECONDS).lift(detector)%0A                .lift(new OperationToOperator%3CLong, Long%3E
+%0A        // every 100ms%0A                .interval(100, TimeUnit.MILLISECONDS)%0A                // detect unsubscribe%0A                .lift(detector)%0A                // use toOperator%0A                .lift(toOperator
 (Fun
@@ -1012,16 +1012,56 @@
 first()%0A
+                // block and get result%0A
         
@@ -1114,65 +1114,1234 @@
     
-assertTrue(detector.latch().await(1, TimeUnit.SECONDS));%0A
+// wait for expected unsubscription%0A        assertTrue(detector.latch().await(AWAIT_SECONDS, TimeUnit.SECONDS));%0A%0A    %7D%0A%0A    @Test%0A    public void testUnsubscribeFromSynchronousSource() throws InterruptedException %7B%0A        UnsubscribeDetector%3CInteger%3E detector = UnsubscribeDetector.detect();%0A        PublishSubject%3CInteger%3E subject = PublishSubject.create();%0A        subject%0A        // detect unsubscribe%0A        .lift(detector)%0A        // use toOperator%0A                .lift(toOperator(Functions.%3CObservable%3CInteger%3E%3E identity()))%0A                // get first only%0A                .take(1)%0A                // subscribe and ignore events%0A                .subscribe();%0A        subject.onNext(1);%0A        //should have unsubscribed because of take(1)%0A        assertTrue(detector.latch().await(AWAIT_SECONDS, TimeUnit.SECONDS));%0A    %7D%0A%0A    @Test%0A    public void testMultipleNonSimultaeousSubscriptions() %7B%0A        Observable%3CInteger%3E sequence = Observable.range(1, 3).lift(%0A                toOperator(Functions.%3CObservable%3CInteger%3E%3E identity()));%0A        assertEquals(asList(1, 2, 3), sequence.toList().toBlockingObservable().single());%0A        assertEquals(asList(1, 2, 3), sequence.toList().toBlockingObservable().single());
 %0A   
@@ -2383,16 +2383,28 @@
 ultipleS
+imultaneousS
 ubscript
@@ -2411,16 +2411,844 @@
 ions() %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        subject.onCompleted();%0A    %7D%0A%0A    private static class Recorder implements Action1%3CInteger%3E %7B%0A%0A        private final List%3CInteger%3E list = new ArrayList%3CInteger%3E();%0A%0A        @Override%0A        public void call(Integer t) %7B%0A            list.add(t);%0A        %7D%0A%0A        List%3CInteger%3E list() %7B%0A            return list;%0A        %7D
 %0A%0A    %7D%0A

add toOoperator unit tests