Merge branch 'master' into sample

Author: jmhofer

CHANGES.md

@@ -1,5 +1,18 @@
 # RxJava Releases #
 
+### Version 0.8.0 ([Maven Central](http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22com.netflix.rxjava%22%20AND%20v%3A%220.8.0%22)) ###
+
+This is a breaking (non-backwards compatible) release that updates the Scheduler implementation released in 0.7.0.
+
+See See https://github.com/Netflix/RxJava/issues/19 for background, discussion and status of Schedulers.
+
+It is believed that the public signatures of Scheduler and related objects is now stabilized but ongoing feedback and review by the community could still result in changes.
+
+* [Issue 19](https://github.com/Netflix/RxJava/issues/19) Schedulers improvements, changes and additions
+* [Issue 202](https://github.com/Netflix/RxJava/issues/202) Fix Concat bugs
+* [Issue 65](https://github.com/Netflix/RxJava/issues/65) Multicast
+* [Pull 218](https://github.com/Netflix/RxJava/pull/218) ReplaySubject
+
 ### Version 0.7.0 ([Maven Central](http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22com.netflix.rxjava%22%20AND%20v%3A%220.7.0%22)) ###
 
 This release adds the foundations of Rx Schedulers.

gradle.properties

@@ -1 +1 @@
-version=0.7.1-SNAPSHOT
+version=0.8.1-SNAPSHOT

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

@@ -36,18 +36,21 @@
 import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
 
+import rx.observables.ConnectableObservable;
 import rx.observables.GroupedObservable;
 import rx.operators.OperationAll;
 import rx.operators.OperationConcat;
 import rx.operators.OperationDefer;
 import rx.operators.OperationDematerialize;
+import rx.operators.OperationGroupBy;
 import rx.operators.OperationFilter;
 import rx.operators.OperationFinally;
 import rx.operators.OperationMap;
 import rx.operators.OperationMaterialize;
 import rx.operators.OperationMerge;
 import rx.operators.OperationMergeDelayError;
 import rx.operators.OperationMostRecent;
+import rx.operators.OperationMulticast;
 import rx.operators.OperationNext;
 import rx.operators.OperationObserveOn;
 import rx.operators.OperationOnErrorResumeNextViaFunction;
@@ -59,19 +62,19 @@
 import rx.operators.OperationSynchronize;
 import rx.operators.OperationTake;
 import rx.operators.OperationTakeLast;
+import rx.operators.OperationTakeUntil;
 import rx.operators.OperationTakeWhile;
+import rx.operators.OperationToIterator;
 import rx.operators.OperationToObservableFuture;
 import rx.operators.OperationToObservableIterable;
 import rx.operators.OperationToObservableList;
 import rx.operators.OperationToObservableSortedList;
 import rx.operators.OperationWhere;
 import rx.operators.OperationZip;
-import rx.operators.OperatorGroupBy;
-import rx.operators.OperatorTakeUntil;
-import rx.operators.OperatorToIterator;
 import rx.plugins.RxJavaErrorHandler;
 import rx.plugins.RxJavaObservableExecutionHook;
 import rx.plugins.RxJavaPlugins;
+import rx.subjects.Subject;
 import rx.subscriptions.BooleanSubscription;
 import rx.subscriptions.Subscriptions;
 import rx.util.AtomicObservableSubscription;
@@ -585,6 +588,17 @@ public void call(Object args) {
         });
     }
 
+    /**
+     * Returns a connectable observable sequence that upon connection causes the source sequence to push results into the specified subject.
+     *
+     * @param subject the subject to push source elements into.
+     * @param <R> result type
+     * @return a connectable observable sequence that upon connection causes the source sequence to push results into the specified subject.
+     */
+    public <R> ConnectableObservable<R> multicast(Subject<T, R> subject) {
+        return multicast(this, subject);
+    }
+
     /**
      * Returns the only element of an observable sequence and throws an exception if there is not exactly one element in the observable sequence.
      * 
@@ -1303,7 +1317,7 @@ public static <T> Observable<T> merge(Observable<T>... source) {
      * @return An observable sequence containing the elements of the source sequence up to the point the other sequence interrupted further propagation.
      */
     public static <T, E> Observable<T> takeUntil(final Observable<T> source, final Observable<E> other) {
-        return OperatorTakeUntil.takeUntil(source, other);
+        return OperationTakeUntil.takeUntil(source, other);
     }
 
     /**
@@ -1355,7 +1369,7 @@ public static <T> Observable<T> finallyDo(Observable<T> source, Action0 action)
      * @return an observable of observable groups, each of which corresponds to a unique key value, containing all elements that share that same key value.
      */
     public static <K, T, R> Observable<GroupedObservable<K, R>> groupBy(Observable<T> source, final Func1<T, K> keySelector, final Func1<T, R> elementSelector) {
-        return create(OperatorGroupBy.groupBy(source, keySelector, elementSelector));
+        return create(OperationGroupBy.groupBy(source, keySelector, elementSelector));
     }
 
     /**
@@ -1372,7 +1386,7 @@ public static <K, T, R> Observable<GroupedObservable<K, R>> groupBy(Observable<T
      * @return an observable of observable groups, each of which corresponds to a unique key value, containing all elements that share that same key value.
      */
     public static <K, T> Observable<GroupedObservable<K, T>> groupBy(Observable<T> source, final Func1<T, K> keySelector) {
-        return create(OperatorGroupBy.groupBy(source, keySelector));
+        return create(OperationGroupBy.groupBy(source, keySelector));
     }
 
     /**
@@ -2041,7 +2055,7 @@ public Iterator<T> iterator() {
      * @return the iterator that could be used to iterate over the elements of the observable.
      */
     public static <T> Iterator<T> getIterator(Observable<T> that) {
-        return OperatorToIterator.toIterator(that);
+        return OperationToIterator.toIterator(that);
     }
 
     /**
@@ -2072,9 +2086,22 @@ public static <T> Iterable<T> mostRecent(Observable<T> source, T initialValue) {
         return OperationMostRecent.mostRecent(source, initialValue);
     }
 
+    /**
+     * Returns a connectable observable sequence that upon connection causes the source sequence to push results into the specified subject.
+     *
+     * @param source the source sequence whose elements will be pushed into the specified subject.
+     * @param subject the subject to push source elements into.
+     * @param <T> source type
+     * @param <R> result type
+     * @return a connectable observable sequence that upon connection causes the source sequence to push results into the specified subject.
+     */
+    public static <T, R> ConnectableObservable<R> multicast(Observable<T> source, final Subject<T, R> subject) {
+        return OperationMulticast.multicast(source, subject);
+    }
+
     /**
      * Returns the only element of an observable sequence and throws an exception if there is not exactly one element in the observable sequence.
-     * 
+     *
      * @param that
      *            the source Observable
      * @return The single element in the observable sequence.

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

@@ -15,15 +15,99 @@
  */
 package rx;
 
+import java.util.Date;
 import java.util.concurrent.TimeUnit;
 
+import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action0;
 import rx.util.functions.Func0;
+import rx.util.functions.Func1;
+import rx.util.functions.Func2;
 
 /**
  * Represents an object that schedules units of work.
+ * <p>
+ * The methods left to implement are:
+ * <ul>
+ * <li>{@code <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, long delayTime, TimeUnit unit)}</li>
+ * <li>{@code <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action)}</li>
+ * </ul>
+ * <p>
+ * Why is this an abstract class instead of an interface?
+ * <p>
+ * <ol>
+ * <li>Java doesn't support extension methods and there are many overload methods needing default implementations.</li>
+ * <li>Virtual extension methods aren't available until Java8 which RxJava will not set as a minimum target for a long time.</li>
+ * <li>If only an interface were used Scheduler implementations would then need to extend from an AbstractScheduler pair that gives all of the functionality unless they intend on copy/pasting the
+ * functionality.</li>
+ * <li>Without virtual extension methods even additive changes are breaking and thus severely impede library maintenance.</li>
+ * </ol>
  */
-public interface Scheduler {
+public abstract class Scheduler {
+
+    /**
+     * Schedules a cancelable action to be executed.
+     * 
+     * @param state
+     *            State to pass into the action.
+     * @param action
+     *            Action to schedule.
+     * @return a subscription to be able to unsubscribe from action.
+     */
+    public abstract <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action);
+
+    /**
+     * Schedules a cancelable action to be executed in delayTime.
+     * 
+     * @param state
+     *            State to pass into the action.
+     * @param action
+     *            Action to schedule.
+     * @param delayTime
+     *            Time the action is to be delayed before executing.
+     * @param unit
+     *            Time unit of the delay time.
+     * @return a subscription to be able to unsubscribe from action.
+     */
+    public abstract <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, long delayTime, TimeUnit unit);
+
+    /**
+     * Schedules a cancelable action to be executed at dueTime.
+     * 
+     * @param state
+     *            State to pass into the action.
+     * @param action
+     *            Action to schedule.
+     * @param dueTime
+     *            Time the action is to be executed. If in the past it will be executed immediately.
+     * @return a subscription to be able to unsubscribe from action.
+     */
+    public <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, Date dueTime) {
+        long scheduledTime = dueTime.getTime();
+        long timeInFuture = scheduledTime - now();
+        if (timeInFuture <= 0) {
+            return schedule(state, action);
+        } else {
+            return schedule(state, action, timeInFuture, TimeUnit.MILLISECONDS);
+        }
+    }
+
+    /**
+     * Schedules a cancelable action to be executed.
+     * 
+     * @param action
+     *            Action to schedule.
+     * @return a subscription to be able to unsubscribe from action.
+     */
+    public Subscription schedule(final Func1<Scheduler, Subscription> action) {
+        return schedule(null, new Func2<Scheduler, Void, Subscription>() {
+
+            @Override
+            public Subscription call(Scheduler scheduler, Void t2) {
+                return action.call(scheduler);
+            }
+        });
+    }
 
     /**
      * Schedules a cancelable action to be executed.
@@ -32,7 +116,15 @@ public interface Scheduler {
      *            action
      * @return a subscription to be able to unsubscribe from action.
      */
-    Subscription schedule(Func0<Subscription> action);
+    public Subscription schedule(final Func0<Subscription> action) {
+        return schedule(null, new Func2<Scheduler, Void, Subscription>() {
+
+            @Override
+            public Subscription call(Scheduler scheduler, Void t2) {
+                return action.call();
+            }
+        });
+    }
 
     /**
      * Schedules an action to be executed.
@@ -41,29 +133,78 @@ public interface Scheduler {
      *            action
      * @return a subscription to be able to unsubscribe from action.
      */
-    Subscription schedule(Action0 action);
+    public Subscription schedule(final Action0 action) {
+        return schedule(null, new Func2<Scheduler, Void, Subscription>() {
+
+            @Override
+            public Subscription call(Scheduler scheduler, Void t2) {
+                action.call();
+                return Subscriptions.empty();
+            }
+        });
+    }
 
     /**
-     * Schedules an action to be executed in dueTime.
+     * Schedules a cancelable action to be executed in delayTime.
+     * 
+     * @param action
+     *            Action to schedule.
+     * @param delayTime
+     *            Time the action is to be delayed before executing.
+     * @param unit
+     *            Time unit of the delay time.
+     * @return a subscription to be able to unsubscribe from action.
+     */
+    public Subscription schedule(final Func1<Scheduler, Subscription> action, long delayTime, TimeUnit unit) {
+        return schedule(null, new Func2<Scheduler, Void, Subscription>() {
+
+            @Override
+            public Subscription call(Scheduler scheduler, Void t2) {
+                return action.call(scheduler);
+            }
+        }, delayTime, unit);
+    }
+
+    /**
+     * Schedules an action to be executed in delayTime.
      * 
      * @param action
      *            action
      * @return a subscription to be able to unsubscribe from action.
      */
-    Subscription schedule(Action0 action, long dueTime, TimeUnit unit);
+    public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit) {
+        return schedule(null, new Func2<Scheduler, Void, Subscription>() {
+
+            @Override
+            public Subscription call(Scheduler scheduler, Void t2) {
+                action.call();
+                return Subscriptions.empty();
+            }
+        }, delayTime, unit);
+    }
 
     /**
-     * Schedules a cancelable action to be executed in dueTime.
+     * Schedules a cancelable action to be executed in delayTime.
      * 
      * @param action
      *            action
      * @return a subscription to be able to unsubscribe from action.
      */
-    Subscription schedule(Func0<Subscription> action, long dueTime, TimeUnit unit);
+    public Subscription schedule(final Func0<Subscription> action, long delayTime, TimeUnit unit) {
+        return schedule(null, new Func2<Scheduler, Void, Subscription>() {
+
+            @Override
+            public Subscription call(Scheduler scheduler, Void t2) {
+                return action.call();
+            }
+        }, delayTime, unit);
+    }
 
     /**
-     * Returns the scheduler's notion of current time.
+     * Returns the scheduler's notion of current absolute time in milliseconds.
      */
-    long now();
+    public long now() {
+        return System.currentTimeMillis();
+    }
 
 }