Implemented the 'ofType' operator

Author: zsxwing

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

@@ -4350,6 +4350,27 @@ public <R> Observable<R> cast(final Class<R> klass) {
         return create(OperationCast.cast(this, klass));
     }
 
+    /**
+     * Filters the elements of an observable sequence based on the specified
+     * type.
+     *
+     * @param klass
+     *            The class type to filter the elements in the source sequence
+     *            on.
+     *
+     * @return An observable sequence that contains elements from the input
+     *         sequence of type klass.
+     *
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229380(v=vs.103).aspx">MSDN: Observable.OfType</a>
+     */
+    public <R> Observable<R> ofType(final Class<R> klass) {
+        return filter(new Func1<T, Boolean>() {
+            public Boolean call(T t) {
+                return klass.isAssignableFrom(t.getClass());
+            }
+        }).cast(klass);
+    }
+
     /**
      * Whether a given {@link Function} is an internal implementation inside rx.* packages or not.
      * <p>

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

@@ -701,4 +701,21 @@ public void onNext(String v) {
             fail("It should be a NumberFormatException");
         }
     }
+    
+    @Test
+    public void testOfType() {
+        Observable<String> observable = Observable.from(1, "abc", false, 2L).ofType(String.class);
+
+        @SuppressWarnings("unchecked")
+        Observer<Object> aObserver = mock(Observer.class);
+        observable.subscribe(aObserver);
+        verify(aObserver, never()).onNext(1);
+        verify(aObserver, times(1)).onNext("abc");
+        verify(aObserver, never()).onNext(false);
+        verify(aObserver, never()).onNext(2L);
+        verify(aObserver, never()).onError(
+                org.mockito.Matchers.any(Throwable.class));
+        verify(aObserver, times(1)).onCompleted();
+    }
+
 }