Skip to content
Prev Previous commit
Next Next commit
further work
  • Loading branch information
glennawatson committed Sep 12, 2026
commit 2e3ea0d933d60a23c31b9e8a45da6856dce95567
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,9 @@ namespace ReactiveUI.Primitives.Blazor.Components;
#endif

/// <summary>Base component that tracks reactive subscriptions and refreshes through Blazor's renderer dispatcher.</summary>
/// <remarks>Every observed callback is marshalled onto the renderer dispatcher, so a handler can touch component state
/// directly, and a callback that throws is dispatched to the enclosing error boundary instead of faulting a detached
/// task. Disposing the component disposes every tracked subscription.</remarks>
[System.Diagnostics.DebuggerDisplay("{DebuggerDisplay,nq}")]
public class ReactiveComponentBase : ComponentBase, IDisposable
{
Expand Down Expand Up @@ -44,7 +47,7 @@ public void Dispose()

/// <summary>Tracks a subscription so it is disposed when the component is disposed.</summary>
/// <param name="subscription">The subscription to track.</param>
/// <returns>The supplied subscription, or <see cref="EmptyDisposable.Instance"/> when the component has already been disposed.</returns>
/// <returns>The supplied subscription, or <see cref="EmptyDisposable.Instance"/> when the component is disposed.</returns>
/// <exception cref="ArgumentNullException"><paramref name="subscription"/> is <see langword="null"/>.</exception>
protected IDisposable Track(IDisposable subscription)
{
Expand Down Expand Up @@ -136,7 +139,7 @@ protected IDisposable Observe<T>(
[MethodImpl(MethodImplOptions.AggressiveInlining)]
protected Task InvalidateAsync() => InvokeAsync(StateHasChanged);

/// <summary>Handles an unhandled subscription error.</summary>
/// <summary>Handles a subscription error that no <c>onError</c> callback was given for.</summary>
/// <param name="error">The observed error.</param>
/// <exception cref="InvalidOperationException">Always thrown to surface the subscription error.</exception>
protected virtual void OnObservedError(Exception error)
Expand Down Expand Up @@ -174,7 +177,7 @@ private async Task InvokeGuardedAsync(Action callback)
}
}

/// <summary>Refreshes the component when requested and when it is still active.</summary>
/// <summary>Refreshes the component when requested and the component is undisposed.</summary>
/// <param name="shouldRefresh">A value indicating whether refresh is requested.</param>
private void Refresh(bool shouldRefresh)
{
Expand Down
12 changes: 6 additions & 6 deletions src/Primitives.Extensions.Shared/Internal/ScheduledDrainState.cs
Original file line number Diff line number Diff line change
Expand Up @@ -36,17 +36,17 @@ internal sealed class ScheduledDrainState<T>(ISequencer scheduler, IDrainTarget
/// <summary>Gets a value indicating whether the sink has reached a terminal state; read it inside the sink's gate.</summary>
internal bool Done => _done;

/// <summary>Enqueues an OnNext notification and schedules a drain pass if one isn't already running.</summary>
/// <summary>Enqueues an OnNext notification and schedules a drain pass when none is in flight.</summary>
/// <param name="value">The value to forward downstream.</param>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
internal void EnqueueNext(T value) => Enqueue(new(DrainNotificationKind.Next, value, null));

/// <summary>Enqueues an OnError notification and schedules a drain pass if one isn't already running.</summary>
/// <summary>Enqueues an OnError notification and schedules a drain pass when none is in flight.</summary>
/// <param name="error">The error to forward downstream.</param>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
internal void EnqueueError(Exception error) => Enqueue(new(DrainNotificationKind.Error, default!, error));

/// <summary>Enqueues an OnCompleted notification and schedules a drain pass if one isn't already running.</summary>
/// <summary>Enqueues an OnCompleted notification and schedules a drain pass when none is in flight.</summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
internal void EnqueueCompleted() => Enqueue(new(DrainNotificationKind.Completed, default!, null));

Expand Down Expand Up @@ -85,7 +85,7 @@ internal bool TryDequeue(out Notification notification)
}
}

/// <summary>Marks the sink done, clears queued notifications, and returns its upstream handle; the caller must hold the gate.</summary>
/// <summary>Marks the sink done and drops every queued notification, taking the gate itself.</summary>
internal void Terminate()
{
lock (_gate)
Expand All @@ -95,7 +95,7 @@ internal void Terminate()
}
}

