Feature/rx disposable

.gitignore

@@ -127,4 +127,7 @@ fabric.properties
 # https://plugins.jetbrains.com/plugin/12206-codestream
 .idea/codestream.xml
 
-# End of https://www.toptal.com/developers/gitignore/api/intellij,dart
+# End of https://www.toptal.com/developers/gitignore/api/intellij,dart
+
+# VS Code
+.vscode/

README.md

@@ -185,7 +185,25 @@ A Stream that provides synchronous access to the last emitted item, but not repl
 -   Single-subscription
     -   [ValueStreamController](https://pub.dev/documentation/rxdart_ext/latest/not_replay_value_stream/ValueStreamController-class.html)
     -   [toNotReplayValueStream](https://pub.dev/documentation/rxdart_ext/latest/not_replay_value_stream/ToNotReplayValueStreamExtension/toNotReplayValueStream.html)
-
+
+### 5. Disposable
+
+A mixin that makes it easay to dispose streams without having to store and close a [StreamSubscription] variable.
+
+Typical usage is as follows:
+```dart
+class DisposableExample with Disposable {
+  DisposableExample({
+    required Stream<DateTime> dateTimeStream,
+  }) {
+    dateTimeStream.takeUntil(dispose$).listen(
+          (value) => print('Disposable example: $value'),
+        );
+  }
+}
+```
+
+
 ## License
 
     MIT License

example/lib/rxdart_ext_example.dart

@@ -26,4 +26,36 @@ void main() async {
     Single.fromCallable(() => delay(200).then((_) => 2)),
     (int p0, int p1) => p0 + p1,
   ).doOnData(state$.add).forEach((_) => print('<<State>> ${state$.value}'));
+
+  final dateTimeStream = BehaviorSubject<DateTime>.seeded(
+    DateTime.now().toUtc(),
+  );
+  print('Disposable example created');
+  final disposableExample = DisposableExample(
+    dateTimeStream: dateTimeStream,
+  );
+  print('Periodic stream created');
+  final periodicStreamSub = Stream.periodic(
+    const Duration(milliseconds: 100),
+  ).listen((_) {
+    final value = DateTime.now().toUtc();
+    print('Periodic stream: $value');
+    dateTimeStream.add(value);
+  });
+  await delay(500);
+  disposableExample.dispose();
+  print('Disposable example disposed');
+  await delay(500);
+  periodicStreamSub.cancel();
+  print('Periodic stream disposed');
+}
+
+class DisposableExample with Disposable {
+  DisposableExample({
+    required Stream<DateTime> dateTimeStream,
+  }) {
+    dateTimeStream.takeUntil(dispose$).listen(
+          (value) => print('Disposable example: $value'),
+        );
+  }
 }

lib/src/utils/disposable.dart

@@ -0,0 +1,31 @@
+import 'package:meta/meta.dart';
+import 'package:rxdart/rxdart.dart';
+
+/// This mixin adds an easy option to dispose streams without having to store a
+/// [StreamSubscription] variable.
+///
+/// {@tool snippet}
+/// Typical usage is as follows:
+///
+/// ```dart
+/// class ExampleProvider with DisposableMixin {
+///   void init() {
+///     _stream.takeUntil(dispose$).listen(_handleStream);
+///   }
+/// }
+/// ```
+/// {@end-tool}
+mixin Disposable {
+  final _dispose$ = PublishSubject<void>();
+
+  /// The [PublishSubject] that emits null when the dispose method is called.
+  Stream<void> get dispose$ => _dispose$.stream.asBroadcastStream();
+
+  /// Dispose the object and emit null to the [dispose$] stream.
+  @mustCallSuper
+  void dispose() {
+    _dispose$
+      ..add(null)
+      ..close();
+  }
+}

lib/src/utils/index.dart

@@ -1,3 +1,4 @@
 export 'add_null.dart';
 export 'delay.dart';
+export 'disposable.dart';
 export 'identity.dart';

test/utils/disposable_test.dart

@@ -0,0 +1,26 @@
+import 'dart:async';
+
+import 'package:rxdart_ext/utils.dart';
+import 'package:test/test.dart';
+
+class _MockClass with Disposable {
+  _MockClass(
+    this.completer,
+  ) {
+    dispose$.listen((_) => completer.complete());
+  }
+
+  final Completer<void> completer;
+}
+
+void main() {
+  test('Disposable', () async {
+    final completer = Completer<void>();
+    final disposable = _MockClass(completer);
+
+    disposable.dispose();
+    await completer.future.timeout(Duration(milliseconds: 100));
+
+    expect(completer.isCompleted, isTrue);
+  });
+}