ModifiableSnapshotView.fromStream behavior

CHANGELOG.md

@@ -1,3 +1,3 @@
-## 0.1.0-dev.2
+## 0.1.0-dev.3
 
 - Initial version

lib/src/snapshot_view.dart

@@ -1,5 +1,7 @@
 library snapshot.view;
 
+import 'dart:async';
+
 import 'package:meta/meta.dart';
 import 'package:rxdart/rxdart.dart';
 import 'package:snapshot/snapshot.dart';
@@ -107,9 +109,24 @@ class ModifiableSnapshotView with SnapshotView {
   }
 
   ModifiableSnapshotView.fromStream(Stream<Snapshot> stream) {
-    _snapshots.addStream(stream.doOnDone(() {
-      _snapshots.close();
-    }));
+    StreamSubscription subscription;
+    _snapshots.onListen = () {
+      subscription ??= stream.listen(_snapshots.add,
+          onError: _snapshots.addError, onDone: _snapshots.close);
+      subscription.resume();
+    };
+    _snapshots.onCancel = () {
+      if (stream.isBroadcast) {
+        subscription.cancel();
+        subscription = null;
+      } else {
+        subscription.pause();
+      }
+    };
+    _snapshots.done.then((v) {
+      subscription?.cancel();
+      subscription = null;
+    });
   }
 
   @override

pubspec.yaml

@@ -1,6 +1,6 @@
 name: snapshot
 description: A library that can be used to implement data classes and simplifies conversion of JSON to data classes
-version: 0.1.0-dev.2
+version: 0.1.0-dev.3
 homepage: https://github.com/appsup-dart/snapshot
 
 environment:

test/snapshot_view_test.dart

@@ -1,3 +1,5 @@
+import 'dart:async';
+
 import 'package:snapshot/snapshot.dart';
 import 'package:test/test.dart';
 
@@ -106,6 +108,90 @@ void main() {
         expect(snapshot.get('address/city'), 'New York');
       });
     });
+
+    group('ModifiableSnapshotView.fromStream', () {
+      test('Lifecycle for non broadcast stream', () async {
+        var listenCalled = false,
+            pauseCalled = false,
+            cancelCalled = false,
+            resumeCalled = false;
+        var controller = StreamController<Snapshot>(onListen: () {
+          listenCalled = true;
+        }, onPause: () {
+          pauseCalled = true;
+        }, onResume: () {
+          resumeCalled = true;
+        }, onCancel: () {
+          cancelCalled = true;
+        });
+        var view = ModifiableSnapshotView.fromStream(controller.stream);
+
+        expect(view.snapshot, isNull);
+        expect(listenCalled, false);
+
+        // listening on onChanged should trigger listen on controller
+        var s = view.onChanged.listen((_) => null);
+        expect(view.snapshot, isNull);
+        expect(listenCalled, true);
+
+        // adding a snapshot to the controller should update the view
+        controller.add(Snapshot.empty());
+        await Future.microtask(() => null);
+        expect(view.snapshot, isNotNull);
+
+        // canceling the subscription onChanged should pause the controller
+        expect(pauseCalled, false);
+        await s.cancel();
+        expect(pauseCalled, true);
+
+        // listening again should trigger resume on controller
+        expect(resumeCalled, false);
+        s = view.onChanged.listen((_) => null);
+        expect(resumeCalled, true);
+
+        // disposing the snapshot view should cancel the controller
+        expect(cancelCalled, false);
+        await view.dispose();
+        expect(cancelCalled, true);
+      });
+      test('Lifecycle for broadcast stream', () async {
+        var listenCalled = false, cancelCalled = false;
+        var controller = StreamController<Snapshot>.broadcast(onListen: () {
+          listenCalled = true;
+        }, onCancel: () {
+          cancelCalled = true;
+        });
+        var view = ModifiableSnapshotView.fromStream(controller.stream);
+
+        expect(view.snapshot, isNull);
+        expect(listenCalled, false);
+
+        // listening on onChanged should trigger listen on controller
+        var s = view.onChanged.listen((_) => null);
+        expect(view.snapshot, isNull);
+        expect(listenCalled, true);
+
+        // adding a snapshot to the controller should update the view
+        controller.add(Snapshot.empty());
+        await Future.microtask(() => null);
+        expect(view.snapshot, isNotNull);
+
+        // canceling the subscription should cancel the controller
+        expect(cancelCalled, false);
+        await s.cancel();
+        expect(cancelCalled, true);
+
+        // listening again should trigger listen on controller
+        listenCalled = false;
+        s = view.onChanged.listen((_) => null);
+        expect(listenCalled, true);
+
+        // disposing the snapshot view should cancel the controller
+        cancelCalled = false;
+        await view.dispose();
+        expect(cancelCalled, true);
+      });
+    });
   });
 }