Added: BO.Latest, fixed: BO.next, BO.mostRecent, BO.toIterable

rxjava-core/src/main/java/rx/observables/BlockingObservable.java

@@ -23,6 +23,7 @@
 import rx.Observable;
 import rx.Observer;
 import rx.Subscription;
+import rx.operators.OperationLatest;
 import rx.operators.OperationMostRecent;
 import rx.operators.OperationNext;
 import rx.operators.OperationToFuture;
@@ -266,6 +267,24 @@ public Iterable<T> next() {
         return OperationNext.next(o);
     }
 
+    /**
+     * Returns the latest item emitted by the underlying Observable, waiting if necessary
+     * for one to become available.
+     * <p>
+     * If the underlying observable produces items faster than the Iterator.next() takes them
+     * onNext events might be skipped, but onError or onCompleted events are not.
+     * <p>
+     * The difference between BlockingObservable.next() and BlockingObservable.latest() is that
+     * the former does not overwrite untaken values whereas the latter does.
+     * <p>
+     * Note also that an onNext() directly followed by onCompleted() might hide the given onNext() event.
+     * 
+     * @return the Iterable sequence
+     */
+    public Iterable<T> latest() {
+        return OperationLatest.latest(o);
+    }
+
     /**
      * If the {@link Observable} completes after emitting a single item, return that item,
      * otherwise throw an exception.

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

@@ -0,0 +1,184 @@
+/**
+ * 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 java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import rx.Notification;
+import rx.Observable;
+import rx.Observer;
+import rx.util.Exceptions;
+
+/**
+ * Wait for and iterate over the latest values of the source observable.
+ * If the source works faster than the iterator, values may be skipped, but
+ * not the onError or onCompleted events.
+ */
+public final class OperationLatest {
+    /** Utility class. */
+    private OperationLatest() { throw new IllegalStateException("No instances!"); }  
+
+    public static <T> Iterable<T> latest(final Observable<? extends T> source) {
+        return new Iterable<T>() {
+            @Override
+            public Iterator<T> iterator() {
+                LatestObserverIterator<T> lio = new LatestObserverIterator<T>();
+                source.subscribe(lio);
+                return lio;
+            }
+        };
+    }
+
+    /** Observer of source, iterator for output. */
+    static final class LatestObserverIterator<T> implements Observer<T>, Iterator<T> {
+        final Lock lock = new ReentrantLock();
+        final Semaphore notify = new Semaphore(0);
+        // observer's values
+        boolean oHasValue;
+        Notification.Kind oKind;
+        T oValue;
+        Throwable oError;
+        @Override
+        public void onNext(T args) {
+            boolean wasntAvailable;
+            lock.lock();
+            try {
+                wasntAvailable = !oHasValue;
+                oHasValue = true;
+                oValue = args;
+                oKind = Notification.Kind.OnNext;
+            } finally {
+                lock.unlock();
+            }
+            if (wasntAvailable) {
+                notify.release();
+            }
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            boolean wasntAvailable;
+            lock.lock();
+            try {
+                wasntAvailable = !oHasValue;
+                oHasValue = true;
+                oValue = null;
+                oError = e;
+                oKind = Notification.Kind.OnError;
+            } finally {
+                lock.unlock();
+            }
+            if (wasntAvailable) {
+                notify.release();
+            }
+        }
+
+        @Override
+        public void onCompleted() {
+            boolean wasntAvailable;
+            lock.lock();
+            try {
+                wasntAvailable = !oHasValue;
+                oHasValue = true;
+                oValue = null;
+                oKind = Notification.Kind.OnCompleted;
+            } finally {
+                lock.unlock();
+            }
+            if (wasntAvailable) {
+                notify.release();
+            }
+        }
+
+        // iterator's values
+
+        boolean iDone;
+        boolean iHasValue;
+        T iValue;
+        Throwable iError;
+        Notification.Kind iKind;
+
+        @Override
+        public boolean hasNext() {
+            if (iError != null) {
+                Exceptions.propagate(iError);
+            }
+            if (!iDone) {
+                if (!iHasValue) {
+                    try {
+                        notify.acquire();
+                    } catch (InterruptedException ex) {
+                        iError = ex;
+                        iHasValue = true;
+                        iKind = Notification.Kind.OnError;
+                        return true;
+                    }
+
+                    lock.lock();
+                    try {
+                        iKind = oKind;
+                        switch (oKind) {
+                        case OnNext:
+                            iValue = oValue;
+                            oValue = null; // handover
+                            break;
+                        case OnError:
+                            iError = oError;
+                            oError = null; // handover
+                            if (iError != null) {
+                                Exceptions.propagate(iError);
+                            }
+                            break;
+                        case OnCompleted:
+                            iDone = true;
+                            break;
+                        }
+                        oHasValue = false;
+                    } finally {
+                        lock.unlock();
+                    }
+                    iHasValue = true;
+                }
+            }
+            return !iDone;
+        }
+
+        @Override
+        public T next() {
+            if (iKind == Notification.Kind.OnError) {
+                Exceptions.propagate(iError);
+            }
+            if (hasNext()) {
+                if (iKind == Notification.Kind.OnNext) {
+                    T v = iValue;
+                    iValue = null; // handover
+                    iHasValue = false;
+                    return v;
+                }
+            }
+            throw new NoSuchElementException();
+        }
+
+        @Override
+        public void remove() {
+            throw new UnsupportedOperationException("Read-only iterator.");
+        }
+
+    }
+}

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

