Ensure source subscription is disposed

SainsburyWellcomeCentre · Oct 23, 2023 · 26cadeb · 26cadeb
1 parent 4ffa38c
commit 26cadeb
Showing 1 changed file with 4 additions and 2 deletions.
diff --git a/src/Aeon.Acquisition/ObservableExtensions.cs b/src/Aeon.Acquisition/ObservableExtensions.cs
@@ -60,8 +60,9 @@ public static IObservable<Unit> MergeUnit<TSource, TOther>(this IObservable<TSou
             {
                 var pc = clock.Publish();
                 var ps = source.Publish();
+                var sourceSubscription = new SingleAssignmentDisposable();
                 var trigger = Observer.Create<HarpMessage>(
-                    _ => ps.Connect(),
+                    _ => sourceSubscription.Disposable = ps.Connect(),
                     observer.OnError);
                 var result = ps.CombineLatest(pc, (data, message) => (data, message))
                                .Sample(ps.MergeUnit(pc.Take(1)))
@@ -73,7 +74,8 @@ public static IObservable<Unit> MergeUnit<TSource, TOther>(this IObservable<TSou
                 return new CompositeDisposable(
                     result.SubscribeSafe(observer),
                     pc.Take(1).SubscribeSafe(trigger),
-                    pc.Connect());
+                    pc.Connect(),
+                    sourceSubscription);
             });
         }
     }