Used 'isInstance' to replace 'isAssignableFrom' and added a unit test

Author: zsxwing

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

@@ -4366,7 +4366,7 @@ public <R> Observable<R> cast(final Class<R> klass) {
     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());
+                return klass.isInstance(t);
             }
         }).cast(klass);
     }

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

@@ -20,6 +20,8 @@
 import static org.mockito.Mockito.*;
 
 import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -718,4 +720,25 @@ public void testOfType() {
         verify(aObserver, times(1)).onCompleted();
     }
 
+    @Test
+    public void testOfTypeWithPolymorphism() {
+        ArrayList<Integer> l1 = new ArrayList<Integer>();
+        l1.add(1);
+        LinkedList<Integer> l2 = new LinkedList<Integer>();
+        l2.add(2);
+
+        @SuppressWarnings("rawtypes")
+        Observable<List> observable = Observable.<Object>from(l1, l2, "123").ofType(List.class);
+
+        @SuppressWarnings("unchecked")
+        Observer<Object> aObserver = mock(Observer.class);
+        observable.subscribe(aObserver);
+        verify(aObserver, times(1)).onNext(l1);
+        verify(aObserver, times(1)).onNext(l2);
+        verify(aObserver, never()).onNext("123");
+        verify(aObserver, never()).onError(
+                org.mockito.Matchers.any(Throwable.class));
+        verify(aObserver, times(1)).onCompleted();
+    }
+
 }