// Path: src/main/java/com/plexobject/rx/scheduler/Scheduler.java
// public interface Scheduler extends Disposable {
//     /**
//      * This method registers user-defined function that is invoked by scheduler
//      * 
//      * @param consumer
//      *            - callback function to notify tick
//      * @param handle
//      *            to pass in with consumer
//      */
//     <T> void scheduleBackgroundTask(Consumer<T> consumer, T handle);
// 
//     public static Scheduler newThreadPoolScheduler(int poolSize) {
//         return new ThreadPoolScheduler(poolSize);
//     }
// 
//     public static Scheduler newImmediateScheduler() {
//         return new ImmediateScheduler();
//     }
// 
//     public static Scheduler newNewThreadScheduler() {
//         return new NewThreadScheduler();
//     }
// 
//     public static Scheduler newTimerSchedulerWithMilliInterval(long interval) {
//         return new TimerScheduler(interval);
//     }
// }


import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import com.plexobject.rx.scheduler.Scheduler;
        setupCallback(observable, (v) -> {
            throw new RuntimeException("test error");
        }, true);
        latch.await(100, TimeUnit.MILLISECONDS);

        assertEquals(0, onCompleted.get());
        assertNotNull(onError.get());
        assertEquals(0, onNext.get());
    }

    @Test
    public void testSubscribeParallel() throws Exception {
        Observable<Integer> observable = Observable.range(1, 101)
                .subscribeOn(Scheduler.newNewThreadScheduler()).parallel();

        initLatch(100 + 1); // N*onNext + onCompleted
        List<Integer> processedList = Collections
                .synchronizedList(new ArrayList<>());
        setupCallback(observable, (v) -> {
            processedList.add(v);
        }, true);
        latch.await(100, TimeUnit.MILLISECONDS);
        System.out.println(processedList);

        assertFalse(isSorted(processedList)); // parallel processing will
                                              // execute list in unsorted
                                              // fashion
        assertEquals(100, processedList.size());
        assertEquals(100, onNext.get());
        assertNull(onError.get());
// Path: src/main/java/com/plexobject/rx/scheduler/Scheduler.java
// public interface Scheduler extends Disposable {
//     /**
//      * This method registers user-defined function that is invoked by scheduler
//      * 
//      * @param consumer
//      *            - callback function to notify tick
//      * @param handle
//      *            to pass in with consumer
//      */
//     <T> void scheduleBackgroundTask(Consumer<T> consumer, T handle);
// 
//     public static Scheduler newThreadPoolScheduler(int poolSize) {
//         return new ThreadPoolScheduler(poolSize);
//     }
// 
//     public static Scheduler newImmediateScheduler() {
//         return new ImmediateScheduler();
//     }
// 
//     public static Scheduler newNewThreadScheduler() {
//         return new NewThreadScheduler();
//     }
// 
//     public static Scheduler newTimerSchedulerWithMilliInterval(long interval) {
//         return new TimerScheduler(interval);
//     }
// }


// Path: src/test/java/com/plexobject/rx/ObservableParallelTest.java
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import com.plexobject.rx.scheduler.Scheduler;

        setupCallback(observable, (v) -> {
            throw new RuntimeException("test error");
        }, true);
        latch.await(100, TimeUnit.MILLISECONDS);

        assertEquals(0, onCompleted.get());
        assertNotNull(onError.get());
        assertEquals(0, onNext.get());
    }

    @Test
    public void testSubscribeParallel() throws Exception {
        Observable<Integer> observable = Observable.range(1, 101)
                .subscribeOn(Scheduler.newNewThreadScheduler()).parallel();

        initLatch(100 + 1); // N*onNext + onCompleted
        List<Integer> processedList = Collections
                .synchronizedList(new ArrayList<>());
        setupCallback(observable, (v) -> {
            processedList.add(v);
        }, true);
        latch.await(100, TimeUnit.MILLISECONDS);
        System.out.println(processedList);

        assertFalse(isSorted(processedList)); // parallel processing will
                                              // execute list in unsorted
                                              // fashion
        assertEquals(100, processedList.size());
        assertEquals(100, onNext.get());
        assertNull(onError.get());
        assertEquals(1, onCompleted.get());