implemented count operator (#32)

Author: jmhofer

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

@@ -2043,6 +2043,21 @@ public Observable<T> reduce(Func2<? super T, ? super T, ? extends T> accumulator
         return create(OperationScan.scan(this, accumulator)).takeLast(1);
     }
 
+    /**
+     * Returns an Observable that counts the total number of elements in the source Observable.
+     * @return an Observable emitting the number of counted elements of the source Observable 
+     *         as its single item.
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229470%28v=vs.103%29.aspx">MSDN: Observable.Count</a>
+     */
+    public Observable<Integer> count() {
+        return reduce(0, new Func2<Integer, T, Integer>() {
+            @Override
+            public Integer call(Integer t1, T t2) {
+                return t1 + 1;
+            }
+        });
+    }
+    
     /**
      * Returns a {@link ConnectableObservable} that shares a single subscription to the underlying
      * Observable that will replay all of its items and notifications to any future {@link Observer}.

rxjava-core/src/test/java/rx/ObservableTests.java

@@ -27,7 +27,6 @@
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mock;
-import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
 
 import rx.Observable.OnSubscribeFunc;
@@ -69,10 +68,47 @@ public Subscription onSubscribe(Observer<? super String> Observer) {
         verify(aObserver, times(1)).onNext("one");
         verify(aObserver, times(1)).onNext("two");
         verify(aObserver, times(1)).onNext("three");
-        verify(aObserver, Mockito.never()).onError(any(Throwable.class));
+        verify(aObserver, never()).onError(any(Throwable.class));
         verify(aObserver, times(1)).onCompleted();
     }
 
+    @Test
+    public void testCountAFewItems() {
+        Observable<String> observable = Observable.from("a", "b", "c", "d");
+        observable.count().subscribe(w);
+        // we should be called only once
+        verify(w, times(1)).onNext(anyInt());
+        verify(w).onNext(4);
+        verify(w, never()).onError(any(Throwable.class));
+        verify(w, times(1)).onCompleted();
+    }
+    
+    @Test
+    public void testCountZeroItems() {
+        Observable<String> observable = Observable.empty();
+        observable.count().subscribe(w);
+        // we should be called only once
+        verify(w, times(1)).onNext(anyInt());
+        verify(w).onNext(0);
+        verify(w, never()).onError(any(Throwable.class));
+        verify(w, times(1)).onCompleted();
+    }
+
+    @Test
+    public void testCountError() {
+        Observable<String> o = Observable.create(new OnSubscribeFunc<String>() {
+            @Override
+            public Subscription onSubscribe(Observer<? super String> obsv) {
+                obsv.onError(new RuntimeException());
+                return Subscriptions.empty();
+            }
+        });
+        o.count().subscribe(w);
+        verify(w, never()).onNext(anyInt());
+        verify(w, never()).onCompleted();
+        verify(w, times(1)).onError(any(RuntimeException.class));
+    }
+    
     @Test
     public void testReduce() {
         Observable<Integer> observable = Observable.from(1, 2, 3, 4);