@Test(timeout = 3000) public void testSubscribeOn() { final int start = 1; final int count = 100; AbstractOnSubscribe<Integer, Void> aos = new AbstractOnSubscribe<Integer, Void>() { @Override protected void next(SubscriptionState<Integer, Void> state) { long calls = state.calls(); if (calls <= count) { state.onNext((int)calls + start); if (calls == count) { state.onCompleted(); } } } }; @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); InOrder inOrder = inOrder(o); TestSubscriber<Object> ts = new TestSubscriber<Object>(o); aos.toObservable().subscribeOn(Schedulers.newThread()).subscribe(ts); ts.awaitTerminalEvent(); verify(o, never()).onError(any(Throwable.class)); for (int i = 1; i <= count; i++) { inOrder.verify(o).onNext(i); } inOrder.verify(o).onCompleted(); inOrder.verifyNoMoreInteractions(); }
public final Observable<T> toObservable() { return Observable.create(this); }
AbstractOnSubscribe implements OnSubscribe<T> { public final Observable<T> toObservable() { return Observable.create(this); } }
AbstractOnSubscribe implements OnSubscribe<T> { public final Observable<T> toObservable() { return Observable.create(this); }  }
AbstractOnSubscribe implements OnSubscribe<T> { public final Observable<T> toObservable() { return Observable.create(this); }  @Override final void call(final Subscriber<? super T> subscriber); final Observable<T> toObservable(); static AbstractOnSubscribe<T, S> create(Action1<SubscriptionState<T, S>> next); static AbstractOnSubscribe<T, S> create(Action1<SubscriptionState<T, S>> next,
            Func1<? super Subscriber<? super T>, ? extends S> onSubscribe); static AbstractOnSubscribe<T, S> create(Action1<SubscriptionState<T, S>> next,
            Func1<? super Subscriber<? super T>, ? extends S> onSubscribe, Action1<? super S> onTerminated); }
AbstractOnSubscribe implements OnSubscribe<T> { public final Observable<T> toObservable() { return Observable.create(this); }  @Override final void call(final Subscriber<? super T> subscriber); final Observable<T> toObservable(); static AbstractOnSubscribe<T, S> create(Action1<SubscriptionState<T, S>> next); static AbstractOnSubscribe<T, S> create(Action1<SubscriptionState<T, S>> next,
            Func1<? super Subscriber<? super T>, ? extends S> onSubscribe); static AbstractOnSubscribe<T, S> create(Action1<SubscriptionState<T, S>> next,
            Func1<? super Subscriber<? super T>, ? extends S> onSubscribe, Action1<? super S> onTerminated);  }