OnBackpressureLatest: Non-blocking version of the toBlocking().latest() operator.

src/main/java/rx/Observable.java

@@ -5367,6 +5367,27 @@ public final Observable<T> onBackpressureBlock() {
         return onBackpressureBlock(rx.internal.util.RxRingBuffer.SIZE);
     }
 
+    /**
+     * Instructs an Observable that is emitting items faster than its observer can consume them to 
+     * hold onto the latest value and emit that on request.
+     * <p>
+     * Its behavior is logically equivalent to toBlocking().latest() with the exception that
+     * the downstream is not blocking while requesting more values.
+     * <p>
+     * Note that if the upstream Observable does support backpressure, this operator ignores that capability
+     * and doesn't propagate any backpressure requests from downstream.
+     * <p>
+     * Note that due to the nature of how backpressure requests are propagated through subscribeOn/observeOn,
+     * requesting more than 1 from downstream doesn't guarantee a continuous delivery of onNext events.
+     * @return
+     * @Experimental The behavior of this can change at any time. 
+     * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+     */
+    @Experimental
+    public final Observable<T> onBackpressureLatest() {
+        return lift(OperatorOnBackpressureLatest.<T>instance());
+    }
+
     /**
      * Instructs an Observable to pass control to another Observable rather than invoking
      * {@link Observer#onError onError} if it encounters an error.

src/main/java/rx/internal/operators/OperatorOnBackpressureLatest.java

@@ -0,0 +1,225 @@
+/**
+ * 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 java.util.concurrent.atomic.*;
+
+import rx.Observable.Operator;
+import rx.*;
+
+/**
+ * An operator which drops all but the last received value in case the downstream
+ * doesn't request more.
+ */
+public final class OperatorOnBackpressureLatest<T> implements Operator<T, T> {
+    /** Holds a singleton instance initialized on class-loading. */
+    static final class Holder {
+        static final OperatorOnBackpressureLatest<Object> INSTANCE = new OperatorOnBackpressureLatest<Object>();
+    }
+
+    /**
+     * Returns a singleton instance of the OnBackpressureLatest operator since it is stateless.
+     * @return the single instanceof OperatorOnBackpressureLatest
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> OperatorOnBackpressureLatest<T> instance() {
+        return (OperatorOnBackpressureLatest<T>)Holder.INSTANCE;
+    }
+
+    @Override
+    public Subscriber<? super T> call(Subscriber<? super T> child) {
+        final LatestEmitter<T> producer = new LatestEmitter<T>(child);
+        LatestSubscriber<T> parent = new LatestSubscriber<T>(producer);
+        producer.parent = parent;
+        child.add(parent);
+        child.add(producer);
+        child.setProducer(producer);
+        return parent;
+    }
+    /**
+     * A terminatable producer which emits the latest items on request.
+     * @param <T>
+     */
+    static final class LatestEmitter<T> extends AtomicLong implements Producer, Subscription, Observer<T> {
+        /** */
+        private static final long serialVersionUID = -1364393685005146274L;
+        final Subscriber<? super T> child;
+        LatestSubscriber<? super T> parent;
+        final AtomicReference<Object> value;
+        /** Written before done, read after done. */
+        Throwable terminal;
+        volatile boolean done;
+        /** Guarded by this. */
+        boolean emitting;
+        /** Guarded by this. */
+        boolean missed;
+        static final Object EMPTY = new Object();
+        static final long NOT_REQUESTED = Long.MIN_VALUE / 2;
+        public LatestEmitter(Subscriber<? super T> child) {
+            this.child = child;
+            this.value = new AtomicReference<Object>(EMPTY);
+            this.lazySet(NOT_REQUESTED); // not 
+        }
+        @Override
+        public void request(long n) {
+            if (n >= 0) {
+                for (;;) {
+                    long r = get();
+                    if (r == Long.MIN_VALUE) {
+                        return;
+                    }
+                    long u;
+                    if (r == NOT_REQUESTED) {
+                        u = n;
+                    } else {
+                        u = r + n;
+                        if (u < 0) {
+                            u = Long.MAX_VALUE;
+                        }
+                    }
+                    if (compareAndSet(r, u)) {
+                        if (r == NOT_REQUESTED) {
+                            parent.requestMore(Long.MAX_VALUE);
+                        }
+                        emit();
+                        return;
+                    }
+                }
+            }
+        }
+        long produced(long n) {
+            for (;;) {
+                long r = get();
+                if (r < 0) {
+                    return r;
+                }
+                long u = r - n;
+                if (compareAndSet(r, u)) {
+                    return u;
+                }
+            }
+        }
+        @Override
+        public boolean isUnsubscribed() {
+            return get() == Long.MIN_VALUE;
+        }
+        @Override
+        public void unsubscribe() {
+            if (get() >= 0) {
+                getAndSet(Long.MIN_VALUE);
+            }
+        }
+
+        @Override
+        public void onNext(T t) {
+            value.lazySet(t); // emit's synchronized block does a full release
+            emit();
+        }
+        @Override
+        public void onError(Throwable e) {
+            terminal = e;
+            done = true;
+            emit();
+        }
+        @Override
+        public void onCompleted() {
+            done = true;
+            emit();
+        }
+        void emit() {
+            synchronized (this) {
+                if (emitting) {
+                    missed = true;
+                    return;
+                }
+                emitting = true;
+                missed = false;
+            }
+            boolean skipFinal = false;
+            try {
+                for (;;) {
+                    long r = get();
+                    if (r == Long.MIN_VALUE) {
+                        skipFinal = true;
+                        break;
+                    }
+                    Object v = value.get();
+                    if (r > 0 && v != EMPTY) {
+                        @SuppressWarnings("unchecked")
+                        T v2 = (T)v;
+                        child.onNext(v2);
+                        value.compareAndSet(v, EMPTY);
+                        produced(1);
+                        v = EMPTY;
+                    }
+                    if (v == EMPTY && done) {
+                        Throwable e = terminal;
+                        if (e != null) {
+                            child.onError(e);
+                        } else {
+                            child.onCompleted();
+                        }
+                    }
+                    synchronized (this) {
+                        if (!missed) {
+                            emitting = false;
+                            skipFinal = true;
+                            break;
+                        }
+                        missed = false;
+                    }
+                }
+            } finally {
+                if (!skipFinal) {
+                    synchronized (this) {
+                        emitting = false;
+                    }
+                }
+            }
+        }
+    }
+    static final class LatestSubscriber<T> extends Subscriber<T> {
+        private final LatestEmitter<T> producer;
+
+        private LatestSubscriber(LatestEmitter<T> producer) {
+            this.producer = producer;
+        }
+
+        @Override
+        public void onStart() {
+            // don't run until the child actually requested to avoid synchronous problems
+            request(0); 
+        }
+
+        @Override
+        public void onNext(T t) {
+            producer.onNext(t);
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            producer.onError(e);
+        }
+
+        @Override
+        public void onCompleted() {
+            producer.onCompleted();
+        }
+        void requestMore(long n) {
+            request(n);
+        }
+    }
+}