Hide Buffer/Blocking SubscribeOn Behavior

- I want to have it for internal usage but am not ready to publicly expose it.

Author: benjchristensen

rxjava-core/src/main/java/rx/Observable.java

@@ -7055,32 +7055,9 @@ public final Subscription subscribe(Subscriber<? super T> observer, Scheduler sc
      * @see #subscribeOn(rx.Scheduler, int) 
      */
     public final Observable<T> subscribeOn(Scheduler scheduler) {
-        return nest().lift(new OperatorSubscribeOn<T>(scheduler, false));
+        return nest().lift(new OperatorSubscribeOn<T>(scheduler));
     }
 
-    /**
-     * Asynchronously subscribes and unsubscribes Observers to this Observable on the specified {@link Scheduler}
-     * and allows buffering some events emitted from the source in the time gap between the original and
-     * actual subscription, and any excess events will block the source until the actual subscription happens.
-     * <p>
-     * This overload should help mitigate issues when subscribing to a PublishSubject (and derivatives
-     * such as GroupedObservable in operator groupBy) and events fired between the original and actual subscriptions
-     * are lost. 
-     * <p>
-     * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/subscribeOn.png">
-     * 
-     * @param scheduler
-     *            the {@link Scheduler} to perform subscription and unsubscription actions on
-     * @param bufferSize the number of events to buffer before blocking the source while in the time gap,
-     *                   negative value indicates an unlimited buffer
-     * @return the source Observable modified so that its subscriptions and unsubscriptions happen
-     *         on the specified {@link Scheduler}
-     * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#wiki-subscribeon">RxJava Wiki: subscribeOn()</a>
-     */
-    public final Observable<T> subscribeOn(Scheduler scheduler, int bufferSize) {
-        return nest().lift(new OperatorSubscribeOn<T>(scheduler, true, bufferSize));
-    }
-
     /**
      * Returns an Observable that extracts a Double from each of the items emitted by the source
      * Observable via a function you specify, and then emits the sum of these Doubles.

rxjava-core/src/main/java/rx/operators/OperatorSubscribeOn.java

@@ -20,8 +20,6 @@
 import rx.Scheduler;
 import rx.Scheduler.Inner;
 import rx.Subscriber;
-import rx.observables.GroupedObservable;
-import rx.subjects.PublishSubject;
 import rx.util.functions.Action1;
 
 /**
@@ -46,24 +44,24 @@ public class OperatorSubscribeOn<T> implements Operator<T, Observable<T>> {
     /** The buffer size to avoid flooding. Negative value indicates an unbounded buffer. */
     private final int bufferSize;
 
-    public OperatorSubscribeOn(Scheduler scheduler, boolean dontLoseEvents) {
-        this(scheduler, dontLoseEvents, -1);
+    public OperatorSubscribeOn(Scheduler scheduler) {
+        this.scheduler = scheduler;
+        this.dontLoseEvents = false;
+        this.bufferSize = -1;
     }
 
     /**
      * Construct a SubscribeOn operator.
      * 
      * @param scheduler
      *            the target scheduler
-     * @param dontLoseEvents
-     *            indicate that events should be buffered until the actual subscription happens
      * @param bufferSize
      *            if dontLoseEvents == true, this indicates the buffer size. Filling the buffer will
      *            block the source. -1 indicates an unbounded buffer
      */
-    public OperatorSubscribeOn(Scheduler scheduler, boolean dontLoseEvents, int bufferSize) {
+    public OperatorSubscribeOn(Scheduler scheduler, int bufferSize) {
         this.scheduler = scheduler;
-        this.dontLoseEvents = dontLoseEvents;
+        this.dontLoseEvents = true;
         this.bufferSize = bufferSize;
     }
 
@@ -82,18 +80,7 @@ public void onError(Throwable e) {
             }
 
             boolean checkNeedBuffer(Observable<?> o) {
-                /*
-                 * Included are some Observable types known to be "hot" and thus needing
-                 * buffering when subscribing across thread boundaries otherwise
-                 * we can lose data.
-                 * 
-                 * See https://github.com/Netflix/RxJava/issues/844 for more information.
-                 */
-                return dontLoseEvents
-                        || ((o instanceof GroupedObservable<?, ?>)
-                        || (o instanceof PublishSubject<?>)
-                        // || (o instanceof BehaviorSubject<?, ?>)
-                        );
+                return dontLoseEvents;
             }
 
             @Override

rxjava-core/src/test/java/rx/operators/OperatorGroupByTest.java

@@ -700,7 +700,7 @@ public void call() {
 
                             });
                 } else {
-                    return group.subscribeOn(Schedulers.newThread(), 1).delay(400, TimeUnit.MILLISECONDS).map(new Func1<Integer, String>() {
+                    return group.nest().lift(new OperatorSubscribeOn<Integer>(Schedulers.newThread(), 1)).delay(400, TimeUnit.MILLISECONDS).map(new Func1<Integer, String>() {
 
                         @Override
                         public String call(Integer t1) {
@@ -826,7 +826,7 @@ public Integer call(Integer t) {
 
             @Override
             public Observable<String> call(final GroupedObservable<Integer, Integer> group) {
-                return group.subscribeOn(Schedulers.newThread(), 0).map(new Func1<Integer, String>() {
+                return group.nest().lift(new OperatorSubscribeOn<Integer>(Schedulers.newThread(), 0)).map(new Func1<Integer, String>() {
 
                     @Override
                     public String call(Integer t1) {

rxjava-core/src/test/java/rx/operators/OperatorSubscribeOnTest.java

@@ -186,7 +186,7 @@ public Subscription schedule(final Action1<Scheduler.Inner> action, final long d
     public void testSubscribeOnPublishSubjectWithSlowScheduler() {
         PublishSubject<Integer> ps = PublishSubject.create();
         TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
-        ps.subscribeOn(new SlowScheduler()).subscribe(ts);
+        ps.nest().lift(new OperatorSubscribeOn<Integer>(new SlowScheduler(), 0)).subscribe(ts);
         ps.onNext(1);
         ps.onNext(2);
         ps.onCompleted();
@@ -220,7 +220,7 @@ public Integer call(Integer t) {
 
             @Override
             public Observable<String> call(final GroupedObservable<Integer, Integer> group) {
-                return group.subscribeOn(Schedulers.newThread()).map(new Func1<Integer, String>() {
+                return group.nest().lift(new OperatorSubscribeOn<Integer>(Schedulers.newThread(), 0)).map(new Func1<Integer, String>() {
 
                     @Override
                     public String call(Integer t1) {
@@ -326,8 +326,8 @@ void testBoundedBufferingWithSize(int size) throws Exception {
 
         final List<Long> deltas = Collections.synchronizedList(new ArrayList<Long>());
 
-        Subscription s = timer.timestamp().subscribeOn(
-                new SlowScheduler(Schedulers.computation(), 1, TimeUnit.SECONDS), size).map(new Func1<Timestamped<Long>, Long>() {
+        Subscription s = timer.timestamp().nest().lift(new OperatorSubscribeOn<Timestamped<Long>>(
+                new SlowScheduler(Schedulers.computation(), 1, TimeUnit.SECONDS), size)).map(new Func1<Timestamped<Long>, Long>() {
             @Override
             public Long call(Timestamped<Long> t1) {
                 long v = System.currentTimeMillis() - t1.getTimestampMillis();