Merge branch 'master' into debug

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

Author: abersnaze

build.gradle

@@ -44,9 +44,19 @@ subprojects {
 		//include /src/examples folder
         examples
 		//include /src/perf folder
-		// perf //-> Not working so commented out
+        perf {
+            java {
+                srcDir 'src/perf/java'
+                compileClasspath += main.output
+                runtimeClasspath += main.output
+            }
+        }
     }
-    
+
+    dependencies {
+        perfCompile 'org.openjdk.jmh:jmh-core:0.2'
+    }
+
     tasks.build { 
 		//include 'examples' in build task
         dependsOn(examplesClasses)
@@ -58,6 +68,7 @@ subprojects {
         classpath {  
             // include 'provided' dependencies on the classpath
             plusConfigurations += configurations.provided
+            plusConfigurations += configurations.perfCompile
 
             downloadSources = true
             downloadJavadoc = true
@@ -68,6 +79,8 @@ subprojects {
         module {
             // include 'provided' dependencies on the classpath
             scopes.PROVIDED.plus += configurations.provided
+            // TODO not sure what to add it to
+            //scopes.PROVIDED.plus += configurations.perfCompile
         }
     }
 }

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

@@ -404,7 +404,7 @@ class RxScalaDemo extends JUnitSuite {
       import Notification._
       o.materialize.subscribe(n => n match {
         case OnNext(v) => println("Got value " + v)
-        case OnCompleted() => println("Completed")
+        case OnCompleted => println("Completed")
         case OnError(err) => println("Error: " + err.getMessage)
       })
     }
@@ -420,11 +420,19 @@ class RxScalaDemo extends JUnitSuite {
     import Notification._
     List(1, 2, 3).toObservable.materialize.subscribe(n => n match {
       case OnNext(v) => println("Got value " + v)
-      case OnCompleted() => println("Completed")
+      case OnCompleted => println("Completed")
       case OnError(err) => println("Error: " + err.getMessage)
     })
   }
 
+  @Test def notificationSubtyping() {
+    import Notification._
+    val oc1: Notification[Nothing] = OnCompleted
+    val oc2: Notification[Int] = OnCompleted
+    val oc3: rx.Notification[_ <: Int] = oc2.asJavaNotification
+    val oc4: rx.Notification[_ <: Any] = oc2.asJavaNotification
+  }
+  
   @Test def elementAtReplacement() {
     assertEquals("b", List("a", "b", "c").toObservable.drop(1).first.toBlockingObservable.single)
   }

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

@@ -41,7 +41,7 @@ sealed trait Notification[+T] {
     this match {
       case Notification.OnNext(value)  => onNext(value)
       case Notification.OnError(error) => onError(error)
-      case Notification.OnCompleted()  => onCompleted()
+      case Notification.OnCompleted  => onCompleted()
     }
   }
 
@@ -58,7 +58,7 @@ sealed trait Notification[+T] {
     this match {
       case Notification.OnNext(value)  => observer.onNext(value)
       case Notification.OnError(error) => observer.onError(error)
-      case Notification.OnCompleted()  => observer.onCompleted()
+      case Notification.OnCompleted  => observer.onCompleted()
     }
   }
 
@@ -82,7 +82,7 @@ object Notification {
 
   private [scala] def apply[T](n: rx.Notification[_ <: T]): Notification[T] = n.getKind match {
     case rx.Notification.Kind.OnNext => new OnNext(n)
-    case rx.Notification.Kind.OnCompleted => new OnCompleted(n)
+    case rx.Notification.Kind.OnCompleted => OnCompleted
     case rx.Notification.Kind.OnError => new OnError(n)
   }
 
@@ -150,26 +150,9 @@ object Notification {
     override def toString = s"OnError($error)"
   }
 
-  object OnCompleted {
-
-    /**
-     * Constructor for onCompleted notifications.
-     */
-    def apply[T](): Notification[T] = {
-      Notification(rx.Notification.createOnCompleted[T]())
-    }
-
-    /**
-     * Extractor for onCompleted notifications.
-     */
-    def unapply[U](notification: Notification[U]): Option[Unit] = notification match {
-      case onCompleted: OnCompleted[U] => Some()
-      case _ => None
-    }
-  }
-
-  class OnCompleted[T] private[scala](val asJavaNotification: rx.Notification[_ <: T]) extends Notification[T] {
-    override def toString = "OnCompleted()"
+  object OnCompleted extends Notification[Nothing] {
+    override def toString = "OnCompleted"
+    val asJavaNotification = rx.Notification.createOnCompleted[Nothing]()
   }
 
 }

language-adaptors/rxjava-scala/src/test/scala/rx/lang/scala/NotificationTests.scala

@@ -37,8 +37,8 @@ class NotificationTests extends JUnitSuite {
     val onError = OnError(oops)
       assertEquals(oops, onError match { case OnError(error) => error })
 
-    val onCompleted = OnCompleted()
-      assertEquals((), onCompleted match { case OnCompleted() => () })
+    val onCompleted = OnCompleted
+      assertEquals((), onCompleted match { case OnCompleted => () })
   }
 
   @Test
@@ -51,7 +51,7 @@ class NotificationTests extends JUnitSuite {
     val onError = OnError(oops)
       assertEquals(4711, onError(x=>42, e=>4711,()=>13))
 
-    val onCompleted = OnCompleted()
+    val onCompleted = OnCompleted
       assertEquals(13, onCompleted(x=>42, e=>4711,()=>13))
 
   }

rxjava-core/build.gradle

@@ -1,5 +1,6 @@
 apply plugin: 'maven'
 apply plugin: 'osgi'
+apply plugin:'application'
 
 sourceCompatibility = JavaVersion.VERSION_1_6
 targetCompatibility = JavaVersion.VERSION_1_6
@@ -31,3 +32,9 @@ jar {
     }
 }
 
+task time(type:JavaExec) {
+    classpath = sourceSets.perf.runtimeClasspath
+    group 'Application'
+    description 'Execute the calipser benchmark timing of Rx'
+    main 'rx.operators.ObservableBenchmark'
+}

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

@@ -27,13 +27,14 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
-import rx.Observable.OnSubscribe;
 import rx.joins.Pattern2;
 import rx.joins.Plan0;
 import rx.observables.BlockingObservable;
 import rx.observables.ConnectableObservable;
 import rx.observables.GroupedObservable;
 import rx.observers.SafeSubscriber;
+import rx.operators.OnSubscribeFromIterable;
+import rx.operators.OnSubscribeRange;
 import rx.operators.OperationAll;
 import rx.operators.OperationAmb;
 import rx.operators.OperationAny;
@@ -50,8 +51,6 @@
 import rx.operators.OperationDematerialize;
 import rx.operators.OperationDistinct;
 import rx.operators.OperationDistinctUntilChanged;
-import rx.operators.Operator;
-import rx.operators.OperatorDoOnEach;
 import rx.operators.OperationElementAt;
 import rx.operators.OperationFilter;
 import rx.operators.OperationFinally;
@@ -65,13 +64,11 @@
 import rx.operators.OperationMergeDelayError;
 import rx.operators.OperationMinMax;
 import rx.operators.OperationMulticast;
-import rx.operators.OperationObserveOn;
 import rx.operators.OperationOnErrorResumeNextViaFunction;
 import rx.operators.OperationOnErrorResumeNextViaObservable;
 import rx.operators.OperationOnErrorReturn;
 import rx.operators.OperationOnExceptionResumeNextViaObservable;
 import rx.operators.OperationParallelMerge;
-import rx.operators.OperatorRepeat;
 import rx.operators.OperationReplay;
 import rx.operators.OperationRetry;
 import rx.operators.OperationSample;
@@ -98,19 +95,21 @@
 import rx.operators.OperationToObservableFuture;
 import rx.operators.OperationUsing;
 import rx.operators.OperationWindow;
-import rx.operators.OperatorSubscribeOn;
-import rx.operators.OperatorZip;
 import rx.operators.Operator;
 import rx.operators.OperatorCast;
-import rx.operators.OperatorFromIterable;
+import rx.operators.OperatorDoOnEach;
 import rx.operators.OperatorGroupBy;
 import rx.operators.OperatorMap;
 import rx.operators.OperatorMerge;
+import rx.operators.OperatorObserveOn;
 import rx.operators.OperatorParallel;
+import rx.operators.OperatorRepeat;
+import rx.operators.OperatorSubscribeOn;
 import rx.operators.OperatorTake;
 import rx.operators.OperatorTimestamp;
 import rx.operators.OperatorToObservableList;
 import rx.operators.OperatorToObservableSortedList;
+import rx.operators.OperatorZip;
 import rx.operators.OperatorZipIterable;
 import rx.plugins.RxJavaObservableExecutionHook;
 import rx.plugins.RxJavaPlugins;
@@ -121,8 +120,8 @@
 import rx.subjects.ReplaySubject;
 import rx.subjects.Subject;
 import rx.subscriptions.Subscriptions;
+import rx.util.Exceptions;
 import rx.util.OnErrorNotImplementedException;
-import rx.util.Range;
 import rx.util.TimeInterval;
 import rx.util.Timestamped;
 import rx.util.functions.Action0;
@@ -1218,7 +1217,7 @@ public final static <T> Observable<T> from(Future<? extends T> future, Scheduler
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Creating-Observables#wiki-from">RxJava Wiki: from()</a>
      */
     public final static <T> Observable<T> from(Iterable<? extends T> iterable) {
-        return create(new OperatorFromIterable<T>(iterable));
+        return create(new OnSubscribeFromIterable<T>(iterable));
     }
 
     /**
@@ -1240,7 +1239,7 @@ public final static <T> Observable<T> from(Iterable<? extends T> iterable) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh212140.aspx">MSDN: Observable.ToObservable</a>
      */
     public final static <T> Observable<T> from(Iterable<? extends T> iterable, Scheduler scheduler) {
-        return create(new OperatorFromIterable<T>(iterable)).subscribeOn(scheduler);
+        return create(new OnSubscribeFromIterable<T>(iterable)).subscribeOn(scheduler);
     }
 
     /**
@@ -2440,7 +2439,13 @@ public final static <T> Observable<Observable<T>> parallelMerge(Observable<Obser
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229460.aspx">MSDN: Observable.Range</a>
      */
     public final static Observable<Integer> range(int start, int count) {
-        return from(Range.createWithCount(start, count));
+        if (count < 1) {
+            throw new IllegalArgumentException("Count must be positive");
+        }
+        if ((start + count) > Integer.MAX_VALUE) {
+            throw new IllegalArgumentException("start + count can not exceed Integer.MAX_VALUE");
+        }
+        return Observable.create(new OnSubscribeRange(start, start + (count - 1)));
     }
 
     /**
@@ -2460,7 +2465,7 @@ public final static Observable<Integer> range(int start, int count) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211896.aspx">MSDN: Observable.Range</a>
      */
     public final static Observable<Integer> range(int start, int count, Scheduler scheduler) {
-        return from(Range.createWithCount(start, count), scheduler);
+        return range(start, count).subscribeOn(scheduler);
     }
 
     /**
@@ -5140,8 +5145,7 @@ public final <R> ConnectableObservable<R> multicast(Subject<? super T, ? extends
     }
 
     /**
-     * Modify the source Observable so that it asynchronously notifies {@link Observer}s on the
-     * specified {@link Scheduler}.
+     * Move notifications to the specified {@link Scheduler} one `onNext` at a time.
      * <p>
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/observeOn.png">
      * 
@@ -5152,9 +5156,26 @@ public final <R> ConnectableObservable<R> multicast(Subject<? super T, ? extends
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#wiki-observeon">RxJava Wiki: observeOn()</a>
      */
     public final Observable<T> observeOn(Scheduler scheduler) {
-        return create(OperationObserveOn.observeOn(this, scheduler));
+        return lift(new OperatorObserveOn<T>(scheduler));
     }
 
+    /**
+     * Move notifications to the specified {@link Scheduler} asynchronously with a buffer of the given size.
+     * <p>
+     * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/observeOn.png">
+     * 
+     * @param scheduler
+     *            the {@link Scheduler} to notify {@link Observer}s on
+     * @param bufferSize
+     *            that will be rounded up to the next power of 2
+     * @return the source Observable modified so that its {@link Observer}s are notified on the
+     *         specified {@link Scheduler}
+     * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#wiki-observeon">RxJava Wiki: observeOn()</a>
+     */
+    public final Observable<T> observeOn(Scheduler scheduler, int bufferSize) {
+        return lift(new OperatorObserveOn<T>(scheduler, bufferSize));
+    }
+    
     /**
      * Filters the items emitted by an Observable, only emitting those of the specified type.
      * <p>
@@ -5297,7 +5318,9 @@ public final Observable<T> onExceptionResumeNext(final Observable<? extends T> r
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#wiki-parallel">RxJava Wiki: parallel()</a>
      */
     public final <R> Observable<R> parallel(Func1<Observable<T>, Observable<R>> f) {
-        return lift(new OperatorParallel<T, R>(f, Schedulers.computation()));
+        // TODO move this back to Schedulers.computation() again once that is properly using eventloops
+        // see https://github.com/Netflix/RxJava/issues/713 for why this was changed
+        return lift(new OperatorParallel<T, R>(f, Schedulers.newThread()));
     }
 
     /**
@@ -6966,10 +6989,9 @@ public void call() {
                 }
 
             });
-        } catch (OnErrorNotImplementedException e) {
-            // special handling when onError is not implemented ... we just rethrow
-            throw e;
         } catch (Throwable e) {
+            // special handling for certain Throwable/Error/Exception types
+            Exceptions.throwIfFatal(e);
             // if an unhandled error occurs executing the onSubscribe we will propagate it
             try {
                 observer.onError(hook.onSubscribeError(this, e));

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

@@ -19,11 +19,9 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import rx.Subscriber;
-import rx.Subscription;
-import rx.operators.SafeObservableSubscription;
 import rx.plugins.RxJavaPlugins;
-import rx.subscriptions.Subscriptions;
 import rx.util.CompositeException;
+import rx.util.Exceptions;
 import rx.util.OnErrorNotImplementedException;
 
 /**
@@ -74,6 +72,9 @@ public void onCompleted() {
             try {
                 actual.onCompleted();
             } catch (Throwable e) {
+                // we handle here instead of another method so we don't add stacks to the frame
+                // which can prevent it from being able to handle StackOverflow
+                Exceptions.throwIfFatal(e);
                 // handle errors if the onCompleted implementation fails, not just if the Observable fails
                 _onError(e);
             } finally {
@@ -85,6 +86,9 @@ public void onCompleted() {
 
     @Override
     public void onError(Throwable e) {
+        // we handle here instead of another method so we don't add stacks to the frame
+        // which can prevent it from being able to handle StackOverflow
+        Exceptions.throwIfFatal(e);
         if (isFinished.compareAndSet(false, true)) {
             _onError(e);
         }
@@ -97,6 +101,9 @@ public void onNext(T args) {
                 actual.onNext(args);
             }
         } catch (Throwable e) {
+            // we handle here instead of another method so we don't add stacks to the frame
+            // which can prevent it from being able to handle StackOverflow
+            Exceptions.throwIfFatal(e);
             // handle errors if the onNext implementation fails, not just if the Observable fails
             onError(e);
         }