From 26cadeb12a112bc9b162690dded938fcc7dc1d4e Mon Sep 17 00:00:00 2001
From: glopesdev <glopesdev@users.noreply.github.com>
Date: Tue, 24 Oct 2023 00:40:24 +0100
Subject: [PATCH] Ensure source subscription is disposed

---
 src/Aeon.Acquisition/ObservableExtensions.cs | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/src/Aeon.Acquisition/ObservableExtensions.cs b/src/Aeon.Acquisition/ObservableExtensions.cs
index 2afbbdb..92d3363 100644
--- 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);
             });
         }
     }