Avoid blocking WithLatestFrom value updates (#2630)

Reimplement WithLatestFrom on a two-lock strategy, one gate serializing
downstream delivery and a separate lock guarding only the cached latest
value, so updates from the value sequence no longer wait on downstream
processing of a source element. A slow downstream observer now delays
only further source elements, not refreshes of the latest value, for
example when software-timestamping a data sequence against a high-rate
clock. Observable behavior is otherwise unchanged; the two-lock logic
follows the current dotnet/reactive WithLatestFrom.

Closes #2600

Author: glopesdev

Bonsai.Core.Tests/WithLatestFromTests.cs

@@ -0,0 +1,136 @@
+using System;
+using System.Collections.Generic;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
+using Bonsai.Reactive;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+namespace Bonsai.Core.Tests
+{
+    [TestClass]
+    public class WithLatestFromTests
+    {
+        static IObservable<Tuple<TSource, TOther>> Process<TSource, TOther>(
+            IObservable<TSource> source,
+            IObservable<TOther> other)
+        {
+            return new WithLatestFrom().Process(source, other);
+        }
+
+        [TestMethod]
+        public void Process_SourceEmitsAfterValueSeen_EmitsPairWithLatestValue()
+        {
+            var source = new Subject<int>();
+            var other = new Subject<string>();
+            var results = new List<Tuple<int, string>>();
+            using (Process(source, other).Subscribe(results.Add))
+            {
+                other.OnNext("a");
+                source.OnNext(1);
+                other.OnNext("b");
+                source.OnNext(2);
+            }
+
+            CollectionAssert.AreEqual(
+                new[] { Tuple.Create(1, "a"), Tuple.Create(2, "b") },
+                results);
+        }
+
+        [TestMethod]
+        public void Process_SourceCompletes_CompletesResult()
+        {
+            var source = new Subject<int>();
+            var other = new Subject<string>();
+            var completed = false;
+            using (Process(source, other).Subscribe(_ => { }, () => completed = true))
+            {
+                other.OnNext("a");
+                source.OnCompleted();
+            }
+
+            Assert.IsTrue(completed);
+        }
+
+        [TestMethod]
+        public void Process_ValueSequenceCompletes_ContinuesWithLastValue()
+        {
+            var source = new Subject<int>();
+            var other = new Subject<string>();
+            var results = new List<Tuple<int, string>>();
+            var completed = false;
+            using (Process(source, other).Subscribe(results.Add, () => completed = true))
+            {
+                other.OnNext("a");
+                other.OnCompleted();
+                source.OnNext(1);
+                source.OnNext(2);
+            }
+
+            Assert.IsFalse(completed);
+            CollectionAssert.AreEqual(
+                new[] { Tuple.Create(1, "a"), Tuple.Create(2, "a") },
+                results);
+        }
+
+        [TestMethod]
+        public void Process_SourceEmitsBeforeAnyValue_DropsNotification()
+        {
+            var source = new Subject<int>();
+            var other = new Subject<string>();
+            var results = new List<Tuple<int, string>>();
+            using (Process(source, other).Subscribe(results.Add))
+            {
+                source.OnNext(1);
+                other.OnNext("a");
+                source.OnNext(2);
+            }
+
+            CollectionAssert.AreEqual(
+                new[] { Tuple.Create(2, "a") },
+                results);
+        }
+
+        [TestMethod]
+        public void Process_SourceErrors_FaultsDownstream()
+        {
+            var source = new Subject<int>();
+            var other = new Subject<string>();
+            var error = new InvalidOperationException();
+            Exception observed = null;
+            using (Process(source, other).Subscribe(_ => { }, ex => observed = ex))
+            {
+                other.OnNext("a");
+                source.OnError(error);
+            }
+
+            Assert.AreSame(error, observed);
+        }
+
+        [TestMethod]
+        public void Process_ValueSequenceErrors_FaultsDownstream()
+        {
+            var source = new Subject<int>();
+            var other = new Subject<string>();
+            var error = new InvalidOperationException();
+            Exception observed = null;
+            using (Process(source, other).Subscribe(_ => { }, ex => observed = ex))
+            {
+                other.OnError(error);
+            }
+
+            Assert.AreSame(error, observed);
+        }
+
+        [TestMethod]
+        public void Process_ValueAvailableSynchronously_EmitsOnImmediateSourceElement()
+        {
+            var source = Observable.Return(1);
+            var other = Observable.Return("a");
+            var results = new List<Tuple<int, string>>(Process(source, other).ToList().Wait());
+
+            CollectionAssert.AreEqual(
+                new[] { Tuple.Create(1, "a") },
+                results);
+        }
+    }
+}

Bonsai.Core/Reactive/WithLatestFrom.cs

@@ -1,6 +1,9 @@
 using System;
 using System.ComponentModel;
+using System.Reactive;
+using System.Reactive.Disposables;
 using System.Reactive.Linq;
+using System.Threading;
 using System.Xml.Serialization;
 
 namespace Bonsai.Reactive
@@ -35,10 +38,70 @@ public IObservable<Tuple<TSource, TOther>> Process<TSource, TOther>(
             IObservable<TSource> source,
             IObservable<TOther> other)
         {
-            return source.Publish(ps =>
-                ps.CombineLatest(other, (xs, ys) => Tuple.Create(xs, ys))
-                  .Sample(ps)
-                  .TakeUntil(ps.IgnoreElements().LastOrDefaultAsync()));
+            return Observable.Create<Tuple<TSource, TOther>>(observer =>
+            {
+                var gate = new object();
+                var latestGate = new object();
+                var hasLatest = false;
+                var latest = default(TOther);
+                var otherDisposable = new SingleAssignmentDisposable();
+                var otherObserver = Observer.Create<TOther>(
+                    value =>
+                    {
+                        lock (latestGate)
+                        {
+                            latest = value;
+                        }
+
+                        if (!Volatile.Read(ref hasLatest))
+                        {
+                            Volatile.Write(ref hasLatest, true);
+                        }
+                    },
+                    error =>
+                    {
+                        lock (gate)
+                        {
+                            observer.OnError(error);
+                        }
+                    },
+                    otherDisposable.Dispose);
+                otherDisposable.Disposable = other.SubscribeSafe(otherObserver);
+
+                var sourceObserver = Observer.Create<TSource>(
+                    value =>
+                    {
+                        if (Volatile.Read(ref hasLatest))
+                        {
+                            TOther latestValue;
+                            lock (latestGate)
+                            {
+                                latestValue = latest;
+                            }
+
+                            lock (gate)
+                            {
+                                observer.OnNext(Tuple.Create(value, latestValue));
+                            }
+                        }
+                    },
+                    error =>
+                    {
+                        lock (gate)
+                        {
+                            observer.OnError(error);
+                        }
+                    },
+                    () =>
+                    {
+                        lock (gate)
+                        {
+                            observer.OnCompleted();
+                        }
+                    });
+                var sourceDisposable = source.SubscribeSafe(sourceObserver);
+                return new CompositeDisposable(otherDisposable, sourceDisposable);
+            });
         }
     }
 }