Merge pull request #1433 from Applied-Duality/async

Take 3

Author: benjchristensen

language-adaptors/rxjava-scala/src/examples/scala/rx/lang/scala/examples/RxScalaDemo.scala

@@ -521,7 +521,7 @@ class RxScalaDemo extends JUnitSuite {
 
   @Test def exampleWithReplay5() {
     val numbers = Observable.interval(100 millis).take(10)
-    val sharedNumbers = numbers.replay[Long, Long]((o: Observable[Long]) => o.map(_ * 2))
+    val sharedNumbers = numbers.replay(o => o.map(_ * 2))
     sharedNumbers.subscribe(n => println(s"subscriber gets $n"))
     waitFor(sharedNumbers)
   }

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/ImplicitFunctionConversions.scala

@@ -47,13 +47,6 @@ object ImplicitFunctionConversions {
       }
     }
 
-  implicit def scalaFunction1ToOnSubscribeFunc[T](f: rx.lang.scala.Observer[T] => Subscription) =
-    new rx.Observable.OnSubscribeFunc[T] {
-      def onSubscribe(obs: rx.Observer[_ >: T]): rx.Subscription = {
-        f(obs)
-      }
-    }
-
   implicit def scalaAction1ToOnSubscribe[T](f: Subscriber[T] => Unit) =
     new rx.Observable.OnSubscribe[T] {
       def call(s: rx.Subscriber[_ >: T]): Unit = {

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/Observable.scala

@@ -1287,10 +1287,10 @@ trait Observable[+T]
    * @return an Observable that emits items that are the results of invoking the selector on a `ConnectableObservable`
    *         that shares a single subscription to the source Observable
    */
-  def replay[U >: T, R](selector: Observable[U] => Observable[R]): Observable[R] = {
-    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
-    val fJava: Func1[rx.Observable[U], rx.Observable[R]] =
-      (jo: rx.Observable[U]) => selector(toScalaObservable[U](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
+  def replay[R](selector: Observable[T] => Observable[R]): Observable[R] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val fJava: Func1[rx.Observable[T], rx.Observable[R]] =
+      (jo: rx.Observable[T]) => selector(toScalaObservable[T](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
     toScalaObservable[R](thisJava.replay(fJava))
   }
 
@@ -1308,10 +1308,10 @@ trait Observable[+T]
    *         a `ConnectableObservable` that shares a single subscription to the source Observable  replaying
    *         no more than `bufferSize` items
    */
-  def replay[U >: T, R](selector: Observable[U] => Observable[R], bufferSize: Int): Observable[R] = {
-    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
-    val fJava: Func1[rx.Observable[U], rx.Observable[R]] =
-      (jo: rx.Observable[U]) => selector(toScalaObservable[U](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
+  def replay[R](selector: Observable[T] => Observable[R], bufferSize: Int): Observable[R] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val fJava: Func1[rx.Observable[T], rx.Observable[R]] =
+      (jo: rx.Observable[T]) => selector(toScalaObservable[T](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
     toScalaObservable[R](thisJava.replay(fJava, bufferSize))
   }
 
@@ -1330,10 +1330,10 @@ trait Observable[+T]
    *         a `ConnectableObservable` that shares a single subscription to the source Observable, and
    *         replays no more than `bufferSize` items that were emitted within the window defined by `time`
    */
-  def replay[U >: T, R](selector: Observable[U] => Observable[R], bufferSize: Int, time: Duration): Observable[R] = {
-    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
-    val fJava: Func1[rx.Observable[U], rx.Observable[R]] =
-      (jo: rx.Observable[U]) => selector(toScalaObservable[U](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
+  def replay[R](selector: Observable[T] => Observable[R], bufferSize: Int, time: Duration): Observable[R] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val fJava: Func1[rx.Observable[T], rx.Observable[R]] =
+      (jo: rx.Observable[T]) => selector(toScalaObservable[T](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
     toScalaObservable[R](thisJava.replay(fJava, bufferSize, time.length, time.unit))
   }
 
@@ -1354,10 +1354,10 @@ trait Observable[+T]
    *         replays no more than `bufferSize` items that were emitted within the window defined by `time`
    * @throws IllegalArgumentException if `bufferSize` is less than zero
    */
-  def replay[U >: T, R](selector: Observable[U] => Observable[R], bufferSize: Int, time: Duration, scheduler: Scheduler): Observable[R] = {
-    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
-    val fJava: Func1[rx.Observable[U], rx.Observable[R]] =
-      (jo: rx.Observable[U]) => selector(toScalaObservable[U](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
+  def replay[R](selector: Observable[T] => Observable[R], bufferSize: Int, time: Duration, scheduler: Scheduler): Observable[R] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val fJava: Func1[rx.Observable[T], rx.Observable[R]] =
+      (jo: rx.Observable[T]) => selector(toScalaObservable[T](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
     toScalaObservable[R](thisJava.replay(fJava, bufferSize, time.length, time.unit, scheduler))
   }
 
@@ -1376,10 +1376,10 @@ trait Observable[+T]
    *         a `ConnectableObservable` that shares a single subscription to the source Observable,
    *         replaying no more than `bufferSize` notifications
    */
-  def replay[U >: T, R](selector: Observable[U] => Observable[R], bufferSize: Int, scheduler: Scheduler): Observable[R] = {
-    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
-    val fJava: Func1[rx.Observable[U], rx.Observable[R]] =
-      (jo: rx.Observable[U]) => selector(toScalaObservable[U](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
+  def replay[R](selector: Observable[T] => Observable[R], bufferSize: Int, scheduler: Scheduler): Observable[R] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val fJava: Func1[rx.Observable[T], rx.Observable[R]] =
+      (jo: rx.Observable[T]) => selector(toScalaObservable[T](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
     toScalaObservable[R](thisJava.replay(fJava, bufferSize, scheduler))
   }
 
@@ -1397,10 +1397,10 @@ trait Observable[+T]
    *         a `ConnectableObservable` that shares a single subscription to the source Observable,
    *         replaying all items that were emitted within the window defined by `time`
    */
-  def replay[U >: T, R](selector: Observable[U] => Observable[R], time: Duration, scheduler: Scheduler): Observable[R] = {
-    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
-    val fJava: Func1[rx.Observable[U], rx.Observable[R]] =
-      (jo: rx.Observable[U]) => selector(toScalaObservable[U](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
+  def replay[R](selector: Observable[T] => Observable[R], time: Duration, scheduler: Scheduler): Observable[R] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val fJava: Func1[rx.Observable[T], rx.Observable[R]] =
+      (jo: rx.Observable[T]) => selector(toScalaObservable[T](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
     toScalaObservable[R](thisJava.replay(fJava, time.length, time.unit, scheduler))
   }
 
@@ -1417,10 +1417,10 @@ trait Observable[+T]
    *         a `ConnectableObservable` that shares a single subscription to the source Observable,
    *         replaying all items
    */
-  def replay[U >: T, R](selector: Observable[U] => Observable[R], scheduler: Scheduler): Observable[R] = {
-    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
-    val fJava: Func1[rx.Observable[U], rx.Observable[R]] =
-      (jo: rx.Observable[U]) => selector(toScalaObservable[U](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
+  def replay[R](selector: Observable[T] => Observable[R], scheduler: Scheduler): Observable[R] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val fJava: Func1[rx.Observable[T], rx.Observable[R]] =
+      (jo: rx.Observable[T]) => selector(toScalaObservable[T](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
     toScalaObservable[R](thisJava.replay(fJava, scheduler))
   }
 
@@ -1470,10 +1470,10 @@ trait Observable[+T]
    *         a `ConnectableObservable` that shares a single subscription to the source Observable,
    *         replaying all items that were emitted within the window defined by `time`
    */
-  def replay[U >: T, R](selector: Observable[U] => Observable[R], time: Duration): Observable[R] = {
-    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
-    val fJava: Func1[rx.Observable[U], rx.Observable[R]] =
-      (jo: rx.Observable[U]) => selector(toScalaObservable[U](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
+  def replay[R](selector: Observable[T] => Observable[R], time: Duration): Observable[R] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val fJava: Func1[rx.Observable[T], rx.Observable[R]] =
+      (jo: rx.Observable[T]) => selector(toScalaObservable[T](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
     toScalaObservable[R](thisJava.replay(fJava, time.length, time.unit))
   }
 
@@ -1626,9 +1626,9 @@ trait Observable[+T]
    * @param initialValue the initial value to be emitted by the resulting Observable
    * @return a `ConnectableObservable` that shares a single subscription to the underlying Observable and starts with `initialValue`
    */
-  def publish[U >: T](initialValue: U): ConnectableObservable[U] = {
-    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
-    new ConnectableObservable[U](thisJava.publish(initialValue))
+  def publish[T](initialValue: T): ConnectableObservable[T] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[T]]
+    new ConnectableObservable[T](thisJava.publish(initialValue))
   }
 
   /**
@@ -1643,10 +1643,10 @@ trait Observable[+T]
    * @return an Observable that emits the results of invoking the selector on the items emitted by a `ConnectableObservable`
    *         that shares a single subscription to the underlying sequence
    */
-  def publish[U >: T, R](selector: Observable[U] => Observable[R]): Observable[R] = {
-    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
-    val fJava: Func1[rx.Observable[U], rx.Observable[R]] =
-      (jo: rx.Observable[U]) => selector(toScalaObservable[U](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
+  def publish[R](selector: Observable[T] => Observable[R]): Observable[R] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val fJava: Func1[rx.Observable[T], rx.Observable[R]] =
+      (jo: rx.Observable[T]) => selector(toScalaObservable[T](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
     toScalaObservable[R](thisJava.publish(fJava))
   }
 
@@ -1664,10 +1664,10 @@ trait Observable[+T]
    * @return an Observable that emits `initialValue` followed by the results of invoking the selector
    *         on a `ConnectableObservable` that shares a single subscription to the underlying Observable
    */
-  def publish[U >: T, R](selector: Observable[U] => Observable[R], initialValue: U): Observable[R] = {
-    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
-    val fJava: Func1[rx.Observable[U], rx.Observable[R]] =
-      (jo: rx.Observable[U]) => selector(toScalaObservable[U](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
+  def publish[R](selector: Observable[T] => Observable[R], initialValue: T @uncheckedVariance): Observable[R] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[T]]
+    val fJava: Func1[rx.Observable[T], rx.Observable[R]] =
+      (jo: rx.Observable[T]) => selector(toScalaObservable[T](jo)).asJavaObservable.asInstanceOf[rx.Observable[R]]
     toScalaObservable[R](thisJava.publish(fJava, initialValue))
   }
 
@@ -1971,7 +1971,7 @@ trait Observable[+T]
    * @see <a href="https://github.com/Netflix/RxJava/wiki/Filtering-Observables#wiki-skipuntil">RxJava Wiki: skipUntil()</a>
    * @see <a href="http://msdn.microsoft.com/en-us/library/hh229358.aspx">MSDN: Observable.SkipUntil</a>
    */
-  def dropUntil[E](other: Observable[E]): Observable[T] = {
+  def dropUntil(other: Observable[Any]): Observable[T] = {
     toScalaObservable[T](asJavaObservable.skipUntil(other))
   }
 
@@ -2129,12 +2129,10 @@ trait Observable[+T]
    * @param that
    *            the Observable whose first emitted item will cause `takeUntil` to stop
    *            emitting items from the source Observable
-   * @tparam E
-   *            the type of items emitted by `other`
-   * @return an Observable that emits the items of the source Observable until such time as
+    * @return an Observable that emits the items of the source Observable until such time as
    *         `other` emits its first item
    */
-  def takeUntil[E](that: Observable[E]): Observable[T] = {
+  def takeUntil(that: Observable[Any]): Observable[T] = {
     toScalaObservable[T](asJavaObservable.takeUntil(that.asJavaObservable))
   }
 
@@ -2732,8 +2730,8 @@ trait Observable[+T]
    *         the source Observable takes longer to arrive than the time window defined by the
    *         selector for the previously emitted item
    */
-  def timeout[V](timeoutSelector: T => Observable[V]): Observable[T] = {
-    toScalaObservable[T](asJavaObservable.timeout({ t: T => timeoutSelector(t).asJavaObservable.asInstanceOf[rx.Observable[V]] }))
+  def timeout(timeoutSelector: T => Observable[Any]): Observable[T] = {
+    toScalaObservable[T](asJavaObservable.timeout({ t: T => timeoutSelector(t).asJavaObservable.asInstanceOf[rx.Observable[Any]] }))
   }
 
   /**
@@ -2755,10 +2753,10 @@ trait Observable[+T]
    *         fallback Observable if a item emitted by the source Observable takes longer to arrive
    *         than the time window defined by the selector for the previously emitted item
    */
-  def timeout[V, O >: T](timeoutSelector: T => Observable[V], other: Observable[O]): Observable[O] = {
-    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[O]]
-    toScalaObservable[O](thisJava.timeout(
-      { t: O => timeoutSelector(t.asInstanceOf[T]).asJavaObservable.asInstanceOf[rx.Observable[V]] },
+  def timeout[U >: T](timeoutSelector: T => Observable[Any], other: Observable[U]): Observable[U] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
+    toScalaObservable[U](thisJava.timeout(
+      { t: U => timeoutSelector(t.asInstanceOf[T]).asJavaObservable.asInstanceOf[rx.Observable[Any]] },
       other.asJavaObservable))
   }
 
@@ -2779,10 +2777,10 @@ trait Observable[+T]
    * @return an Observable that mirrors the source Observable, but emits a TimeoutException if either the first item or any subsequent item doesn't
    *         arrive within the time windows specified by the timeout selectors
    */
-  def timeout[U, V](firstTimeoutSelector: () => Observable[U], timeoutSelector: T => Observable[V]): Observable[T] = {
+  def timeout(firstTimeoutSelector: () => Observable[Any], timeoutSelector: T => Observable[Any]): Observable[T] = {
     toScalaObservable[T](asJavaObservable.timeout(
-      { firstTimeoutSelector().asJavaObservable.asInstanceOf[rx.Observable[U]] },
-      { t: T => timeoutSelector(t).asJavaObservable.asInstanceOf[rx.Observable[V]] }))
+      { firstTimeoutSelector().asJavaObservable.asInstanceOf[rx.Observable[Any]] },
+      { t: T => timeoutSelector(t).asJavaObservable.asInstanceOf[rx.Observable[Any]] }))
   }
 
   /**
@@ -2804,11 +2802,11 @@ trait Observable[+T]
    * @return an Observable that mirrors the source Observable, but switches to the `other` Observable if either the first item emitted by the source Observable or any
    *         subsequent item don't arrive within time windows defined by the timeout selectors
    */
-  def timeout[U, V, O >: T](firstTimeoutSelector: () => Observable[U], timeoutSelector: T => Observable[V], other: Observable[O]): Observable[O] = {
-    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[O]]
-    toScalaObservable[O](thisJava.timeout(
-      { firstTimeoutSelector().asJavaObservable.asInstanceOf[rx.Observable[U]] },
-      { t: O => timeoutSelector(t.asInstanceOf[T]).asJavaObservable.asInstanceOf[rx.Observable[V]] },
+  def timeout[U >: T](firstTimeoutSelector: () => Observable[Any], timeoutSelector: T => Observable[Any], other: Observable[U]): Observable[U] = {
+    val thisJava = this.asJavaObservable.asInstanceOf[rx.Observable[U]]
+    toScalaObservable[U](thisJava.timeout(
+      { firstTimeoutSelector().asJavaObservable.asInstanceOf[rx.Observable[Any]] },
+      { t: U => timeoutSelector(t.asInstanceOf[T]).asJavaObservable.asInstanceOf[rx.Observable[Any]] },
       other.asJavaObservable))
   }
 

rxjava-contrib/rxjava-async-util/src/main/java/rx/util/async/Async.java

@@ -15,49 +15,19 @@
  */
 package rx.util.async;
 
-import java.util.concurrent.Callable;
-import java.util.concurrent.Future;
-import java.util.concurrent.FutureTask;
-
-import rx.Observable;
-import rx.Observer;
-import rx.Scheduler;
+import rx.*;
 import rx.Scheduler.Worker;
-import rx.Subscriber;
-import rx.Subscription;
-import rx.functions.Action0;
-import rx.functions.Action1;
-import rx.functions.Action2;
-import rx.functions.Action3;
-import rx.functions.Action4;
-import rx.functions.Action5;
-import rx.functions.Action6;
-import rx.functions.Action7;
-import rx.functions.Action8;
-import rx.functions.Action9;
-import rx.functions.ActionN;
-import rx.functions.Actions;
-import rx.functions.Func0;
-import rx.functions.Func1;
-import rx.functions.Func2;
-import rx.functions.Func3;
-import rx.functions.Func4;
-import rx.functions.Func5;
-import rx.functions.Func6;
-import rx.functions.Func7;
-import rx.functions.Func8;
-import rx.functions.Func9;
-import rx.functions.FuncN;
+import rx.functions.*;
 import rx.schedulers.Schedulers;
 import rx.subjects.AsyncSubject;
 import rx.subjects.PublishSubject;
 import rx.subjects.Subject;
 import rx.subscriptions.SerialSubscription;
-import rx.util.async.operators.Functionals;
-import rx.util.async.operators.OperatorDeferFuture;
-import rx.util.async.operators.OperatorForEachFuture;
-import rx.util.async.operators.OperatorFromFunctionals;
-import rx.util.async.operators.OperatorStartFuture;
+import rx.util.async.operators.*;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
 
 /**
  * Utility methods to convert functions and actions into asynchronous operations through the Observable/Observer
@@ -95,7 +65,7 @@ public static <T> Observable<T> start(Func0<T> func) {
      * Observable. Multiple subscriptions to this Observable observe the same return value.
      * <p>
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/start.s.png">
-     *
+     * <p>
      * @param <T> the result value type
      * @param func function to run asynchronously
      * @param scheduler Scheduler to run the function on
@@ -111,7 +81,7 @@ public static <T> Observable<T> start(Func0<T> func, Scheduler scheduler) {
      * Convert a synchronous action call into an asynchronous function call through an Observable.
      * <p>
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/toAsync.an.png">
-     *
+     * <p>
      * @param action the action to convert
      * @return a function that returns an Observable that executes the {@code action} and emits {@code null}
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Async-Operators#wiki-toasync-or-asyncaction-or-asyncfunc">RxJava Wiki: toAsync()</a>

settings.gradle

@@ -18,4 +18,4 @@ include 'rxjava-core', \
 'rxjava-contrib:rxjava-async-util', \
 'rxjava-contrib:rxjava-computation-expressions',\
 'rxjava-contrib:rxjava-scalaz', \
-'rxjava-contrib:rxjava-quasar'
+'rxjava-contrib:rxjava-quasar'