Merge pull request #1380 from benjchristensen/fixes

Variety of Fixes

Author: benjchristensen

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

@@ -7261,7 +7261,7 @@ public final Observable<List<T>> takeLastBuffer(long time, TimeUnit unit, Schedu
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Conditional-and-Boolean-Operators#takeuntil">RxJava Wiki: takeUntil()</a>
      */
     public final <E> Observable<T> takeUntil(Observable<? extends E> other) {
-        return OperatorTakeUntil.takeUntil(this, other);
+        return lift(new OperatorTakeUntil<T, E>(other));
     }
 
     /**

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

@@ -81,9 +81,6 @@ boolean casFirst(int expected, int next) {
         void setObserverRef(Observer<? super T> o) {
             observerRef = o;
         }
-        boolean casObserverRef(Observer<? super T> expected, Observer<? super T> next) {
-            return OBSERVER_UPDATER.compareAndSet(this, expected, next);
-        }
     }
 
     static final class OnSubscribeAction<T> implements OnSubscribe<T> {
@@ -188,7 +185,7 @@ private void drainIfNeededAndSwitchToActual() {
             }
             // now we can safely change over to the actual and get rid of the pass-thru
             // but only if not unsubscribed
-            state.casObserverRef(this, actual);
+            state.setObserverRef(actual);
         }
 
     }

rxjava-core/src/main/java/rx/internal/operators/NotificationLite.java

@@ -53,10 +53,18 @@ public static <T> NotificationLite<T> instance() {
 
     private static final Object ON_COMPLETED_SENTINEL = new Serializable() {
         private static final long serialVersionUID = 1;
+        
+        public String toString() {
+            return "Notification=>Completed";
+        }
     };
 
     private static final Object ON_NEXT_NULL_SENTINEL = new Serializable() {
         private static final long serialVersionUID = 2;
+        
+        public String toString() {
+            return "Notification=>NULL";
+        }
     };
 
     private static class OnErrorSentinel implements Serializable {
@@ -66,6 +74,10 @@ private static class OnErrorSentinel implements Serializable {
         public OnErrorSentinel(Throwable e) {
             this.e = e;
         }
+        
+        public String toString() {
+            return "Notification=>Error:" + e.getMessage();
+        }
     }
 
     /**

rxjava-core/src/main/java/rx/internal/operators/OnSubscribeDelay.java

@@ -55,7 +55,7 @@ public Observable<T> call(T x) {
                 worker.schedule(e, delay, unit);
                 return Observable.create(e);
             }
-        })).subscribe(child);
+        })).unsafeSubscribe(child);
     }
 
     /**

rxjava-core/src/main/java/rx/internal/operators/OperatorTakeUntil.java

@@ -16,100 +16,64 @@
 package rx.internal.operators;
 
 import rx.Observable;
+import rx.Observable.Operator;
 import rx.Subscriber;
-import rx.functions.Func1;
-
-import static rx.Observable.Operator;
 
 /**
  * Returns an Observable that emits the items from the source Observable until another Observable
  * emits an item.
  * <p>
  * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/takeUntil.png">
  */
-public final class OperatorTakeUntil {
-
-    /**
-     * Returns the values from the source observable sequence until the other observable sequence produces a value.
-     * 
-     * @param source
-     *            the source sequence to propagate elements for.
-     * @param other
-     *            the observable sequence that terminates propagation of elements of the source sequence.
-     * @param <T>
-     *            the type of source.
-     * @param <E>
-     *            the other type.
-     * @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<? extends T> source, final Observable<? extends E> other) {
-        Observable<Object> s = source.lift(new SourceObservable<T>());
-        Observable<Object> o = other.lift(new OtherObservable<E>());
-
-        Observable<Object> result = Observable.merge(s, o);
-
-        final NotificationLite<T> notification = NotificationLite.instance();
-
-        return result.takeWhile(new Func1<Object, Boolean>() {
+public final class OperatorTakeUntil<T, E> implements Operator<T, T> {
+
+    private final Observable<? extends E> other;
+
+    public OperatorTakeUntil(final Observable<? extends E> other) {
+        this.other = other;
+    }
+
+    @Override
+    public Subscriber<? super T> call(final Subscriber<? super T> child) {
+        final Subscriber<T> parent = new Subscriber<T>(child) {
+
+            @Override
+            public void onCompleted() {
+                child.onCompleted();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                child.onError(e);
+            }
+
             @Override
-            public Boolean call(Object args) {
-                return !notification.isCompleted(args);
+            public void onNext(T t) {
+                child.onNext(t);
             }
-        }).map(new Func1<Object, T>() {
+
+        };
+
+        other.unsafeSubscribe(new Subscriber<E>(child) {
+
+            @Override
+            public void onCompleted() {
+                parent.onCompleted();
+            }
+
             @Override
-            public T call(Object args) {
-                return notification.getValue(args);
+            public void onError(Throwable e) {
+                parent.onError(e);
             }
+
+            @Override
+            public void onNext(E t) {
+                parent.onCompleted();
+            }
+
         });
-    }
 
-    private final static class SourceObservable<T> implements Operator<Object, T> {
-
-        private final NotificationLite<T> notification = NotificationLite.instance();
-
-        @Override
-        public Subscriber<? super T> call(final Subscriber<? super Object> subscriber) {
-            return new Subscriber<T>(subscriber) {
-                @Override
-                public void onCompleted() {
-                    subscriber.onNext(notification.completed());
-                }
-
-                @Override
-                public void onError(Throwable e) {
-                    subscriber.onError(e);
-                }
-
-                @Override
-                public void onNext(T args) {
-                    subscriber.onNext(notification.next(args));
-                }
-            };
-        }
+        return parent;
     }
 
-    private final static class OtherObservable<E> implements Operator<Object, E> {
-
-        private final NotificationLite<E> notification = NotificationLite.instance();
-
-        @Override
-        public Subscriber<? super E> call(final Subscriber<? super Object> subscriber) {
-            return new Subscriber<E>(subscriber) {
-                @Override
-                public void onCompleted() {
-                    subscriber.onNext(notification.completed());
-                }
-
-                @Override
-                public void onError(Throwable e) {
-                    subscriber.onError(e);
-                }
-
-                @Override
-                public void onNext(E args) {
-                    subscriber.onNext(notification.completed());
-                }
-            };
-        }
-    }
 }

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

@@ -150,6 +150,12 @@ public void assertUnsubscribed() {
         }
     }
 
+    public void assertNoErrors() {
+        if (getOnErrorEvents().size() > 0) {
+            throw new AssertionError("Unexpected onError events: " + getOnErrorEvents().size(), getOnErrorEvents().get(0));
+        }
+    }
+
     /**
      * @warn javadoc missing
      */

rxjava-core/src/main/java/rx/schedulers/TrampolineScheduler.java

@@ -17,6 +17,7 @@
 
 import java.util.PriorityQueue;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
 import rx.Scheduler;
@@ -44,15 +45,20 @@ public Worker createWorker() {
     /* package accessible for unit tests */TrampolineScheduler() {
     }
 
-    private static final ThreadLocal<PriorityQueue<TimedAction>> QUEUE = new ThreadLocal<PriorityQueue<TimedAction>>();
+    private static final ThreadLocal<PriorityQueue<TimedAction>> QUEUE = new ThreadLocal<PriorityQueue<TimedAction>>() {
+        @Override
+        protected PriorityQueue<TimedAction> initialValue() {
+            return new PriorityQueue<TimedAction>();
+        }
+    };
 
     volatile int counter;
-    static final AtomicIntegerFieldUpdater<TrampolineScheduler> COUNTER_UPDATER
-            = AtomicIntegerFieldUpdater.newUpdater(TrampolineScheduler.class, "counter");
+    static final AtomicIntegerFieldUpdater<TrampolineScheduler> COUNTER_UPDATER = AtomicIntegerFieldUpdater.newUpdater(TrampolineScheduler.class, "counter");
 
     private class InnerCurrentThreadScheduler extends Scheduler.Worker implements Subscription {
 
         private final BooleanSubscription innerSubscription = new BooleanSubscription();
+        private final AtomicInteger wip = new AtomicInteger();
 
         @Override
         public Subscription schedule(Action0 action) {
@@ -71,24 +77,16 @@ private Subscription enqueue(Action0 action, long execTime) {
                 return Subscriptions.empty();
             }
             PriorityQueue<TimedAction> queue = QUEUE.get();
-            boolean exec = queue == null;
-
-            if (exec) {
-                queue = new PriorityQueue<TimedAction>();
-                QUEUE.set(queue);
-            }
-
             final TimedAction timedAction = new TimedAction(action, execTime, COUNTER_UPDATER.incrementAndGet(TrampolineScheduler.this));
             queue.add(timedAction);
 
-            if (exec) {
-                while (!queue.isEmpty()) {
+            if (wip.getAndIncrement() == 0) {
+                do {
                     queue.poll().action.call();
-                }
-
-                QUEUE.set(null);
+                } while (wip.decrementAndGet() > 0);
                 return Subscriptions.empty();
             } else {
+                // queue wasn't empty, a parent is already processing so we just add to the end of the queue
                 return Subscriptions.create(new Action0() {
 
                     @Override
@@ -118,9 +116,9 @@ public boolean isUnsubscribed() {
     private static class TimedAction implements Comparable<TimedAction> {
         final Action0 action;
         final Long execTime;
-        final Integer count; // In case if time between enqueueing took less than 1ms
+        final int count; // In case if time between enqueueing took less than 1ms
 
-        private TimedAction(Action0 action, Long execTime, Integer count) {
+        private TimedAction(Action0 action, Long execTime, int count) {
             this.action = action;
             this.execTime = execTime;
             this.count = count;
@@ -130,10 +128,15 @@ private TimedAction(Action0 action, Long execTime, Integer count) {
         public int compareTo(TimedAction that) {
             int result = execTime.compareTo(that.execTime);
             if (result == 0) {
-                return count.compareTo(that.count);
+                return compare(count, that.count);
             }
             return result;
         }
     }
+    
+    // because I can't use Integer.compare from Java 7
+    private static int compare(int x, int y) {
+        return (x < y) ? -1 : ((x == y) ? 0 : 1);
+    }
 
 }

rxjava-core/src/test/java/rx/internal/operators/NotificationLiteTest.java

@@ -0,0 +1,38 @@
+/**
+ * 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.internal.operators;
+
+import static org.junit.Assert.*;
+
+import org.junit.Test;
+
+
+public class NotificationLiteTest {
+
+    @Test
+    public void testComplete() {
+        NotificationLite<Object> on = NotificationLite.instance();
+        Object n = on.next("Hello");
+        Object c = on.completed();
+        
+        assertTrue(on.isCompleted(c));
+        assertFalse(on.isCompleted(n));
+        
+        assertEquals("Hello", on.getValue(n));
+    }
+    
+    
+}

rxjava-core/src/test/java/rx/internal/operators/OperatorMergeMapTest.java

@@ -29,6 +29,7 @@
 import rx.Observable;
 import rx.Observer;
 import rx.exceptions.TestException;
+import rx.functions.Action1;
 import rx.functions.Func0;
 import rx.functions.Func1;
 import rx.functions.Func2;
@@ -204,9 +205,10 @@ public void testFlatMapTransformsException() {
         Observable<Integer> onError = Observable.from(Arrays.asList(5));
 
         Observable<Integer> source = Observable.concat(
-                Observable.from(Arrays.asList(10, 20, 30))
-                , Observable.<Integer> error(new RuntimeException("Forced failure!"))
+                Observable.from(Arrays.asList(10, 20, 30)),
+                Observable.<Integer> error(new RuntimeException("Forced failure!"))
                 );
+        
 
         @SuppressWarnings("unchecked")
         Observer<Object> o = mock(Observer.class);