Add whereNotNull (dart-lang/stream_transform#156)

This is useful during null safety migrations for the same reason as
`Iterable.whereNotNull` from `package:collection`.

pkgs/stream_transform/CHANGELOG.md

diff --git a/pkgs/stream_transform/CHANGELOG.md b/pkgs/stream_transform/CHANGELOG.md
index f425a1a..3d2cd69 100644
--- a/pkgs/stream_transform/CHANGELOG.md
+++ b/pkgs/stream_transform/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 2.1.0
+
+- Add `whereNotNull`.
+
 ## 2.0.1
 
 - Require Dart 2.14 or greater.

pkgs/stream_transform/lib/src/where.dart

diff --git a/pkgs/stream_transform/lib/src/where.dart b/pkgs/stream_transform/lib/src/where.dart
index 569dc31..6d210ea 100644
--- a/pkgs/stream_transform/lib/src/where.dart
+++ b/pkgs/stream_transform/lib/src/where.dart
@@ -58,3 +58,14 @@
     });
   }
 }
+
+extension WhereNotNull<T extends Object> on Stream<T?> {
+  /// Discards `null` events from this stream.
+  ///
+  /// If the source stream is a broadcast stream the result will be as well.
+  ///
+  /// Errors from the source stream are forwarded directly to the result stream.
+  Stream<T> whereNotNull() => transformByHandlers(onData: (event, sink) {
+        if (event != null) sink.add(event);
+      });
+}

pkgs/stream_transform/pubspec.yaml

diff --git a/pkgs/stream_transform/pubspec.yaml b/pkgs/stream_transform/pubspec.yaml
index 1815178..a531122 100644
--- a/pkgs/stream_transform/pubspec.yaml
+++ b/pkgs/stream_transform/pubspec.yaml
@@ -1,5 +1,5 @@
 name: stream_transform
-version: 2.0.1
+version: 2.1.0
 description: A collection of utilities to transform and manipulate streams.
 repository: https://github.com/dart-lang/stream_transform
 

pkgs/stream_transform/test/where_not_null_test.dart

diff --git a/pkgs/stream_transform/test/where_not_null_test.dart b/pkgs/stream_transform/test/where_not_null_test.dart
new file mode 100644
index 0000000..c7e4e9b
--- /dev/null
+++ b/pkgs/stream_transform/test/where_not_null_test.dart
@@ -0,0 +1,56 @@
+// 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';
+
+import 'package:stream_transform/stream_transform.dart';
+import 'package:test/test.dart';
+
+void main() {
+  test('forwards only events that match the type', () async {
+    var values = Stream.fromIterable([null, 'a', null, 'b']);
+    var filtered = values.whereNotNull();
+    expect(await filtered.toList(), ['a', 'b']);
+  });
+
+  test('can result in empty stream', () async {
+    var values = Stream<Object?>.fromIterable([null, null]);
+    var filtered = values.whereNotNull();
+    expect(await filtered.isEmpty, true);
+  });
+
+  test('forwards values to multiple listeners', () async {
+    var values = StreamController<Object?>.broadcast();
+    var filtered = values.stream.whereNotNull();
+    var firstValues = [];
+    var secondValues = [];
+    filtered
+      ..listen(firstValues.add)
+      ..listen(secondValues.add);
+    values
+      ..add(null)
+      ..add('a')
+      ..add(null)
+      ..add('b');
+    await Future(() {});
+    expect(firstValues, ['a', 'b']);
+    expect(secondValues, ['a', 'b']);
+  });
+
+  test('closes streams with multiple listeners', () async {
+    var values = StreamController<Object?>.broadcast();
+    var filtered = values.stream.whereNotNull();
+    var firstDone = false;
+    var secondDone = false;
+    filtered
+      ..listen(null, onDone: () => firstDone = true)
+      ..listen(null, onDone: () => secondDone = true);
+    values
+      ..add(null)
+      ..add('a');
+    await values.close();
+    expect(firstDone, true);
+    expect(secondDone, true);
+  });
+}