From c669c43f9d488dc306362153476045296144043c Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Sun, 14 Jun 2026 02:48:52 -0700 Subject: [PATCH] fix(observables): serialize the initial emit with concurrent handlers - Run the subscription's read-decision-emit sequence under a per-instance gate, shared with the change handler through a new EmitCurrent. A write from another thread landing between the handler attach and the constructor's state writes previously emitted the same value twice, breaking the no-consecutive-duplicates contract of distinctUntilChanged: true. - Apply the distinct-until-changed test to the initial emit as well, collapsing the duplicate a competing handler leaves behind when it emits first while no value has been recorded yet. The first emission on an ordinary subscribe is unaffected, since no value has been recorded at that point either. - Give PropertyChangingObservable the same shape. The out-of-order case is reachable there too: two writes passing through the constructor's read-to-emit gap deliver the initial, now stale, value after a newer one. - Cover both with forced interleavings driven from the source's event accessor, the property read and the downstream observer, so the schedules are pinned rather than raced for. --- .../Observables/PropertyChangingObservable.cs | 38 +- .../Observables/PropertyObservable.cs | 55 +- ...ObservableInitialEmitSerializationTests.cs | 220 ++++++++ ...ObservableInitialEmitSerializationTests.cs | 522 ++++++++++++++++++ .../TestModels/EmissionRecorder.cs | 116 ++++ 5 files changed, 922 insertions(+), 29 deletions(-) create mode 100644 src/tests/ReactiveUI.Binding.Tests/Observables/PropertyChangingObservableInitialEmitSerializationTests.cs create mode 100644 src/tests/ReactiveUI.Binding.Tests/Observables/PropertyObservableInitialEmitSerializationTests.cs create mode 100644 src/tests/ReactiveUI.Binding.Tests/TestModels/EmissionRecorder.cs diff --git a/src/ReactiveUI.Binding.Shared/Observables/PropertyChangingObservable.cs b/src/ReactiveUI.Binding.Shared/Observables/PropertyChangingObservable.cs index fbc1398..9c8b598 100644 --- a/src/ReactiveUI.Binding.Shared/Observables/PropertyChangingObservable.cs +++ b/src/ReactiveUI.Binding.Shared/Observables/PropertyChangingObservable.cs @@ -60,6 +60,13 @@ internal sealed class Subscription : IDisposable /// The parent observable that owns the source and property metadata. private readonly PropertyChangingObservable _parent; + /// + /// Serializes the initial emit in the constructor with concurrent + /// invocations on other threads, so a racing handler emit and the constructor's initial emit do + /// not interleave on the downstream observer. + /// + private readonly Lock _gate = new(); + /// The downstream observer. Set to on disposal. private IObserver? _observer; @@ -72,10 +79,7 @@ public Subscription(PropertyChangingObservable parent, IObserver observer) _observer = observer; parent._source.PropertyChanging += OnPropertyChanging; - - // Emit initial (StartWith) value - var initial = parent._getter(parent._source); - observer.OnNext(initial!); + EmitCurrent(); } /// @@ -106,14 +110,28 @@ private void OnPropertyChanging(object? sender, PropertyChangingEventArgs e) return; } - var observer = Volatile.Read(ref _observer); - if (observer is null) + EmitCurrent(); + } + + /// + /// Reads the current property value under and forwards it to the downstream + /// observer. Holding across the read-emit pair ensures the constructor's + /// initial emit and any concurrent invocation cannot + /// interleave on the downstream observer. + /// + private void EmitCurrent() + { + lock (_gate) { - return; + var observer = Volatile.Read(ref _observer); + if (observer is null) + { + return; + } + + var value = _parent._getter(_parent._source); + observer.OnNext(value!); } - - var value = _parent._getter(_parent._source); - observer.OnNext(value!); } } } diff --git a/src/ReactiveUI.Binding.Shared/Observables/PropertyObservable.cs b/src/ReactiveUI.Binding.Shared/Observables/PropertyObservable.cs index cd73d66..7d440df 100644 --- a/src/ReactiveUI.Binding.Shared/Observables/PropertyObservable.cs +++ b/src/ReactiveUI.Binding.Shared/Observables/PropertyObservable.cs @@ -68,6 +68,13 @@ internal sealed class Subscription : IDisposable /// The equality comparer used for distinct-until-changed filtering. private readonly EqualityComparer _comparer; + /// + /// Serializes the initial emit in the constructor with concurrent + /// invocations on other threads, so the handler always sees a consistent + /// / snapshot regardless of timing. + /// + private readonly Lock _gate = new(); + /// The downstream observer. Set to on disposal. private IObserver? _observer; @@ -90,12 +97,7 @@ public Subscription(PropertyObservable parent, IObserver observer) _comparer = EqualityComparer.Default; parent._source.PropertyChanged += OnPropertyChanged; - - // Emit initial (StartWith) value - var initial = parent._getter(parent._source); - _lastValue = initial; - _hasValue = true; - observer.OnNext(initial!); + EmitCurrent(); } /// @@ -129,22 +131,37 @@ private void OnPropertyChanged(object? sender, PropertyChangedEventArgs e) return; } - var observer = Volatile.Read(ref _observer); - if (observer is null) - { - return; - } - - var value = _parent._getter(_parent._source); + EmitCurrent(); + } - if (_parent._distinctUntilChanged && _hasValue && _comparer.Equals(value!, _lastValue!)) + /// + /// Reads the current property value under and forwards it to the downstream + /// observer when the distinct-until-changed gate allows. Holding across the + /// read-decision-emit sequence ensures the constructor's initial emit and any concurrent + /// invocation cannot interleave on the downstream observer or + /// publish a duplicate when both see the same current value. + /// + private void EmitCurrent() + { + lock (_gate) { - return; + var observer = Volatile.Read(ref _observer); + if (observer is null) + { + return; + } + + var value = _parent._getter(_parent._source); + + if (_parent._distinctUntilChanged && _hasValue && _comparer.Equals(value!, _lastValue!)) + { + return; + } + + _lastValue = value; + _hasValue = true; + observer.OnNext(value!); } - - _lastValue = value; - _hasValue = true; - observer.OnNext(value!); } } } diff --git a/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyChangingObservableInitialEmitSerializationTests.cs b/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyChangingObservableInitialEmitSerializationTests.cs new file mode 100644 index 0000000..e5ffd2e --- /dev/null +++ b/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyChangingObservableInitialEmitSerializationTests.cs @@ -0,0 +1,220 @@ +// Copyright (c) 2019-2026 ReactiveUI Association Incorporated. All rights reserved. +// ReactiveUI Association Incorporated licenses this file to you under the MIT license. +// See the LICENSE file in the project root for full license information. + +using System.ComponentModel; +using ReactiveUI.Binding.Observables; +using ReactiveUI.Binding.Tests.TestModels; + +namespace ReactiveUI.Binding.Tests.Observables; + +/// +/// Tests that serializes the initial emit it performs while +/// the subscription is still being built against +/// notifications arriving at the same time. +/// +/// +/// The window under test is the gap between the constructor reading the property and delivering that +/// read downstream. A thread can be descheduled there, and these tests force that schedule by driving +/// the competing writes from inside the property read itself. +/// +public class PropertyChangingObservableInitialEmitSerializationTests +{ + /// The value the second competing write stores. + private const int SecondWrite = 2; + + /// + /// A source that raises before it writes - the conventional shape - still lets a stale initial emit + /// land after a newer one once two writes pass through the constructor's read-to-emit gap. The + /// serialization is therefore load-bearing on this type, not merely a consistency measure. + /// + /// A representing the asynchronous unit test. + [Test] + public async Task Subscribe_ConventionalSourceWritesDuringInitialEmit_DoesNotDeliverAStaleValueLast() + { + var source = new RaiseThenWriteViewModel { Version = 0 }; + var recorder = new EmissionRecorder(); + + using var competitor = new InitialEmitCompetitor(version => source.Version = version); + + var observable = new PropertyChangingObservable( + source, + nameof(RaiseThenWriteViewModel.Version), + competitor.CreateContendedRead(() => source.Version)); + + using (observable.Subscribe(recorder)) + { + competitor.WaitForCompletion(); + + // The competing emits are held behind the initial emit, so the initial 0 lands first and the + // pre-change values that follow it only ever move forward. + await AssertSequence(recorder.Snapshot(), 0, 0, 1); + } + } + + /// + /// A source that writes before raising its before-change event - the shape the change calls + /// atypical - is the case the serialization is claimed to guard. Without it the handler emits both + /// written values before the initial emit lands, leaving the oldest value last. + /// + /// A representing the asynchronous unit test. + [Test] + public async Task Subscribe_AtypicalSourceWritesBeforeRaising_DoesNotDeliverAStaleValueLast() + { + var source = new WriteThenRaiseViewModel { Version = 0 }; + var recorder = new EmissionRecorder(); + + using var competitor = new InitialEmitCompetitor(version => source.Version = version); + + var observable = new PropertyChangingObservable( + source, + nameof(WriteThenRaiseViewModel.Version), + competitor.CreateContendedRead(() => source.Version)); + + using (observable.Subscribe(recorder)) + { + competitor.WaitForCompletion(); + + await AssertSequence(recorder.Snapshot(), 0, 1, SecondWrite); + } + } + + /// + /// The initial emit stays unconditional on an ordinary subscribe, including for a value equal to the + /// default for its type, and every subscriber receives its own. + /// + /// A representing the asynchronous unit test. + [Test] + public async Task Subscribe_NoConcurrentNotification_EmitsTheInitialValueToEverySubscriber() + { + var source = new RaiseThenWriteViewModel { Version = 0 }; + var observable = new PropertyChangingObservable( + source, + nameof(RaiseThenWriteViewModel.Version), + static x => ((RaiseThenWriteViewModel)x).Version); + + var first = new EmissionRecorder(); + var second = new EmissionRecorder(); + + using (observable.Subscribe(first)) + using (observable.Subscribe(second)) + { + await AssertSequence(first.Snapshot(), 0); + await AssertSequence(second.Snapshot(), 0); + } + } + + /// Asserts that a recorded emission sequence matches the expected one exactly. + /// The recorded emissions. + /// The emissions the subscription is required to produce, in order. + /// A representing the assertion. + private static async Task AssertSequence(IReadOnlyList actual, params int[] expected) + { + await Assert.That(actual).Count().IsEqualTo(expected.Length); + + for (var index = 0; index < expected.Length; index++) + { + await Assert.That(actual[index]).IsEqualTo(expected[index]); + } + } + + /// + /// Drives two competing property writes from inside the subscription's initial property read, so + /// both notifications fall in the constructor's read-to-emit gap. + /// + private sealed class InitialEmitCompetitor : IDisposable + { + /// + /// How long to give the competing thread to complete its emits while the initial read and emit + /// are still in progress. A serialized subscription blocks it for the whole window, so the wait + /// always expires; an unserialized one lets it run to completion in microseconds. + /// + private const int InterleaveWindowMilliseconds = 500; + + /// Signals that the competing thread is running and about to write. + private readonly ManualResetEventSlim _started = new(false); + + /// The thread performing the competing writes. + private readonly Thread _thread; + + /// Whether the contention has been driven already, so later reads run plainly. + private bool _contended; + + /// Initializes a new instance of the class. + /// Writes the observed property. + public InitialEmitCompetitor(Action write) + { + _thread = new(() => + { + _started.Set(); + write(1); + write(SecondWrite); + }) { IsBackground = true }; + } + + /// + /// Builds a property read that, the first time it runs, releases the competing thread and holds + /// for it before returning the value read on entry. + /// + /// Reads the current property value. + /// The property read to hand to the observable. + public Func CreateContendedRead(Func read) => source => + { + var valueOnEntry = read(); + + if (_contended) + { + return valueOnEntry; + } + + _contended = true; + _thread.Start(); + _started.Wait(); + _ = _thread.Join(InterleaveWindowMilliseconds); + + return valueOnEntry; + }; + + /// Waits for the competing writes to finish. + public void WaitForCompletion() => _thread.Join(); + + /// + public void Dispose() => _started.Dispose(); + } + + /// A view model with the conventional before-change ordering: raise, then write. + private sealed class RaiseThenWriteViewModel : INotifyPropertyChanging + { + /// + public event PropertyChangingEventHandler? PropertyChanging; + + /// Gets or sets the observed property, raising the event before the write lands. + public int Version + { + get; + set + { + PropertyChanging?.Invoke(this, new(nameof(Version))); + field = value; + } + } + } + + /// A view model that writes before raising its before-change event, which is atypical. + private sealed class WriteThenRaiseViewModel : INotifyPropertyChanging + { + /// + public event PropertyChangingEventHandler? PropertyChanging; + + /// Gets or sets the observed property, raising the event after the write has landed. + public int Version + { + get; + set + { + field = value; + PropertyChanging?.Invoke(this, new(nameof(Version))); + } + } + } +} diff --git a/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyObservableInitialEmitSerializationTests.cs b/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyObservableInitialEmitSerializationTests.cs new file mode 100644 index 0000000..eb18798 --- /dev/null +++ b/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyObservableInitialEmitSerializationTests.cs @@ -0,0 +1,522 @@ +// Copyright (c) 2019-2026 ReactiveUI Association Incorporated. All rights reserved. +// ReactiveUI Association Incorporated licenses this file to you under the MIT license. +// See the LICENSE file in the project root for full license information. + +using System.ComponentModel; +using ReactiveUI.Binding.Observables; +using ReactiveUI.Binding.Tests.TestModels; + +namespace ReactiveUI.Binding.Tests.Observables; + +/// +/// Tests that serializes the initial emit it performs while the +/// subscription is still being built against +/// notifications arriving at the same time. +/// +/// +/// Every interleaving here is forced rather than raced for: the source's event accessor, the property +/// read and the downstream observer are each used as a hook to drive a notification into one specific +/// point of the subscription's construction. Those are the same points a thread can be descheduled at, +/// so the schedules are real ones - they are only made to happen every run instead of occasionally. +/// +public class PropertyObservableInitialEmitSerializationTests +{ + /// The property value present before any competing write. + private const string InitialName = "Alice"; + + /// The property value a competing thread writes. + private const string UpdatedName = "Bob"; + + /// + /// How long to give a competing thread to complete its emit while the initial emit is still on the + /// stack. A serialized subscription blocks that thread for the whole window, so the wait always + /// expires; an unserialized one lets it through in microseconds. + /// + private const int InterleaveWindowMilliseconds = 500; + + /// + /// Subscriptions the unforced sweep builds. Sized from measurement: against unserialized code this + /// count reports the defect on every target framework with a wide margin, where a tenth of it misses + /// on some of them because the early iterations run before the loop is fully optimized. + /// + private const int SweepIterations = 10_000; + + /// + /// The reported defect: a notification landing after the handler is attached but before the + /// constructor has read and emitted delivers the same value twice, breaking the + /// no-consecutive-duplicates contract the caller asked for with distinctUntilChanged: true. + /// + /// A representing the asynchronous unit test. + [Test] + public async Task Subscribe_PropertyChangedRaisedOnAnotherThreadWhileAttaching_EmitsTheValueOnce() + { + var source = new HookedViewModel { Name = InitialName }; + var recorder = new EmissionRecorder(); + + // Drive a whole mutation through the source the instant the handler is attached, so the handler + // has already emitted by the time the constructor reads the property. + source.AfterHandlerAttached = () => RunToCompletionOnAnotherThread(() => source.Name = UpdatedName); + + var observable = new PropertyObservable( + source, + nameof(HookedViewModel.Name), + static x => ((HookedViewModel)x).Name, + distinctUntilChanged: true); + + using (observable.Subscribe(recorder)) + { + await AssertNoErrors(recorder); + await AssertSequence(recorder.Snapshot(), UpdatedName); + } + } + + /// + /// The same defect reached re-entrantly rather than across threads: the property read the + /// constructor performs for its initial emit itself raises + /// , so the handler runs part-way through + /// construction on the subscribing thread. This also pins that the serialization is re-entrant, + /// since a non-re-entrant gate would deadlock here rather than fail. + /// + /// A representing the asynchronous unit test. + [Test] + public async Task Subscribe_PropertyChangedRaisedReentrantlyDuringInitialRead_EmitsTheValueOnce() + { + var source = new HookedViewModel { Name = InitialName }; + var recorder = new EmissionRecorder(); + var raised = false; + + string? ReadAndNotifyOnce(INotifyPropertyChanged instance) + { + if (!raised) + { + raised = true; + source.RaisePropertyChanged(nameof(HookedViewModel.Name)); + } + + return ((HookedViewModel)instance).Name; + } + + var observable = new PropertyObservable( + source, + nameof(HookedViewModel.Name), + ReadAndNotifyOnce, + distinctUntilChanged: true); + + using (observable.Subscribe(recorder)) + { + await AssertNoErrors(recorder); + await AssertSequence(recorder.Snapshot(), InitialName); + } + } + + /// + /// Pins the change's central claim - that a competing handler runs either wholly before or wholly + /// after the initial emit, never inside it - by holding a competing thread against the initial emit + /// while it is on the stack and recording whether its emit overlaps. + /// + /// A representing the asynchronous unit test. + [Test] + public async Task Subscribe_PropertyChangedRaisedOnAnotherThreadDuringInitialEmit_DoesNotOverlapTheInitialEmit() + { + var source = new HookedViewModel { Name = InitialName }; + using var competitorStarted = new ManualResetEventSlim(false); + Thread? competitor = null; + + // Runs from inside the downstream call of the initial emit, which is the window the change keeps + // exclusive. The bounded join is what a blocked competing thread looks like from in here. + var recorder = new EmissionRecorder + { + OnFirstValue = () => + { + competitor = new(() => + { + competitorStarted.Set(); + source.Name = UpdatedName; + }) { IsBackground = true }; + + competitor.Start(); + competitorStarted.Wait(); + _ = competitor.Join(InterleaveWindowMilliseconds); + }, + }; + + var observable = new PropertyObservable( + source, + nameof(HookedViewModel.Name), + static x => ((HookedViewModel)x).Name, + distinctUntilChanged: true); + + using (observable.Subscribe(recorder)) + { + competitor!.Join(); + + await AssertNoErrors(recorder); + await Assert.That(recorder.MaxConcurrentEmissions).IsEqualTo(1); + await AssertSequence(recorder.Snapshot(), InitialName, UpdatedName); + } + } + + /// + /// The initial emit stays unconditional on an ordinary subscribe, including when the value equals + /// the default for its type. The change applies the distinct-until-changed test to the initial emit + /// as well, so this guards the first emission against being swallowed by that new test. + /// + /// Whether the subscription suppresses consecutive duplicates. + /// A representing the asynchronous unit test. + [Test] + [Arguments(true)] + [Arguments(false)] + public async Task Subscribe_ValueIsTheTypeDefault_StillEmitsTheInitialValue(bool distinctUntilChanged) + { + var source = new HookedViewModel { Name = null, Count = 0 }; + + var nameRecorder = new EmissionRecorder(); + var nameObservable = new PropertyObservable( + source, + nameof(HookedViewModel.Name), + static x => ((HookedViewModel)x).Name, + distinctUntilChanged); + + var countRecorder = new EmissionRecorder(); + var countObservable = new PropertyObservable( + source, + nameof(HookedViewModel.Count), + static x => ((HookedViewModel)x).Count, + distinctUntilChanged); + + using (nameObservable.Subscribe(nameRecorder)) + using (countObservable.Subscribe(countRecorder)) + { + await Assert.That(nameRecorder.Snapshot()).Count().IsEqualTo(1); + await Assert.That(nameRecorder.Snapshot()[0]).IsNull(); + await AssertSequence(countRecorder.Snapshot(), 0); + } + } + + /// + /// Each subscription carries its own distinct-until-changed state, so a later subscriber still + /// receives the initial value even though an existing subscriber has already been given it, and a + /// re-subscription after disposal receives it again. + /// + /// Whether the subscription suppresses consecutive duplicates. + /// A representing the asynchronous unit test. + [Test] + [Arguments(true)] + [Arguments(false)] + public async Task Subscribe_ValueAlreadyDeliveredToAnotherSubscriber_StillEmitsTheInitialValue(bool distinctUntilChanged) + { + var source = new HookedViewModel { Name = InitialName }; + var observable = new PropertyObservable( + source, + nameof(HookedViewModel.Name), + static x => ((HookedViewModel)x).Name, + distinctUntilChanged); + + var first = new EmissionRecorder(); + var second = new EmissionRecorder(); + var afterDisposal = new EmissionRecorder(); + + using (observable.Subscribe(first)) + using (observable.Subscribe(second)) + { + await AssertSequence(first.Snapshot(), InitialName); + await AssertSequence(second.Snapshot(), InitialName); + } + + using (observable.Subscribe(afterDisposal)) + { + await AssertSequence(afterDisposal.Snapshot(), InitialName); + } + } + + /// + /// An unforced sweep over the same window, reaching interleavings the forced tests do not enumerate. + /// A competing thread writes a fresh value continuously while a subscription is built against it, so + /// the subscribe always lands mid-storm, and no subscriber may ever see two equal values in a row. + /// + /// A representing the asynchronous unit test. + [Test] + public async Task Subscribe_ConcurrentWritesThroughoutSubscription_NeverEmitsConsecutiveDuplicates() + { + var iterationsThatEmitted = 0; + var iterationsWithDuplicate = 0; + + for (var iteration = 0; iteration < SweepIterations; iteration++) + { + var source = new HookedViewModel { Count = 0 }; + using var competitorStarted = new ManualResetEventSlim(false); + var stopped = 0; + + var competitor = new Thread(() => + { + var next = 0; + competitorStarted.Set(); + while (Volatile.Read(ref stopped) == 0) + { + source.Count = ++next; + } + }) { IsBackground = true }; + + competitor.Start(); + competitorStarted.Wait(); + + var recorder = new EmissionRecorder(); + var observable = new PropertyObservable( + source, + nameof(HookedViewModel.Count), + static x => ((HookedViewModel)x).Count, + distinctUntilChanged: true); + + using (observable.Subscribe(recorder)) + { + Volatile.Write(ref stopped, 1); + competitor.Join(); + } + + var observed = recorder.Snapshot(); + if (observed.Count > 0) + { + iterationsThatEmitted++; + } + + for (var index = 1; index < observed.Count; index++) + { + if (observed[index] != observed[index - 1]) + { + continue; + } + + iterationsWithDuplicate++; + break; + } + } + + await Assert.That(iterationsWithDuplicate).IsEqualTo(0); + await Assert.That(iterationsThatEmitted).IsEqualTo(SweepIterations); + } + + /// + /// A source raises its event off a handler snapshot taken before the subscription attached, so the + /// notification for that write never reaches the new handler. On the conventional write-then-raise + /// setter the value is still not lost: the constructor reads the property after attaching, and that + /// read is the backstop. + /// + /// A representing the asynchronous unit test. + [Test] + public async Task Subscribe_HandlerAttachesAfterTheRaiserSnapshotsItsDelegate_StillObservesTheWrittenValue() + { + var source = new ConventionalRaiseOrderViewModel { Name = InitialName }; + var recorder = new EmissionRecorder(); + using var subscriptions = new GrowableCompositeDisposable(); + + source.BeforeRaise = () => subscriptions.Add(new PropertyObservable( + source, + nameof(ConventionalRaiseOrderViewModel.Name), + static x => ((ConventionalRaiseOrderViewModel)x).Name, + distinctUntilChanged: true).Subscribe(recorder)); + + source.Name = UpdatedName; + + await Assert.That(source.HandlerWasInvoked).IsFalse(); + await AssertSequence(recorder.Snapshot(), UpdatedName); + } + + /// + /// The same missed notification against a source that raises before it writes. The backstop read + /// runs too early there, so the subscriber is left holding the pre-write value - a gap owned by the + /// source's ordering rather than by the subscription, and one no amount of serialization closes. + /// + /// A representing the asynchronous unit test. + [Test] + public async Task Subscribe_SourceRaisesPropertyChangedBeforeWritingTheField_ObservesThePreWriteValue() + { + var source = new RaiseBeforeWriteViewModel { Name = InitialName }; + var recorder = new EmissionRecorder(); + using var subscriptions = new GrowableCompositeDisposable(); + + source.BeforeRaise = () => subscriptions.Add(new PropertyObservable( + source, + nameof(RaiseBeforeWriteViewModel.Name), + static x => ((RaiseBeforeWriteViewModel)x).Name, + distinctUntilChanged: true).Subscribe(recorder)); + + source.Name = UpdatedName; + + await Assert.That(source.HandlerWasInvoked).IsFalse(); + await Assert.That(source.Name).IsEqualTo(UpdatedName); + await AssertSequence(recorder.Snapshot(), InitialName); + } + + /// Asserts that nothing was pushed to the observer's error channel. + /// The emitted element type. + /// The recorder to inspect. + /// A representing the assertion. + private static async Task AssertNoErrors(EmissionRecorder recorder) => + await Assert.That(recorder.ErrorSnapshot()).IsEmpty(); + + /// Asserts that a recorded emission sequence matches the expected one exactly. + /// The emitted element type. + /// The recorded emissions. + /// The emissions the subscription is required to produce, in order. + /// A representing the assertion. + private static async Task AssertSequence(IReadOnlyList actual, params T[] expected) + { + await Assert.That(actual).Count().IsEqualTo(expected.Length); + + for (var index = 0; index < expected.Length; index++) + { + await Assert.That(actual[index]).IsEqualTo(expected[index]); + } + } + + /// Runs an action on a dedicated thread and waits for it to finish. + /// The work to run away from the calling thread. + private static void RunToCompletionOnAnotherThread(Action action) + { + var thread = new Thread(action.Invoke) { IsBackground = true }; + thread.Start(); + thread.Join(); + } + + /// + /// A view model whose event accessor exposes the instant a handler becomes attached, which is where + /// the subscribe-time window opens. + /// + private sealed class HookedViewModel : INotifyPropertyChanged + { + /// Serializes handler registration against deregistration. + private readonly Lock _handlerGate = new(); + + /// The registered handlers. + private PropertyChangedEventHandler? _handlers; + + /// + public event PropertyChangedEventHandler? PropertyChanged + { + add + { + lock (_handlerGate) + { + _handlers += value; + } + + var hook = AfterHandlerAttached; + AfterHandlerAttached = null; + hook?.Invoke(); + } + + remove + { + lock (_handlerGate) + { + _handlers -= value; + } + } + } + + /// Gets or sets a one-shot action run once a handler is added, from the adding thread. + public Action? AfterHandlerAttached { get; set; } + + /// Gets or sets the observed reference-typed property, written before the event is raised. + public string? Name + { + get; + set + { + field = value; + RaisePropertyChanged(nameof(Name)); + } + } + + /// Gets or sets a value-typed property used to check the default-valued initial emit. + public int Count + { + get; + set + { + field = value; + RaisePropertyChanged(nameof(Count)); + } + } + + /// Raises without writing anything. + /// The property to report. + public void RaisePropertyChanged(string propertyName) => _handlers?.Invoke(this, new(propertyName)); + } + + /// + /// A view model with the conventional raise order - write the field, snapshot the handlers, raise - + /// exposing the point between the snapshot and the raise. + /// + private sealed class ConventionalRaiseOrderViewModel : INotifyPropertyChanged + { + /// + public event PropertyChangedEventHandler? PropertyChanged; + + /// Gets or sets a one-shot action run after the handler snapshot is taken and before it is invoked. + public Action? BeforeRaise { get; set; } + + /// Gets a value indicating whether the raise reached any handler. + public bool HandlerWasInvoked { get; private set; } + + /// Gets or sets the observed property. + public string? Name + { + get; + set + { + field = value; + + var snapshot = PropertyChanged; + var hook = BeforeRaise; + BeforeRaise = null; + hook?.Invoke(); + + if (snapshot is null) + { + return; + } + + HandlerWasInvoked = true; + snapshot(this, new(nameof(Name))); + } + } + } + + /// + /// A view model that raises before it writes, which is the ordering under which the subscription's + /// post-attach read cannot serve as a backstop. + /// + private sealed class RaiseBeforeWriteViewModel : INotifyPropertyChanged + { + /// + public event PropertyChangedEventHandler? PropertyChanged; + + /// Gets or sets a one-shot action run after the handler snapshot is taken and before it is invoked. + public Action? BeforeRaise { get; set; } + + /// Gets a value indicating whether the raise reached any handler. + public bool HandlerWasInvoked { get; private set; } + + /// Gets or sets the observed property. + public string? Name + { + get; + set + { + var snapshot = PropertyChanged; + var hook = BeforeRaise; + BeforeRaise = null; + hook?.Invoke(); + + if (snapshot is not null) + { + HandlerWasInvoked = true; + snapshot(this, new(nameof(Name))); + } + + field = value; + } + } + } +} diff --git a/src/tests/ReactiveUI.Binding.Tests/TestModels/EmissionRecorder.cs b/src/tests/ReactiveUI.Binding.Tests/TestModels/EmissionRecorder.cs new file mode 100644 index 0000000..62f5618 --- /dev/null +++ b/src/tests/ReactiveUI.Binding.Tests/TestModels/EmissionRecorder.cs @@ -0,0 +1,116 @@ +// Copyright (c) 2019-2026 ReactiveUI Association Incorporated. All rights reserved. +// ReactiveUI Association Incorporated licenses this file to you under the MIT license. +// See the LICENSE file in the project root for full license information. + +namespace ReactiveUI.Binding.Tests.TestModels; + +/// +/// An observer that records what it was given and how the calls overlapped, for tests that need to tell +/// a serialized producer from one whose emissions can run concurrently on the downstream observer. +/// +/// The emitted element type. +internal sealed class EmissionRecorder : IObserver +{ + /// Serializes the recorded state against emissions arriving on several threads. + private readonly Lock _gate = new(); + + /// The values received so far, in arrival order. + private readonly List _values = []; + + /// The errors received so far. + private readonly List _errors = []; + + /// The number of emissions currently in flight. + private int _inFlight; + + /// The high-water mark of . + private int _maxInFlight; + + /// + /// Gets or sets a one-shot action run from inside the first emission, after the value has been + /// recorded and while that call still counts as in flight. It runs outside this recorder's own lock + /// so that what it measures is the producer's serialization rather than the recorder's. + /// + internal Action? OnFirstValue { get; set; } + + /// + /// Gets the greatest number of emissions that were ever in flight at once. Above one means the + /// producer let two emissions overlap on the downstream observer. + /// + internal int MaxConcurrentEmissions + { + get + { + lock (_gate) + { + return _maxInFlight; + } + } + } + + /// + public void OnNext(T value) + { + Action? hook; + + lock (_gate) + { + _inFlight++; + if (_inFlight > _maxInFlight) + { + _maxInFlight = _inFlight; + } + + _values.Add(value); + + hook = _values.Count == 1 ? OnFirstValue : null; + OnFirstValue = null; + } + + try + { + hook?.Invoke(); + } + finally + { + lock (_gate) + { + _inFlight--; + } + } + } + + /// + public void OnError(Exception error) + { + lock (_gate) + { + _errors.Add(error); + } + } + + /// + public void OnCompleted() + { + } + + /// Takes a copy of the values recorded so far. + /// The values received, in arrival order. + internal IReadOnlyList Snapshot() + { + lock (_gate) + { + return [.. _values]; + } + } + + /// Takes a copy of the errors recorded so far. + /// The errors received, in arrival order. + internal IReadOnlyList ErrorSnapshot() + { + lock (_gate) + { + return [.. _errors]; + } + } +}