@Test public void testIndependentStates() { int count = 100; final ConcurrentHashMap<Object, Object> states = new ConcurrentHashMap<Object, Object>(); AbstractOnSubscribe<Integer, Void> aos = new AbstractOnSubscribe<Integer, Void>() { @Override protected void next(SubscriptionState<Integer, Void> state) { states.put(state, state); state.stop(); } }; Observable<Integer> source = aos.toObservable(); for (int i = 0; i < count; i++) { source.subscribe(); } assertEquals(count, states.size()); }
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);  }