Implemented Most Recent

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

@@ -29,27 +29,7 @@
 import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
 
-import rx.operators.OperationConcat;
-import rx.operators.OperationFilter;
-import rx.operators.OperationLast;
-import rx.operators.OperationMap;
-import rx.operators.OperationMaterialize;
-import rx.operators.OperationMerge;
-import rx.operators.OperationMergeDelayError;
-import rx.operators.OperationNext;
-import rx.operators.OperationOnErrorResumeNextViaFunction;
-import rx.operators.OperationOnErrorResumeNextViaObservable;
-import rx.operators.OperationOnErrorReturn;
-import rx.operators.OperationScan;
-import rx.operators.OperationSkip;
-import rx.operators.OperationSynchronize;
-import rx.operators.OperationTake;
-import rx.operators.OperationTakeLast;
-import rx.operators.OperationToObservableFuture;
-import rx.operators.OperationToObservableIterable;
-import rx.operators.OperationToObservableList;
-import rx.operators.OperationToObservableSortedList;
-import rx.operators.OperationZip;
+import rx.operators.*;
 import rx.plugins.RxJavaErrorHandler;
 import rx.plugins.RxJavaPlugins;
 import rx.util.AtomicObservableSubscription;
@@ -1722,6 +1702,18 @@ public static <T> Iterable<T> next(Observable<T> items) {
         return OperationNext.next(items);
     }
 
+    /**
+     * Samples the most recent value in an observable sequence.
+     *
+     * @param source the source observable sequence.
+     * @param <T> the type of observable.
+     * @param initialValue the initial value that will be yielded by the enumerable sequence if no element has been sampled yet.
+     * @return the iterable that returns the last sampled element upon each iteration.
+     */
+    public static <T> Iterable<T> mostRecent(Observable<T> source, T initialValue) {
+        return OperationMostRecent.mostRecent(source, initialValue);
+    }
+
     /**
      * Returns the only element of an observable sequence and throws an exception if there is not exactly one element in the observable sequence.
      * 
@@ -2913,6 +2905,17 @@ public Iterable<T> next() {
         return next(this);
     }
 
+    /**
+     * Samples the most recent value in an observable sequence.
+     *
+     * @param initialValue the initial value that will be yielded by the enumerable sequence if no element has been sampled yet.
+     * @return the iterable that returns the last sampled element upon each iteration.
+     */
+    public Iterable<T> mostRecent(T initialValue) {
+        return mostRecent(this, initialValue);
+    }
+
+
     public static class UnitTest {
 
         @Mock

rxjava-core/src/main/java/rx/operators/OperationMostRecent.java

@@ -0,0 +1,200 @@
+/**
+ * Copyright 2013 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.operators;
+
+import org.junit.Test;
+import rx.Observable;
+import rx.Observer;
+import rx.Subscription;
+import rx.util.Exceptions;
+
+import java.util.Iterator;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Samples the most recent value in an observable sequence.
+ */
+public final class OperationMostRecent {
+
+    public static <T> Iterable<T> mostRecent(final Observable<T> source, T initialValue) {
+
+        MostRecentObserver<T> mostRecentObserver = new MostRecentObserver<T>(initialValue);
+        final MostRecentIterator<T> nextIterator = new MostRecentIterator<T>(mostRecentObserver);
+
+        source.subscribe(mostRecentObserver);
+
+        return new Iterable<T>() {
+            @Override
+            public Iterator<T> iterator() {
+                return nextIterator;
+            }
+        };
+
+    }
+
+    private static class MostRecentIterator<T> implements Iterator<T> {
+
+        private final MostRecentObserver<T> observer;
+
+        private MostRecentIterator(MostRecentObserver<T> observer) {
+            this.observer = observer;
+        }
+
+        @Override
+        public boolean hasNext() {
+            return !observer.isCompleted();
+        }
+
+        @Override
+        public T next() {
+            if (observer.getException() != null) {
+                throw Exceptions.propagate(observer.getException());
+            }
+            return observer.getRecentValue();
+        }
+
+        @Override
+        public void remove() {
+            throw new UnsupportedOperationException("Read only iterator");
+        }
+    }
+
+    private static class MostRecentObserver<T> implements Observer<T> {
+        private final AtomicBoolean completed = new AtomicBoolean(false);
+        private final AtomicReference<T> value;
+        private final AtomicReference<Exception> exception = new AtomicReference<Exception>();
+
+        private MostRecentObserver(T value) {
+            this.value = new AtomicReference<T>(value);
+        }
+
+
+        @Override
+        public void onCompleted() {
+            completed.set(true);
+        }
+
+        @Override
+        public void onError(Exception e) {
+            exception.set(e);
+        }
+
+        @Override
+        public void onNext(T args) {
+            value.set(args);
+        }
+
+        public boolean isCompleted() {
+            return completed.get();
+        }
+
+        public Exception getException() {
+            return exception.get();
+        }
+
+        public T getRecentValue() {
+            return value.get();
+        }
+
+    }
+
+    public static class UnitTest {
+        @Test
+        public void testMostRecent() {
+            Subscription s = mock(Subscription.class);
+            TestObservable observable = new TestObservable(s);
+
+            Iterator<String> it = mostRecent(observable, "default").iterator();
+
+            assertTrue(it.hasNext());
+            assertEquals("default", it.next());
+            assertEquals("default", it.next());
+
+            observable.sendOnNext("one");
+            assertTrue(it.hasNext());
+            assertEquals("one", it.next());
+            assertEquals("one", it.next());
+
+            observable.sendOnNext("two");
+            assertTrue(it.hasNext());
+            assertEquals("two", it.next());
+            assertEquals("two", it.next());
+
+            observable.sendOnCompleted();
+            assertFalse(it.hasNext());
+
+        }
+
+        @Test(expected = TestException.class)
+        public void testMostRecentWithException() {
+            Subscription s = mock(Subscription.class);
+            TestObservable observable = new TestObservable(s);
+
+            Iterator<String> it = mostRecent(observable, "default").iterator();
+
+            assertTrue(it.hasNext());
+            assertEquals("default", it.next());
+            assertEquals("default", it.next());
+
+            observable.sendOnError(new TestException());
+            assertTrue(it.hasNext());
+
+            it.next();
+        }
+
+        private static class TestObservable extends Observable<String> {
+
+            Observer<String> observer = null;
+            Subscription s;
+
+            public TestObservable(Subscription s) {
+                this.s = s;
+            }
+
+            /* used to simulate subscription */
+            public void sendOnCompleted() {
+                observer.onCompleted();
+            }
+
+            /* used to simulate subscription */
+            public void sendOnNext(String value) {
+                observer.onNext(value);
+            }
+
+            /* used to simulate subscription */
+            @SuppressWarnings("unused")
+            public void sendOnError(Exception e) {
+                observer.onError(e);
+            }
+
+            @Override
+            public Subscription subscribe(final Observer<String> observer) {
+                this.observer = observer;
+                return s;
+            }
+        }
+
+        private static class TestException extends RuntimeException {
+
+        }
+
+    }
+
+}