@@ -31,16 +31,16 @@
  */
 public final class OperationMostRecent {
 
-    public static <T> Iterable<T> mostRecent(final Observable<? extends T> source, T initialValue) {
-
-        MostRecentObserver<T> mostRecentObserver = new MostRecentObserver<T>(initialValue);
-        final MostRecentIterator<T> nextIterator = new MostRecentIterator<T>(mostRecentObserver);
-
-        source.subscribe(mostRecentObserver);
+    public static <T> Iterable<T> mostRecent(final Observable<? extends T> source, final T initialValue) {
 
         return new Iterable<T>() {
             @Override
             public Iterator<T> iterator() {
+                MostRecentObserver<T> mostRecentObserver = new MostRecentObserver<T>(initialValue);
+                final MostRecentIterator<T> nextIterator = new MostRecentIterator<T>(mostRecentObserver);
+
+                source.subscribe(mostRecentObserver);
+
                 return nextIterator;
             }
         };

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

@@ -34,19 +34,18 @@
 public final class OperationNext {
 
     public static <T> Iterable<T> next(final Observable<? extends T> items) {
-
-        NextObserver<T> nextObserver = new NextObserver<T>();
-        final NextIterator<T> nextIterator = new NextIterator<T>(nextObserver);
-
-        items.materialize().subscribe(nextObserver);
-
         return new Iterable<T>() {
             @Override
             public Iterator<T> iterator() {
+                NextObserver<T> nextObserver = new NextObserver<T>();
+                final NextIterator<T> nextIterator = new NextIterator<T>(nextObserver);
+
+                items.materialize().subscribe(nextObserver);
+
                 return nextIterator;
             }
         };
-
+        
     }
 
     private static class NextIterator<T> implements Iterator<T> {

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

@@ -16,6 +16,7 @@
 package rx.operators;
 
 import java.util.Iterator;
+import java.util.NoSuchElementException;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 
@@ -63,27 +64,26 @@ public void onNext(Notification<? extends T> args) {
 
         return new Iterator<T>() {
             private Notification<? extends T> buf;
-
+            
             @Override
             public boolean hasNext() {
                 if (buf == null) {
                     buf = take();
                 }
+                if (buf.isOnError()) {
+                    throw Exceptions.propagate(buf.getThrowable());
+                }
                 return !buf.isOnCompleted();
             }
 
             @Override
             public T next() {
-                if (buf == null) {
-                    buf = take();
+                if (hasNext()) {
+                    T result = buf.getValue();
+                    buf = null;
+                    return result;
                 }
-                if (buf.isOnError()) {
-                    throw Exceptions.propagate(buf.getThrowable());
-                }
-
-                T result = buf.getValue();
-                buf = null;
-                return result;
+                throw new NoSuchElementException();
             }
 
             private Notification<? extends T> take() {

rxjava-core/src/test/java/rx/observables/BlockingObservableTest.java

@@ -18,6 +18,8 @@
 import static org.junit.Assert.*;
 
 import java.util.Iterator;
+import java.util.NoSuchElementException;
+import org.junit.Assert;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -201,6 +203,58 @@ public void testToIterable() {
         assertEquals(false, it.hasNext());
 
     }
+    @Test(expected = NoSuchElementException.class)
+    public void testToIterableNextOnly() {
+        BlockingObservable<Integer> obs = BlockingObservable.from(Observable.from(1, 2, 3));
+
+        Iterator<Integer> it = obs.toIterable().iterator();
+
+        Assert.assertEquals((Integer)1, it.next());
+        Assert.assertEquals((Integer)2, it.next());
+        Assert.assertEquals((Integer)3, it.next());
+
+        it.next();
+    }
+
+    @Test(expected = NoSuchElementException.class)
+    public void testToIterableNextOnlyTwice() {
+        BlockingObservable<Integer> obs = BlockingObservable.from(Observable.from(1, 2, 3));
+
+        Iterator<Integer> it = obs.toIterable().iterator();
+
+        Assert.assertEquals((Integer)1, it.next());
+        Assert.assertEquals((Integer)2, it.next());
+        Assert.assertEquals((Integer)3, it.next());
+
+        boolean exc = false;
+        try {
+            it.next();
+        } catch (NoSuchElementException ex) {
+            exc = true;
+        }
+        Assert.assertEquals(true, exc);
+
+        it.next();
+    }
+
+    @Test
+    public void testToIterableManyTimes() {
+        BlockingObservable<Integer> obs = BlockingObservable.from(Observable.from(1, 2, 3));
+
+        Iterable<Integer> iter = obs.toIterable();
+
+        for (int j = 0; j < 3; j++) {
+            Iterator<Integer> it = iter.iterator();
+
+            Assert.assertTrue(it.hasNext());
+            Assert.assertEquals((Integer)1, it.next());
+            Assert.assertTrue(it.hasNext());
+            Assert.assertEquals((Integer)2, it.next());
+            Assert.assertTrue(it.hasNext());
+            Assert.assertEquals((Integer)3, it.next());
+            Assert.assertFalse(it.hasNext());
+        }
+    }
 
     @Test(expected = TestException.class)
     public void testToIterableWithException() {