/// <summary>Marks the sink done, clears queued notifications, and returns its upstream handle; the caller must hold the gate.</summary>
/// <summary>Marks the sink done, leaving the queue untouched; the caller must hold the gate.</summary>
internal void MarkDoneLocked() => _done = true;

/// <summary>Begins disposal under the gate, returning the upstream subscription so the caller disposes it outside the gate, or <see langword="null"/> when the sink has terminated.</summary>
Expand All @@ -119,7 +119,7 @@ internal void Terminate()
return subscription;
}

/// <summary>Enqueues a notification; claims and schedules a single drain pass if one isn't running.</summary>
/// <summary>Enqueues a notification and claims the single drain pass when none is in flight.</summary>
/// <param name="notification">The notification to forward to the drain loop.</param>
private void Enqueue(in Notification notification)
{
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ namespace ReactiveUI.Primitives.Extensions.Internal;
/// <summary>Periodic scheduling helpers for <see cref="ISequencer"/>.</summary>
internal static class SequencerPeriodicExtensions
{
/// <summary>Periodic scheduling helpers for a sequencer.</summary>
/// <summary>Repeating schedule overloads hung off a sequencer instance.</summary>
/// <param name="scheduler">The scheduler used for each tick.</param>
extension(ISequencer scheduler)
{
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ namespace ReactiveUI.Primitives.Extensions.Reactive.Operators;
namespace ReactiveUI.Primitives.Extensions.Operators;
#endif

/// <summary>Projection operator that emits <see cref="RxVoid.Default"/> for every source element without allocating a projection closure.</summary>
/// <summary>Replaces each source element with <see cref="RxVoid.Default"/>, leaving error and completion untouched.</summary>
/// <typeparam name="T">The element type of the source observable (ignored).</typeparam>
/// <param name="source">The source observable whose values are ignored.</param>
internal sealed class AsSignalObservable<T>(IObservable<T> source) : IObservable<RxVoid>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ private sealed class BufferUntilIdleSink(
/// <summary>The gate protecting state transitions and downstream notification.</summary>
private readonly Lock _gate = new();

/// <summary>Shared timer / done-flag plumbing.</summary>
/// <summary>The timer slot and terminal-state flag shared with the operator's handlers.</summary>
private readonly TimerSinkState<IList<T>> _state = new(downstream);

/// <summary>The current buffer of elements.</summary>
Expand Down Expand Up @@ -94,7 +94,7 @@ public void Dispose()
}
}

/// <summary>Schedules a flush after the idle period.</summary>
/// <summary>Replaces any pending flush with one scheduled a further idle period ahead.</summary>
private void ScheduleFlush() => _state.Timer.Disposable = scheduler.Schedule(idleTime, Flush);

/// <summary>Flushes the current buffer to the downstream observer.</summary>
Expand Down
14 changes: 9 additions & 5 deletions src/Primitives.Extensions.Shared/Operators/ConflateObservable.cs
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,12 @@ namespace ReactiveUI.Primitives.Extensions.Reactive.Operators;
namespace ReactiveUI.Primitives.Extensions.Operators;
#endif

/// <summary>Conflates an observable stream by delaying updates that occur within a minimum period.</summary>
/// <summary>
/// Delivers notifications on <paramref name="scheduler"/> and keeps emissions at least
/// <paramref name="minimumUpdatePeriod"/> apart: a value arriving inside that window is deferred to the end of it and a
/// newer value takes the deferred one's place. Completion waits for a deferred value to land, while an error is
/// forwarded at once and discards it.
/// </summary>
/// <typeparam name="T">The type of elements in the source sequence.</typeparam>
/// <param name="source">The source observable.</param>
/// <param name="minimumUpdatePeriod">The minimum period between emissions.</param>
Expand Down Expand Up @@ -48,7 +53,7 @@ internal sealed class ConflateSink : IObserver<T>, IDisposable, IDrainTarget
/// <summary>The gate protecting the queue, throttle window, and downstream notification.</summary>
private readonly Lock _gate = new();

/// <summary>Shared queue / scheduled-drain machinery.</summary>
/// <summary>The notification queue and scheduled-drain bookkeeping shared with the drain loop.</summary>
private readonly ScheduledDrainState<T> _state;

/// <summary>The disposable tracking a scheduled deferred emission.</summary>
Expand All @@ -57,7 +62,7 @@ internal sealed class ConflateSink : IObserver<T>, IDisposable, IDrainTarget
/// <summary>Wall-clock timestamp of the last emission forwarded downstream.</summary>
private DateTimeOffset _lastUpdateTime = DateTimeOffset.MinValue;

/// <summary>Set to <see langword="true"/> when an upstream OnCompleted is queued but a deferred emission is still pending; the completion fires after that emission lands.</summary>
/// <summary>Set when an upstream OnCompleted arrives while a deferred emission is pending, so the completion fires once that emission lands.</summary>
private bool _completionRequested;

/// <summary>Initializes a new instance of the <see cref="ConflateSink"/> class.</summary>
Expand Down Expand Up @@ -128,8 +133,7 @@ void IDrainTarget.Drain()

default:
{
// DrainNotificationKind has only three values; the discard arm absorbs
// Completed so the compiler sees an exhaustive switch.
// Completed is the only remaining kind; this arm keeps the switch exhaustive.
ForwardCompleted();
return;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,10 @@ namespace ReactiveUI.Primitives.Extensions.Reactive.Operators;
namespace ReactiveUI.Primitives.Extensions.Operators;
#endif

/// <summary>Debounces a sequence but emits the first value immediately.</summary>
/// <summary>
/// Emits the first source value inline, then holds each later value for <paramref name="dueTime"/> and emits only the
/// most recent one. A terminal notification flushes a pending value before it is forwarded.
/// </summary>
/// <typeparam name="T">The type of elements in the source sequence.</typeparam>
/// <param name="source">The source observable.</param>
/// <param name="dueTime">The debounce duration.</param>
Expand Down Expand Up @@ -44,7 +47,7 @@ private sealed class DebounceImmediateSink(
/// <summary>The gate for thread safety.</summary>
private readonly Lock _gate = new();

/// <summary>The timer for debouncing.</summary>
/// <summary>The pending debounce timer, replaced whenever a newer value arrives.</summary>
private readonly SwapDisposable _timer = new();

/// <summary>Whether the first value has been emitted.</summary>
Expand Down Expand Up @@ -126,7 +129,7 @@ public void Dispose()
}
}

/// <summary>Emits the last value if any.</summary>
/// <summary>Emits the waiting value, if there is one, and clears it.</summary>
private void Emit()
{
T? toEmit;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,11 @@ namespace ReactiveUI.Primitives.Extensions.Reactive.Operators;
namespace ReactiveUI.Primitives.Extensions.Operators;
#endif

/// <summary>Debounces a sequence until a condition becomes true for an element.</summary>
/// <summary>
/// Forwards a value inline when <paramref name="condition"/> holds for it, cancelling any pending emission, and
/// otherwise emits it once <paramref name="debounce"/> elapses with no newer value. An error or completion drops a
/// pending value.
/// </summary>
/// <typeparam name="T">The type of elements in the source sequence.</typeparam>
/// <param name="source">The source observable.</param>
/// <param name="debounce">The debounce duration.</param>
Expand Down Expand Up @@ -49,7 +53,7 @@ private sealed class DebounceUntilSink(
/// <summary>The gate protecting state transitions and downstream notification.</summary>
private readonly Lock _gate = new();

/// <summary>Shared timer / done-flag plumbing.</summary>
/// <summary>The timer slot and terminal-state flag shared with the operator's handlers.</summary>
private readonly TimerSinkState<T> _state = new(downstream);

/// <inheritdoc/>
Expand Down Expand Up @@ -104,7 +108,7 @@ public void Dispose()
}
}

/// <summary>Emits a debounced value when the sink is still active.</summary>
/// <summary>Emits the debounced value unless the sink has terminated.</summary>
/// <param name="value">The debounced value.</param>
private void EmitDebounced(T value)
{
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,11 @@ namespace ReactiveUI.Primitives.Extensions.Reactive.Operators;
namespace ReactiveUI.Primitives.Extensions.Operators;
#endif

/// <summary>Detects when a sequence becomes stale (no emissions for a specified period).</summary>
/// <summary>
/// Wraps each source value as an update and emits a staleness marker when <paramref name="stalenessPeriod"/> passes
/// without one, measured from subscription. The window re-arms on every value, so a single quiet stretch yields one
/// marker.
/// </summary>
/// <typeparam name="T">The type of elements in the source sequence.</typeparam>
/// <param name="source">The source observable.</param>
/// <param name="stalenessPeriod">The period after which the sequence is considered stale.</param>
Expand Down Expand Up @@ -46,7 +50,7 @@ private sealed class DetectStaleSink(
/// <summary>The gate protecting state transitions and downstream notification.</summary>
private readonly Lock _gate = new();

/// <summary>Shared timer / done-flag plumbing.</summary>
/// <summary>The timer slot and terminal-state flag shared with the operator's handlers.</summary>
private readonly TimerSinkState<Stale<T>> _state = new(downstream);

/// <summary>Upstream subscription handle, set once via <see cref="AttachSourceSubscription"/> and disposed with the sink.</summary>
Expand All @@ -68,7 +72,7 @@ public void AttachSourceSubscription(IDisposable subscription)
}
}

/// <summary>Initializes the staleness timer.</summary>
/// <summary>Arms the first staleness window, which the caller does at subscribe time.</summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void Initialize() => ScheduleStale();

Expand Down Expand Up @@ -116,7 +120,7 @@ public void Dispose()
Interlocked.Exchange(ref _sourceSubscription, null)?.Dispose();
}

/// <summary>Arms the staleness timer, which re-arms on every upstream emission without allocating a per-arm closure.</summary>
/// <summary>Arms the staleness timer, replacing any window that is counting down.</summary>
private void ScheduleStale() =>
_state.Timer.Disposable =
scheduler.Schedule(this, stalenessPeriod, static (_, self) => self.OnStaleTimer());
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,10 +10,10 @@ namespace ReactiveUI.Primitives.Extensions.Reactive.Operators;
namespace ReactiveUI.Primitives.Extensions.Operators;
#endif

/// <summary>Flattening operator that subscribes to an upstream sequence of <see cref="IEnumerable{T}"/> batches and emits each contained element individually.</summary>
/// <summary>Emits each element of every <see cref="IEnumerable{T}"/> the source produces, in order, ignoring a null batch.</summary>
/// <typeparam name="T">The element type emitted to the downstream observer.</typeparam>
/// <param name="source">The source observable of enumerables.</param>
/// <param name="scheduler">An optional scheduler used to marshal source notifications onto.</param>
/// <param name="scheduler">An optional scheduler that source notifications are marshalled onto.</param>
internal sealed class ForEachObservable<T>(
IObservable<IEnumerable<T>> source,
ISequencer? scheduler) : IObservable<T>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ namespace ReactiveUI.Primitives.Extensions.Operators;
/// <summary>Factory operator that emits each element of an <see cref="IEnumerable{T}"/> to a single subscriber and then completes, optionally marshalling the emission through a scheduler.</summary>
/// <typeparam name="T">The element type emitted to the downstream observer.</typeparam>
/// <param name="source">The source enumerable whose elements are pumped on subscription.</param>
/// <param name="scheduler">An optional scheduler used to dispatch the pump.</param>
/// <param name="scheduler">An optional scheduler that dispatches the pump.</param>
internal sealed class FromArrayObservable<T>(
IEnumerable<T> source,
ISequencer? scheduler) : IObservable<T>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,10 @@ namespace ReactiveUI.Primitives.Extensions.Reactive.Operators;
namespace ReactiveUI.Primitives.Extensions.Operators;
#endif

/// <summary>Injects heartbeat values into the sequence when the source remains quiet for a specified period.</summary>
/// <summary>
/// Wraps each source value as an update and emits a heartbeat every <paramref name="heartbeatPeriod"/> that the source
/// stays quiet, restarting the timer on each value.
/// </summary>
/// <typeparam name="T">The type of elements in the source sequence.</typeparam>
/// <param name="source">The source observable.</param>
/// <param name="heartbeatPeriod">The period between heartbeats.</param>
Expand Down Expand Up @@ -71,7 +74,7 @@ public void AttachSourceSubscription(IDisposable subscription)
}
}

/// <summary>Initializes the heartbeat timer.</summary>
/// <summary>Starts the heartbeat timer, which the caller does at subscribe time.</summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void Initialize() => ScheduleHeartbeats();

Expand Down Expand Up @@ -137,7 +140,7 @@ public void Dispose()
subscription?.Dispose();
}

/// <summary>Schedules the next heartbeat.</summary>
/// <summary>Restarts the periodic heartbeat timer, dropping the one it replaces.</summary>
private void ScheduleHeartbeats()
{
lock (_gate)
Expand Down
Loading
Loading
⌂ 换网址