Merge pull request ReactiveX#470 from benjchristensen/operator-last

Operator: Last

Author: benjchristensen

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

@@ -49,6 +49,7 @@
 import rx.operators.OperationFirstOrDefault;
 import rx.operators.OperationGroupBy;
 import rx.operators.OperationInterval;
+import rx.operators.OperationLast;
 import rx.operators.OperationMap;
 import rx.operators.OperationMaterialize;
 import rx.operators.OperationMerge;
@@ -4461,12 +4462,21 @@ public <K> Observable<GroupedObservable<K, T>> groupBy(final Func1<? super T, ?
      * <p>
      * In Rx.Net this is negated as the <code>any</code> operator but renamed in RxJava to better match Java naming idioms.
      * 
-     * @return A subscription function for creating the target Observable.
+     * @return An Observable that emits Boolean.
      * @see <a href= "http://msdn.microsoft.com/en-us/library/hh229905(v=vs.103).aspx" >MSDN: Observable.Any</a>
      */
     public Observable<Boolean> isEmpty() {
         return create(OperationAny.isEmpty(this));
     }
+    
+    /**
+     * Returns an {@link Observable} that emits the last element of the source or an <code>IllegalArgumentException</code> if the source {@link Observable} is empty.
+     * 
+     * @return Observable<T>
+     */
+    public Observable<T> last() {
+        return create(OperationLast.last(this));
+    }
 
     /**
      * Converts an Observable into a {@link BlockingObservable} (an Observable with blocking

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

@@ -178,13 +178,10 @@ public Iterator<T> getIterator() {
      * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/B.last.png">
      * 
      * @return the last item emitted by the source {@link Observable}
+     * @throws IllegalArgumentException if source contains no elements
      */
     public T last() {
-        T result = null;
-        for (T value : toIterable()) {
-            result = value;
-        }
-        return result;
+        return new BlockingObservable<T>(o.last()).single();
     }
 
     /**

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

@@ -0,0 +1,82 @@
+/**
+ * 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.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
+import rx.Observer;
+import rx.Subscription;
+
+/**
+ * Emit an Observable<T> with the last emitted item
+ * or onError(new IllegalArgumentException("Sequence contains no elements")) if no elements received.
+ */
+public class OperationLast {
+
+    /**
+     * Accepts a sequence and returns a sequence that is the last emitted item
+     * or an error if no items are emitted (empty sequence).
+     * 
+     * @param sequence
+     *            the input sequence.
+     * @param <T>
+     *            the type of the sequence.
+     * @return a sequence containing the last emitted item or that has onError invoked on it if no items
+     */
+    public static <T> OnSubscribeFunc<T> last(final Observable<? extends T> sequence) {
+        return new OnSubscribeFunc<T>() {
+            final AtomicReference<T> last = new AtomicReference<T>();
+            final AtomicBoolean hasLast = new AtomicBoolean(false);
+
+            @Override
+            public Subscription onSubscribe(final Observer<? super T> observer) {
+                return sequence.subscribe(new Observer<T>() {
+
+                    @Override
+                    public void onCompleted() {
+                        /*
+                         * We don't need to worry about the following being non-atomic
+                         * since an Observable sequence is serial so we will not receive
+                         * concurrent executions.
+                         */
+                        if (hasLast.get()) {
+                            observer.onNext(last.get());
+                            observer.onCompleted();
+                        } else {
+                            observer.onError(new IllegalArgumentException("Sequence contains no elements"));
+                        }
+                    }
+
+                    @Override
+                    public void onError(Throwable e) {
+                        observer.onError(e);
+                    }
+
+                    @Override
+                    public void onNext(T value) {
+                        last.set(value);
+                        hasLast.set(true);
+                    }
+                });
+            }
+
+        };
+    }
+
+}

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

@@ -49,11 +49,10 @@ public void testLast() {
         assertEquals("three", obs.last());
     }
 
-    @Test
+    @Test(expected = IllegalArgumentException.class)
     public void testLastEmptyObservable() {
         BlockingObservable<Object> obs = BlockingObservable.from(Observable.empty());
-
-        assertNull(obs.last());
+        obs.last();
     }
 
     @Test

rxjava-core/src/test/java/rx/operators/OperationLastTest.java

@@ -0,0 +1,49 @@
+/**
+ * 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 static org.junit.Assert.*;
+
+import org.junit.Test;
+
+import rx.Observable;
+
+public class OperationLastTest {
+
+    @Test
+    public void testLastWithElements() {
+        Observable<Integer> last = Observable.create(OperationLast.last(Observable.from(1, 2, 3)));
+        assertEquals(3, last.toBlockingObservable().single().intValue());
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testLastWithNoElements() {
+        Observable<?> last = Observable.create(OperationLast.last(Observable.empty()));
+        last.toBlockingObservable().single();
+    }
+    
+    @Test
+    public void testLastMultiSubscribe() {
+        Observable<Integer> last = Observable.create(OperationLast.last(Observable.from(1, 2, 3)));
+        assertEquals(3, last.toBlockingObservable().single().intValue());
+        assertEquals(3, last.toBlockingObservable().single().intValue());
+    }
+
+    @Test
+    public void testLastViaObservable() {
+        Observable.from(1, 2, 3).last();
+    }
+}