Add whereType operator and tests.

lib/operators.dart

@@ -38,3 +38,4 @@ export 'src/operators/timeout.dart';
 export 'src/operators/to_list.dart';
 export 'src/operators/to_set.dart';
 export 'src/operators/where.dart';
+export 'src/operators/where_type.dart';

lib/src/operators/where_type.dart

@@ -0,0 +1,22 @@
+library rx.operators.where_type;
+
+import 'package:rx/src/core/observer.dart';
+import 'package:rx/src/core/operator.dart';
+import 'package:rx/src/core/subscriber.dart';
+
+/// Filter items emitted by the source Observable by only emitting those that
+/// are of the specified type.
+OperatorFunction<T, R> whereType<T, R>() =>
+    (source) => source.lift<R>((source, subscriber) =>
+        source.subscribe(_WhereTypeSubscriber<T, R>(subscriber)));
+
+class _WhereTypeSubscriber<T, R> extends Subscriber<T> {
+  _WhereTypeSubscriber(Observer<R> destination) : super(destination);
+
+  @override
+  void onNext(T value) {
+    if (value is R) {
+      doNext(value);
+    }
+  }
+}

test/operators_test.dart

@@ -1971,4 +1971,22 @@ void main() {
       expect(actual, scheduler.isObservable('--#'));
     });
   });
+  group('whereType', () {
+    const values = {'x': 1};
+    test('first value filterd', () {
+      final input = scheduler.cold<Object>('--x--a--|', values: values);
+      final actual = input.pipe(whereType<Object, String>());
+      expect(actual, scheduler.isObservable<String>('-----a--|'));
+    });
+    test('second value filtered', () {
+      final input = scheduler.cold<Object>('--a--x--|', values: values);
+      final actual = input.pipe(whereType<Object, String>());
+      expect(actual, scheduler.isObservable<String>('--a-----|'));
+    });
+    test('second value filtered and error', () {
+      final input = scheduler.cold<Object>('--a--x--#', values: values);
+      final actual = input.pipe(whereType<Object, String>());
+      expect(actual, scheduler.isObservable<String>('--a-----#'));
+    });
+  });
 }