diff --git a/src/ReactiveUI/Internal/ExpressionChainSink.cs b/src/ReactiveUI/Internal/ExpressionChainSink.cs
index 305f4eec69..0de4e70c82 100644
--- a/src/ReactiveUI/Internal/ExpressionChainSink.cs
+++ b/src/ReactiveUI/Internal/ExpressionChainSink.cs
@@ -81,6 +81,16 @@ private sealed class Sink : IDisposable
/// Whether holds a value yet.
private bool _hasLast;
+ ///
+ /// When , the next emission that equals is suppressed
+ /// regardless of . Set immediately after the kicker emits during
+ /// : any PropertyChanged event that races the
+ /// subscribe-then-read window has been queued behind and will re-emit the
+ /// same value once we release the lock. The suppression collapses that single racing duplicate
+ /// without altering the user-facing distinct semantics.
+ ///
+ private bool _suppressNextIfSameAsLast;
+
/// Latched once this chain subscription has been disposed.
private bool _disposed;
@@ -139,47 +149,85 @@ public void Dispose()
/// The value the link produced (the parent for the next level).
private void SetNextParent(int level, object? value) => _levels[level + 1].SetParent(value);
- /// Handles a leaf raw emission: applies skip-initial, the non-null-parent filter, the cast and the distinct gate.
+ /// Handles a leaf raw emission: applies skip-initial, the non-null-parent filter, the cast, the
+ /// kicker dedup and the distinct gate.
/// Whether the leaf's parent was null (a raw emission that the non-null filter drops).
/// The leaf value when the parent is present.
- private void Emit(bool parentMissing, object? value)
+ /// Whether this emission is the kicker push that
+ /// performs immediately after attaching the link subscription.
+ private void Emit(bool parentMissing, object? value, bool fromKicker = false)
{
if (_skipNext)
{
_skipNext = false;
+ _suppressNextIfSameAsLast = false;
return;
}
if (parentMissing)
{
+ _suppressNextIfSameAsLast = false;
return;
}
- TValue typed;
- if (value is null)
+ if (!TryConvert(value, out var typed))
{
- typed = default!;
- }
- else if (value is TValue cast)
- {
- typed = cast;
- }
- else
- {
- _downstream.OnError(new InvalidCastException($"Unable to cast from {value.GetType()} to {typeof(TValue)}."));
return;
}
- if (_isDistinct && _hasLast && EqualityComparer.Default.Equals(typed, _last))
+ if (ShouldSuppress(typed, fromKicker))
{
return;
}
_last = typed;
_hasLast = true;
+ _suppressNextIfSameAsLast = fromKicker;
_downstream.OnNext(new ObservedChange(_source!, _expression, typed));
}
+ /// Coerces the raw observed value into , forwarding a cast error if it
+ /// is neither nor assignable.
+ /// The raw observed value.
+ /// The coerced value when conversion succeeds; the default otherwise.
+ /// when the value was converted; when an error has
+ /// been forwarded to the downstream observer.
+ private bool TryConvert(object? value, out TValue typed)
+ {
+ if (value is null)
+ {
+ typed = default!;
+ return true;
+ }
+
+ if (value is TValue cast)
+ {
+ typed = cast;
+ return true;
+ }
+
+ _downstream.OnError(new InvalidCastException($"Unable to cast from {value.GetType()} to {typeof(TValue)}."));
+ typed = default!;
+ return false;
+ }
+
+ /// Decides whether the coerced value should be suppressed: collapses a single racing duplicate
+ /// produced by the subscribe-then-read window and then applies the user-facing distinct gate.
+ /// The coerced leaf value.
+ /// Whether this emission is the kicker push (which is never suppressed).
+ /// when the emission must be dropped.
+ private bool ShouldSuppress(TValue typed, bool fromKicker)
+ {
+ if (!fromKicker && _suppressNextIfSameAsLast && _hasLast && EqualityComparer.Default.Equals(typed, _last))
+ {
+ _suppressNextIfSameAsLast = false;
+ return true;
+ }
+
+ _suppressNextIfSameAsLast = false;
+ return _isDistinct && _hasLast && EqualityComparer.Default.Equals(typed, _last);
+ }
+
/// A single chain link's watcher: re-subscribes on parent change and reads the link's value.
/// The owning chain sink.
/// This watcher's position in the chain.
@@ -218,10 +266,14 @@ public void SetParent(object? parent)
var link = sink._links[index];
- // Kicker: propagate the current value immediately, then subscribe for updates.
- Push(ReadValue(parent));
+ // Subscribe-then-read: attach the link subscription before reading the kicker value so
+ // any PropertyChanged that fires between the two steps is queued behind sink._gate
+ // (which our caller holds) rather than silently lost. The Push that follows is marked
+ // as the kicker so the leaf can collapse the single racing duplicate that the queued
+ // notification will deliver once we release the lock.
_subscription.Disposable = sink._notify(parent, link, sink._beforeChange, sink._suppressWarnings)
.Subscribe(new Observer(this));
+ Push(ReadValue(parent), fromKicker: true);
}
///
@@ -274,11 +326,13 @@ public void OnNotification(IObservedChange