add asNullable

CHANGELOG.md

@@ -1,3 +1,8 @@
+## TBD
+
+- **operators**:
+    - Add `Stream.asNullable()` operator.
+
 ## 0.3.0 - Jun 16, 2024
 
 - Accept Dart SDK versions above 3.0 (`sdk: '>=2.12.0 <4.0.0'`).

lib/src/operators/as_nullable.dart

@@ -0,0 +1,24 @@
+/// Adapt this `Stream<T>` to be a `Stream<T?>`.
+extension AsNullableStreamExtension<T extends Object> on Stream<T> {
+  /// Adapt this `Stream<T>` to be a `Stream<T?>`.
+  ///
+  /// Useful for broadcast operators that use `null` as the seed or initial value,
+  /// such as `publishState(null)` and `publishValueSeed(null)`.
+  ///
+  /// ### Example
+  /// ```dart
+  /// final stream = Stream.fromIterable([1, 2, 3, 4]);
+  ///
+  /// // ERROR: The argument type 'Null' can't be assigned to the parameter type 'int'.
+  /// stream.publishState(null);
+  ///
+  /// final StateConnectableStream<int?> stateStream =
+  ///     stream.cast<int?>().publishState(null); // OK but it is verbose and unoptimized.
+  ///
+  /// final StateConnectableStream<int?> stateStream =
+  ///     stream.asNullable().publishState(null); // OK and optimized.
+  /// ```
+  @pragma('vm:prefer-inline')
+  @pragma('dart2js:tryInline')
+  Stream<T?> asNullable() => this;
+}

lib/src/operators/index.dart

@@ -1,3 +1,4 @@
+export 'as_nullable.dart';
 export 'debug.dart';
 export 'distinct_by.dart';
 export 'distinct_unique_by.dart';

test/operators/cast_nullable_test.dart

@@ -0,0 +1,40 @@
+import 'dart:async';
+
+import 'package:rxdart_ext/rxdart_ext.dart';
+import 'package:test/expect.dart';
+import 'package:test/scaffolding.dart';
+
+void main() {
+  group('asNullable', () {
+    test('works', () {
+      final stream = Stream.fromIterable([1, 2, 3, 4]);
+      final asNullable = stream.asNullable();
+
+      expect(asNullable, isA<Stream<int?>>());
+      expect(asNullable, same(stream));
+    });
+
+    test('asNullable with publishState', () async {
+      final stream = Stream.fromIterable([1, 1, 2, 2, 3, 3, 4, 4]);
+      final stateStream = stream.asNullable().publishState(null);
+
+      expect(stateStream, isA<StateConnectableStream<int?>>());
+      expect(stateStream.value, null);
+
+      scheduleMicrotask(() => stateStream.connect());
+      await expectLater(stateStream, emitsInOrder([1, 2, 3, 4]));
+    });
+
+    test('asNullable with publishValueSeeded', () async {
+      final stream = Stream.fromIterable([1, 1, 2, 2, 3, 3, 4, 4]);
+      final valueStream = stream.asNullable().publishValueSeeded(null);
+
+      expect(valueStream, isA<ValueConnectableStream<int?>>());
+      expect(valueStream.value, null);
+
+      scheduleMicrotask(() => valueStream.connect());
+      await expectLater(
+          valueStream, emitsInOrder([null, 1, 1, 2, 2, 3, 3, 4, 4]));
+    });
+  });
+}