Add Stream.whereType() extension method.

This is a tool that will help us later.

Change-Id: I7d0411da702c46ce57d88070afd1f35cd1dd0dd7
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/242385
Reviewed-by: Brian Wilkerson <brianwilkerson@google.com>
Commit-Queue: Konstantin Shcheglov <scheglov@google.com>

pkg/analyzer/lib/src/utilities/extensions/stream.dart

@@ -0,0 +1,34 @@
+// Copyright (c) 2022, the Dart project authors. Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+
+class _WhereTypeStreamSink<S, T> implements EventSink<S> {
+  final EventSink<T> _sink;
+
+  _WhereTypeStreamSink(this._sink);
+
+  @override
+  void add(S data) {
+    if (data is T) {
+      _sink.add(data);
+    }
+  }
+
+  @override
+  void addError(e, [StackTrace? stackTrace]) => _sink.addError(e, stackTrace);
+
+  @override
+  void close() => _sink.close();
+}
+
+class _WhereTypeStreamTransformer<S, T> extends StreamTransformerBase<S, T> {
+  @override
+  Stream<T> bind(Stream<S> stream) => Stream.eventTransformed(
+      stream, (sink) => _WhereTypeStreamSink<S, T>(sink));
+}
+
+extension StreamExtension<T> on Stream<T> {
+  Stream<S> whereType<S>() => transform(_WhereTypeStreamTransformer<T, S>());
+}

pkg/analyzer/test/src/utilities/extensions/stream_test.dart

@@ -0,0 +1,23 @@
+// Copyright (c) 2022, the Dart project authors. Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'package:analyzer/src/utilities/extensions/stream.dart';
+import 'package:test/test.dart';
+import 'package:test_reflective_loader/test_reflective_loader.dart';
+
+main() {
+  defineReflectiveSuite(() {
+    defineReflectiveTests(StreamExtensionTest);
+  });
+}
+
+@reflectiveTest
+class StreamExtensionTest {
+  test_whereType() async {
+    var result = await Stream<Object?>.fromIterable([0, '1', 2])
+        .whereType<int>()
+        .toList();
+    expect(result, [0, 2]);
+  }
+}

pkg/analyzer/test/src/utilities/extensions/test_all.dart

@@ -5,11 +5,13 @@
 import 'package:test_reflective_loader/test_reflective_loader.dart';
 
 import 'collection_test.dart' as collection;
+import 'stream_test.dart' as stream;
 import 'string_test.dart' as string;
 
 main() {
   defineReflectiveSuite(() {
     collection.main();
+    stream.main();
     string.main();
   }, name: 'extensions');
 }