1.x: coverage improvements, small adjustments (#4060)

Author: akarnokd

build.gradle

@@ -10,8 +10,8 @@ apply plugin: 'jacoco'
 apply plugin: 'nebula.rxjava-project'
 
 dependencies {
-    testCompile 'junit:junit-dep:4.10'
-    testCompile 'org.mockito:mockito-core:1.8.5'
+    testCompile 'junit:junit:4.12'
+    testCompile 'org.mockito:mockito-core:1.10.19'
 
     perfCompile 'org.openjdk.jmh:jmh-core:1.11.3'
     perfCompile 'org.openjdk.jmh:jmh-generator-annprocess:1.11.3'

src/main/java/rx/Notification.java

@@ -158,15 +158,24 @@ public boolean isOnNext() {
      * @param observer the target observer to call onXXX methods on based on the kind of this Notification instance
      */
     public void accept(Observer<? super T> observer) {
-        if (isOnNext()) {
+        switch (kind) {
+        case OnNext:
             observer.onNext(getValue());
-        } else if (isOnCompleted()) {
-            observer.onCompleted();
-        } else if (isOnError()) {
+            break;
+        case OnError:
             observer.onError(getThrowable());
+            break;
+        case OnCompleted:
+            observer.onCompleted();
+            break;
+        default:
+            throw new AssertionError("Uncovered case: " + kind);
         }
     }
 
+    /**
+     * Specifies the kind of the notification: an element, an error or a completion notification.
+     */
     public enum Kind {
         OnNext, OnError, OnCompleted
     }
@@ -211,19 +220,11 @@ public boolean equals(Object obj) {
             return false;
         }
 
-        if (hasValue() && !getValue().equals(notification.getValue())) {
-            return false;
-        }
-
-        if (hasThrowable() && !getThrowable().equals(notification.getThrowable())) {
-            return false;
-        }
-
-        if (!hasValue() && !hasThrowable() && notification.hasValue()) {
+        if (!(value == notification.value || (value != null && value.equals(notification.value)))) {
             return false;
         }
 
-        if (!hasValue() && !hasThrowable() && notification.hasThrowable()) {
+        if (!(throwable == notification.throwable || (throwable != null && throwable.equals(notification.throwable)))) {
             return false;
         }
 

src/main/java/rx/Observable.java

@@ -3280,7 +3280,8 @@ public static <R> Observable<R> zip(Iterable<? extends Observable<?>> ws, FuncN<
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code zip} does not operate by default on a particular {@link Scheduler}.</dd>
      * </dl>
-     *
+     * 
+     * @param <R> the result type
      * @param ws
      *            an array of source Observables
      * @param zipFunction

src/main/java/rx/Scheduler.java

@@ -130,8 +130,9 @@ public Subscription schedulePeriodically(final Action0 action, long initialDelay
                 long startInNanos = firstStartInNanos;
                 @Override
                 public void call() {
+                    action.call();
+
                     if (!mas.isUnsubscribed()) {
-                        action.call();
 
                         long nextTick;
 

src/main/java/rx/observers/SerializedObserver.java

@@ -43,9 +43,6 @@ public class SerializedObserver<T> implements Observer<T> {
     private FastList queue;
     private final NotificationLite<T> nl = NotificationLite.instance();
 
-    /** Number of iterations without additional safepoint poll in the drain loop. */
-    private static final int MAX_DRAIN_ITERATION = 1024;
-
     static final class FastList {
         Object[] array;
         int size;
@@ -99,31 +96,29 @@ public void onNext(T t) {
             return;
         }
         for (;;) {
-            for (int i = 0; i < MAX_DRAIN_ITERATION; i++) {
-                FastList list;
-                synchronized (this) {
-                    list = queue;
-                    if (list == null) {
-                        emitting = false;
-                        return;
-                    }
-                    queue = null;
+            FastList list;
+            synchronized (this) {
+                list = queue;
+                if (list == null) {
+                    emitting = false;
+                    return;
                 }
-                for (Object o : list.array) {
-                    if (o == null) {
-                        break;
-                    }
-                    try {
-                        if (nl.accept(actual, o)) {
-                            terminated = true;
-                            return;
-                        }
-                    } catch (Throwable e) {
+                queue = null;
+            }
+            for (Object o : list.array) {
+                if (o == null) {
+                    break;
+                }
+                try {
+                    if (nl.accept(actual, o)) {
                         terminated = true;
-                        Exceptions.throwIfFatal(e);
-                        actual.onError(OnErrorThrowable.addValueAsLastCause(e, t));
                         return;
                     }
+                } catch (Throwable e) {
+                    terminated = true;
+                    Exceptions.throwIfFatal(e);
+                    actual.onError(OnErrorThrowable.addValueAsLastCause(e, t));
+                    return;
                 }
             }
         }