Proposed solution to the time gap, using unbounded buffering.

Author: akarnokd

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

@@ -7062,7 +7062,25 @@ public final Subscription subscribe(Subscriber<? super T> observer, Scheduler sc
      * @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) {
-        return nest().lift(new OperatorSubscribeOn<T>(scheduler));
+        return nest().lift(new OperatorSubscribeOn<T>(scheduler, false));
+    }
+    /**
+     * Asynchronously subscribes and unsubscribes Observers to this Observable on the specified {@link Scheduler}
+     * and allows buffering the events emitted from the source in the time gap between the original and
+     * actual subscription.
+     * <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 dontLoseEvents indicate that the operator should buffer events emitted in the time gap
+     *                       between the original and actual subscription and replay it to Observers
+     * @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, boolean dontLoseEvents) {
+        return nest().lift(new OperatorSubscribeOn<T>(scheduler, dontLoseEvents));
     }
 
     /**

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

@@ -0,0 +1,150 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import java.util.LinkedList;
+import java.util.Queue;
+import rx.Subscriber;
+import rx.subscriptions.CompositeSubscription;
+
+/**
+ * Buffers the incoming events until notified, then replays the
+ * buffered events and continues as a simple pass-through subscriber.
+ * @param <T> the streamed value type
+ */
+public class BufferUntilSubscriber<T> extends Subscriber<T> {
+    /** The actual subscriber. */
+    private final Subscriber<? super T> actual;
+    /** The mutual exclusion for the duration of the replay. */
+    private final Object gate = new Object();
+    /** Queued events. */
+    private final Queue<Object> queue = new LinkedList<Object>();
+    /** Indicate the pass-through mode. */
+    private volatile boolean passthroughMode;
+    /** Null sentinel (in case queue type is changed). */
+    private static final Object NULL_SENTINEL = new Object();
+    /** Complete sentinel. */
+    private static final Object COMPLETE_SENTINEL = new Object();
+    /**
+     * Container for an onError event.
+     */
+    private static final class ErrorSentinel {
+        final Throwable t;
+
+        public ErrorSentinel(Throwable t) {
+            this.t = t;
+        }
+        
+    }
+    /**
+     * Constructor that wraps the actual subscriber and shares its subscription.
+     * @param actual 
+     */
+    public BufferUntilSubscriber(Subscriber<? super T> actual) {
+        super(actual);
+        this.actual = actual;
+    }
+    /**
+     * Constructor that wraps the actual subscriber and uses the given composite
+     * subscription.
+     * @param actual
+     * @param cs 
+     */
+    public BufferUntilSubscriber(Subscriber<? super T> actual, CompositeSubscription cs) {
+        super(cs);
+        this.actual = actual;
+    }
+    
+    /**
+     * Call this method to replay the buffered events and continue as a pass-through subscriber.
+     * If already in pass-through mode, this method is a no-op.
+     */
+    public void enterPassthroughMode() {
+        if (!passthroughMode) {
+            synchronized (gate) {
+                if (!passthroughMode) {
+                    while (!queue.isEmpty()) {
+                        Object o = queue.poll();
+                        
+                        if (o == NULL_SENTINEL) {
+                            actual.onNext(null);
+                        } else
+                        if (o == COMPLETE_SENTINEL) {
+                            actual.onCompleted();
+                        } else
+                        if (o instanceof ErrorSentinel) {
+                            actual.onError(((ErrorSentinel)o).t);
+                        } else
+                        if (o != null) {
+                            @SuppressWarnings("unchecked")
+                            T v = (T)o;
+                            actual.onNext(v);
+                        } else {
+                            throw new NullPointerException();
+                        }
+                    }
+                    /* Test artificial back-pressure.
+                    try {
+                        TimeUnit.SECONDS.sleep(2);
+                    } catch (Throwable t) {
+                        
+                    }
+                    */
+                    passthroughMode = true;
+                }
+            }
+        }
+    }
+    
+    @Override
+    public void onNext(T t) {
+        if (!passthroughMode) {
+            synchronized (gate) {
+                if (!passthroughMode) {
+                    queue.offer(t != null ? t : NULL_SENTINEL);
+                    return;
+                }
+            }
+        }
+        actual.onNext(t);
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        if (!passthroughMode) {
+            synchronized (gate) {
+                if (!passthroughMode) {
+                    queue.offer(new ErrorSentinel(e));
+                    return;
+                }
+            }
+        }
+        actual.onError(e);
+    }
+
+    @Override
+    public void onCompleted() {
+        if (!passthroughMode) {
+            synchronized (gate) {
+                if (!passthroughMode) {
+                    queue.offer(COMPLETE_SENTINEL);
+                    return;
+                }
+            }
+        }
+        actual.onCompleted();
+    }
+}

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

@@ -20,6 +20,8 @@
 import rx.Scheduler;
 import rx.Scheduler.Inner;
 import rx.Subscriber;
+import rx.observables.GroupedObservable;
+import rx.subjects.PublishSubject;
 import rx.subscriptions.CompositeSubscription;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action0;
@@ -33,9 +35,15 @@
 public class OperatorSubscribeOn<T> implements Operator<T, Observable<T>> {
 
     private final Scheduler scheduler;
+    /** 
+     * Indicate that events fired between the original subscription time and
+     * the actual subscription time should not get lost.
+     */
+    private final boolean dontLoseEvents;
 
-    public OperatorSubscribeOn(Scheduler scheduler) {
+    public OperatorSubscribeOn(Scheduler scheduler, boolean dontLoseEvents) {
         this.scheduler = scheduler;
+        this.dontLoseEvents = dontLoseEvents;
     }
 
     @Override
@@ -51,9 +59,40 @@ public void onCompleted() {
             public void onError(Throwable e) {
                 subscriber.onError(e);
             }
-
+            boolean checkNeedBuffer(Observable<?> o) {
+                return (o instanceof GroupedObservable<?, ?>)
+                        || (o instanceof PublishSubject<?>)
+                        // || (o instanceof BehaviorSubject<?, ?>)
+                        ;
+            }
             @Override
             public void onNext(final Observable<T> o) {
+                if (dontLoseEvents || checkNeedBuffer(o)) {
+                    final CompositeSubscription cs = new CompositeSubscription();
+                    subscriber.add(cs);
+                    final BufferUntilSubscriber<T> bus = new BufferUntilSubscriber<T>(subscriber, new CompositeSubscription());
+                    o.subscribe(bus);
+                    scheduler.schedule(new Action1<Inner>() {
+
+                        @Override
+                        public void call(final Inner inner) {
+                            cs.add(Subscriptions.create(new Action0() {
+                                @Override
+                                public void call() {
+                                    inner.schedule(new Action1<Inner>() {
+                                        @Override
+                                        public void call(final Inner inner) {
+                                            bus.unsubscribe();
+                                        }
+                                    });
+                                }
+                            }));
+                            bus.enterPassthroughMode();
+                        }
+                        
+                    });
+                    return;
+                }
                 scheduler.schedule(new Action1<Inner>() {
 
                     @Override