// 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/main/java/com/plexobject/rx/util/NatsSpliterator.java
// public class NatsSpliterator implements Spliterator<Integer>, Streamable<Integer> {
//     private int number;
// 
//     public NatsSpliterator(int from) {
//         this.number = from;
//     }
// 
//     @Override
//     public boolean tryAdvance(Consumer<? super Integer> action) {
//         action.accept(number++);
//         return true;
//     }
// 
//     @Override
//     public Spliterator<Integer> trySplit() {
//         return null;
//     }
// 
//     @Override
//     public long estimateSize() {
//         return Integer.MAX_VALUE;
//     }
// 
//     @Override
//     public int characteristics() {
//         return Spliterator.IMMUTABLE;
//     }
// 
//     public Stream<Integer> getStream() {
//         return StreamSupport.stream(this, false);
//     }
// }


import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import org.junit.Test;
import com.plexobject.rx.scheduler.Scheduler;
import com.plexobject.rx.util.NatsSpliterator;
        initLatch(names.size() + 1); // N*onNext + onCompleted

        Observable<String> observable = Observable.from(names.stream());
        setupCallback(observable, null, true);
        latch.await(100, TimeUnit.MILLISECONDS);
        //
        assertEquals(names.size(), onNext.get());
        //
        assertNull(onError.get());
        assertEquals(1, onCompleted.get());
    }

    @Test
    public void testSubscribeFromIterator() throws Exception {
        initLatch(names.size() + 1); // N*onNext + onCompleted

        Observable<String> observable = Observable.from(names.iterator());

        setupCallback(observable, null, true);
        latch.await(100, TimeUnit.MILLISECONDS);
        //
        assertEquals(names.size(), onNext.get());
        //
        assertNull(onError.get());
        assertEquals(1, onCompleted.get());
    }

    @Test
    public void testSubscribeFromInfiniteNats() throws Exception {
        Observable<Integer> observable = Observable
// 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/main/java/com/plexobject/rx/util/NatsSpliterator.java
// public class NatsSpliterator implements Spliterator<Integer>, Streamable<Integer> {
//     private int number;
// 
//     public NatsSpliterator(int from) {
//         this.number = from;
//     }
// 
//     @Override
//     public boolean tryAdvance(Consumer<? super Integer> action) {
//         action.accept(number++);
//         return true;
//     }
// 
//     @Override
//     public Spliterator<Integer> trySplit() {
//         return null;
//     }
// 
//     @Override
//     public long estimateSize() {
//         return Integer.MAX_VALUE;
//     }
// 
//     @Override
//     public int characteristics() {
//         return Spliterator.IMMUTABLE;
//     }
// 
//     public Stream<Integer> getStream() {
//         return StreamSupport.stream(this, false);
//     }
// }


// Path: src/test/java/com/plexobject/rx/ObservableTest.java
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import org.junit.Test;
import com.plexobject.rx.scheduler.Scheduler;
import com.plexobject.rx.util.NatsSpliterator;

        initLatch(names.size() + 1); // N*onNext + onCompleted

        Observable<String> observable = Observable.from(names.stream());
        setupCallback(observable, null, true);
        latch.await(100, TimeUnit.MILLISECONDS);
        //
        assertEquals(names.size(), onNext.get());
        //
        assertNull(onError.get());
        assertEquals(1, onCompleted.get());
    }

    @Test
    public void testSubscribeFromIterator() throws Exception {
        initLatch(names.size() + 1); // N*onNext + onCompleted

        Observable<String> observable = Observable.from(names.iterator());

        setupCallback(observable, null, true);
        latch.await(100, TimeUnit.MILLISECONDS);
        //
        assertEquals(names.size(), onNext.get());
        //
        assertNull(onError.get());
        assertEquals(1, onCompleted.get());
    }

    @Test
    public void testSubscribeFromInfiniteNats() throws Exception {
        Observable<Integer> observable = Observable
                .from(new NatsSpliterator(0));