// 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 java.util.function.Consumer;
import mockit.Mocked;
import mockit.Verifications;
import mockit.integration.junit4.JMockit;
import org.junit.Test;
import org.junit.runner.RunWith;
import com.plexobject.rx.scheduler.Scheduler;
    @Mocked
    private Consumer<Throwable> onError;
    @Mocked
    private OnCompletion onCompleted;

    @Test
    public void testSubscribeCreate() throws Exception {
        Observable<Integer> observable = Observable.create(observer -> {
            observer.onNext(1);
            observer.onNext(2);
            observer.onNext(3);
            observer.onCompleted();
        });

        observable.subscribe(onNext, onError, onCompleted);
        new Verifications() {
            {
                onNext.accept(1);
                onNext.accept(2);
                onNext.accept(3);
                onCompleted.onCompleted();
                onError.accept((Throwable) any);
                times = 0;
            }
        };
    }

    @Test
    public void testSubscribeCreateAsync() throws Exception {
        Observable<Integer> observable = Observable.create(observer -> {
// 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/ObservableCreateTest.java
import java.util.function.Consumer;
import mockit.Mocked;
import mockit.Verifications;
import mockit.integration.junit4.JMockit;
import org.junit.Test;
import org.junit.runner.RunWith;
import com.plexobject.rx.scheduler.Scheduler;

    @Mocked
    private Consumer<Throwable> onError;
    @Mocked
    private OnCompletion onCompleted;

    @Test
    public void testSubscribeCreate() throws Exception {
        Observable<Integer> observable = Observable.create(observer -> {
            observer.onNext(1);
            observer.onNext(2);
            observer.onNext(3);
            observer.onCompleted();
        });

        observable.subscribe(onNext, onError, onCompleted);
        new Verifications() {
            {
                onNext.accept(1);
                onNext.accept(2);
                onNext.accept(3);
                onCompleted.onCompleted();
                onError.accept((Throwable) any);
                times = 0;
            }
        };
    }

    @Test
    public void testSubscribeCreateAsync() throws Exception {
        Observable<Integer> observable = Observable.create(observer -> {
            Scheduler.newNewThreadScheduler().scheduleBackgroundTask((o) -> {