/**
 * Copyright 2013 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.operators;

import static org.mockito.Matchers.*;
import static org.mockito.Mockito.*;

import org.junit.Test;

import org.mockito.InOrder;
import rx.Observable;
import rx.Observer;
import rx.Scheduler;
import rx.Subscription;
import rx.concurrency.Schedulers;
import rx.util.functions.Func1;

public class OperationObserveOn {

    public static <T> Func1<Observer<T>, Subscription> observeOn(Observable<T> source, Scheduler scheduler) {
        return new ObserveOn<T>(source, scheduler);
    }

    private static class ObserveOn<T> implements Func1<Observer<T>, Subscription> {
        private final Observable<T> source;
        private final Scheduler scheduler;

        public ObserveOn(Observable<T> source, Scheduler scheduler) {
            this.source = source;
            this.scheduler = scheduler;
        }

        @Override
        public Subscription call(final Observer<T> observer) {
            return source.subscribe(new ScheduledObserver<T>(observer, scheduler));
        }
    }

    public static class UnitTest {

        @Test
        @SuppressWarnings("unchecked")
        public void testObserveOn() {

            Scheduler scheduler = spy(Tester.UnitTest.forwardingScheduler(Schedulers.immediate()));

            Observer<Integer> observer = mock(Observer.class);
            Observable.create(observeOn(Observable.toObservable(1, 2, 3), scheduler)).subscribe(observer);

            verify(observer, times(1)).onNext(1);
            verify(observer, times(1)).onNext(2);
            verify(observer, times(1)).onNext(3);
            verify(observer, times(1)).onCompleted();
        }


        @Test
        @SuppressWarnings("unchecked")
        public void testOrdering() throws InterruptedException {
            Observable<String> obs = Observable.from("one", null, "two", "three", "four");

            Observer<String> observer = mock(Observer.class);

            InOrder inOrder = inOrder(observer);

            obs.observeOn(Schedulers.threadPoolForComputation()).subscribe(observer);

            Thread.sleep(500); // !!! not a true unit test

            inOrder.verify(observer, times(1)).onNext("one");
            inOrder.verify(observer, times(1)).onNext(null);
            inOrder.verify(observer, times(1)).onNext("two");
            inOrder.verify(observer, times(1)).onNext("three");
            inOrder.verify(observer, times(1)).onNext("four");
            inOrder.verify(observer, times(1)).onCompleted();
            inOrder.verifyNoMoreInteractions();
        }

    }

}

@@ -635,24 +635,20 @@
 org.
-mockito.Matchers
+junit.Assert
 .*;%0A
@@ -696,53 +696,220 @@
 ort 
-org.junit.Test;%0A%0Aimport org.mockito.InOrd
+java.util.concurrent.CountDownLatch;%0Aimport java.util.concurrent.TimeUnit;%0A%0Aimport org.junit.Test;%0Aimport org.mockito.InOrder;%0Aimport org.mockito.invocation.InvocationOnMock;%0Aimport org.mockito.stubbing.Answ
 er;%0A
+%0A
 impo
@@ -2386,17 +2386,16 @@
      %7D%0A%0A
-%0A
         
@@ -2725,141 +2725,579 @@
     
-obs.observeOn(Schedulers.threadPoolForComputation()).subscribe(observer);%0A%0A            Thread.sleep(500); // !!! not a true unit test
+final CountDownLatch completedLatch = new CountDownLatch(1);%0A            doAnswer(new Answer%3CVoid%3E() %7B%0A%0A                @Override%0A                public Void answer(InvocationOnMock invocation) throws Throwable %7B%0A                    completedLatch.countDown();%0A                    return null;%0A                %7D%0A            %7D).when(observer).onCompleted();%0A%0A            obs.observeOn(Schedulers.threadPoolForComputation()).subscribe(observer);%0A%0A            if (!completedLatch.await(1000, TimeUnit.MILLISECONDS)) %7B%0A                fail(%22timed out waiting%22);%0A            %7D
 %0A%0A  

Refactor test to use CountDownLatch instead of Thread.sleep