subscribeOn + groupBy

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

@@ -7052,9 +7052,33 @@ public final Subscription subscribe(Subscriber<? super T> observer, Scheduler sc
      * @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>
+     * @see #subscribeOn(rx.Scheduler, int) 
      */
     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 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));
     }
 
     /**

rxjava-core/src/main/java/rx/observers/TestSubscriber.java

@@ -101,6 +101,14 @@ public void awaitTerminalEvent(long timeout, TimeUnit unit) {
         }
     }
 
+    public void awaitTerminalEventAndUnsubscribeOnTimeout(long timeout, TimeUnit unit) {
+        try {
+            awaitTerminalEvent(timeout, unit);
+        } catch (RuntimeException e) {
+            unsubscribe();
+        }
+    }
+
     public Thread getLastSeenThread() {
         return lastSeenThread;
     }

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

@@ -0,0 +1,193 @@
+/**
+ * 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;
+    /** Indicate the pass-through mode. */
+    private volatile boolean passthroughMode;
+    /** Protect mode transition. */
+    private final Object gate = new Object();
+    /** The buffered items. */
+    private final Queue<Object> queue = new LinkedList<Object>();
+    /** The queue capacity. */
+    private final int capacity;
+    /** 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 capacity the queue capacity to accept before blocking, negative value indicates an unbounded queue
+     * @param actual
+     */
+    public BufferUntilSubscriber(int capacity, Subscriber<? super T> actual) {
+        super(actual);
+        this.actual = actual;
+        this.capacity = capacity;
+    }
+    /**
+     * Constructor that wraps the actual subscriber and uses the given composite
+     * subscription.
+     * @param capacity the queue capacity to accept before blocking, negative value indicates an unbounded queue
+     * @param actual
+     * @param cs 
+     */
+    public BufferUntilSubscriber(int capacity, Subscriber<? super T> actual, CompositeSubscription cs) {
+        super(cs);
+        this.actual = actual;
+        this.capacity = capacity;
+    }
+
+    /**
+     * 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 (!actual.isUnsubscribed()) {
+                            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();
+                            }
+                        }
+                    }
+                    passthroughMode = true;
+                    gate.notifyAll();
+                }
+            }
+        }
+    }
+    @Override
+    public void onNext(T t) {
+        if (!passthroughMode) {
+            synchronized (gate) {
+                if (!passthroughMode) {
+                    if (capacity < 0 || queue.size() < capacity) {
+                        queue.offer(t != null ? t : NULL_SENTINEL);
+                        return;
+                    }
+                    try {
+                        while (!passthroughMode) {
+                            gate.wait();
+                        }
+                        if (actual.isUnsubscribed()) {
+                            return;
+                        }
+                    } catch (InterruptedException ex) {
+                        Thread.currentThread().interrupt();
+                        actual.onError(ex);
+                        return;
+                    }
+                }
+            }
+        }
+        actual.onNext(t);
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        if (!passthroughMode) {
+            synchronized (gate) {
+                if (!passthroughMode) {
+                    if (capacity < 0 || queue.size() < capacity) {
+                        queue.offer(new ErrorSentinel(e));
+                        return;
+                    }
+                    try {
+                        while (!passthroughMode) {
+                            gate.wait();
+                        }
+                        if (actual.isUnsubscribed()) {
+                            return;
+                        }
+                    } catch (InterruptedException ex) {
+                        Thread.currentThread().interrupt();
+                        actual.onError(ex);
+                        return;
+                    }
+                }
+            }
+        }
+        actual.onError(e);
+    }
+
+    @Override
+    public void onCompleted() {
+        if (!passthroughMode) {
+            synchronized (gate) {
+                if (!passthroughMode) {
+                    if (capacity < 0 || queue.size() < capacity) {
+                        queue.offer(COMPLETE_SENTINEL);
+                        return;
+                    }
+                    try {
+                        while (!passthroughMode) {
+                            gate.wait();
+                        }
+                        if (actual.isUnsubscribed()) {
+                            return;
+                        }
+                    } catch (InterruptedException ex) {
+                        Thread.currentThread().interrupt();
+                        actual.onError(ex);
+                        return;
+                    }
+                }
+            }
+        }
+        actual.onCompleted();
+    }
+
+}