switchPath should keep broadcast behavior

CHANGELOG.md

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

lib/src/stream.dart

@@ -70,15 +70,18 @@ extension SnapshotStreamX on Stream<Snapshot> {
   /// [path] whenever this child changes.
   Stream<Snapshot> switchPath(
       String path, Stream<dynamic> Function(Snapshot snapshot) mapper) {
-    var controller = StreamController<Snapshot>(sync: true);
-    var stream = doOnDone(() {
-      controller.close();
+    var controller = BehaviorSubject<Snapshot>(sync: true);
+
+    return doOnData((v) => controller.add(v))
+        .doOnDone(() => controller.close())
+        .map((v) => v.child(path))
+        .distinct()
+        .switchMap((v) {
+      return CombineLatestStream.combine2<Snapshot, dynamic, Snapshot>(
+          controller.stream, mapper(v), (a, b) {
+        return a.setPath(path, b);
+      });
     });
-    return stream.map((v) {
-      controller.add(v);
-      return v;
-    }).asyncSetPath(path,
-        controller.stream.child(path).switchMap((value) => mapper(value)));
   }
 
   /// Updates the content of each child with the value returned by [mapper]

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.3
+version: 0.1.0-dev.4
 homepage: https://github.com/appsup-dart/snapshot
 
 environment:
@@ -13,7 +13,7 @@ dependencies:
   meta: ^1.1.8
   collection: ^1.14.0
   quiver: ^2.1.0
-  rxdart: ^0.24.0
+  rxdart: ^0.25.0
 
 dev_dependencies:
   pedantic: ^1.9.0

test/stream_test.dart

@@ -200,6 +200,20 @@ void main() async {
         ]);
         expect(values, l);
       });
+
+      test('Should keep broadcast behavior', () async {
+        var controller = StreamController.broadcast();
+
+        var stream = controller.stream.toSnapshots();
+
+        stream = stream.switchPath('address', (s) {
+          var v =
+              '${s.child('street').as()} ${s.child('number').as()}, ${s.child('city').as()}';
+          return Stream.value(v);
+        });
+
+        expect(stream.isBroadcast, true);
+      });
     });
 
     group('Stream<Snapshot>.mapChildren', () {