From 9a5fe6bdf59d5d8feeedb2f24d9f5c9c87c9df19 Mon Sep 17 00:00:00 2001
From: Glenn Watson <5834289+glennawatson@users.noreply.github.com>
Date: Sat, 12 Sep 2026 17:55:57 +1000
Subject: [PATCH 01/11] chore: Update benchmarks and fix tests
---
src/Polyfills/ArgumentExceptionHelper.cs | 5 +-
.../ArgumentOutOfRangeExceptionHelper.cs | 4 +-
.../ObjectDisposedExceptionHelper.cs | 5 +-
src/Polyfills/TaskCompletionSource.cs | 6 +-
.../InvalidOperationExceptionHelper.cs | 33 +--
.../Internal/ScheduledDrainState.cs | 29 +-
.../Internal/SequencerPeriodicExtensions.cs | 9 +-
.../ObservableSubscriptionExtensions.cs | 14 +-
.../Operators/AsSignalObservable.cs | 8 +-
.../Operators/BufferUntilIdleObservable.cs | 5 +-
.../Operators/ConflateObservable.cs | 23 +-
.../Operators/DebounceImmediateObservable.cs | 2 +-
.../Operators/DebounceUntilObservable.cs | 5 +-
.../Operators/DetectStaleObservable.cs | 11 +-
.../Operators/ForEachObservable.cs | 7 +-
.../Operators/FromArrayObservable.cs | 7 +-
.../Operators/HeartbeatObservable.cs | 4 +-
.../Operators/ObserveOnIfObservable.cs | 2 +-
.../Operators/ObserveOnObservable.cs | 9 +-
.../Operators/RetryBackoffPolicy.cs | 5 +-
.../Operators/RetryWithDelayObservable.cs | 6 +-
.../Operators/RunAllObservable.cs | 24 +-
.../Operators/ScheduleConfig.cs | 4 +-
.../Operators/ScheduledSourceObservable.cs | 30 +--
.../Operators/ScheduledValueObservable.cs | 21 +-
.../Operators/SyncTimerObservable.cs | 37 +--
.../Operators/ThrottleDistinctObservable.cs | 16 +-
.../Operators/ThrottleObservable.cs | 9 +-
.../Operators/ThrottleUntilTrueObservable.cs | 11 +-
.../Operators/UsingActionObservable.cs | 9 +-
.../Operators/UsingFuncObservable.cs | 8 +-
.../Operators/WhileObservable.cs | 9 +-
.../ReactiveExtensions.cs | 20 +-
src/Primitives.Shared/Advanced/AfterSignal.cs | 6 +-
.../Advanced/AfterSubscription.cs | 4 +-
.../Advanced/AsyncSubscriptionLifetime.cs | 6 +-
.../Advanced/AutoShareSignal{T}.cs | 7 +-
.../Advanced/BufferSignal{T}.cs | 4 +-
.../Advanced/CatchSignal{T}.cs | 77 +++---
.../Advanced/CollectSignal{T}.cs | 4 +-
.../Advanced/CollectWitness{T}.cs | 10 +-
.../Advanced/CreateSafeSignal{T}.cs | 31 +--
.../Advanced/CreateSignal{T,TState}.cs | 62 ++---
.../Advanced/CreateSignal{T}.cs | 32 +--
.../Advanced/CreateSink{T}.cs | 9 +-
.../Advanced/DeferSignal{T}.cs | 26 +-
.../Advanced/EmitIfQuietSignal{T}.cs | 4 +-
.../Advanced/EmitIfQuietWitness{T}.cs | 10 +-
.../Advanced/EmptySignal{T}.cs | 26 +-
src/Primitives.Shared/Advanced/EverySignal.cs | 24 +-
.../Advanced/ExpireCoordinator{T}.cs | 44 ++-
.../Advanced/ExpireSignal{T}.cs | 6 +-
.../Advanced/FinallySignal{T}.cs | 60 ++---
.../Advanced/FromAsyncSubscription{T}.cs | 10 +-
...Signal{TEventHandler,TCallback,TResult}.cs | 6 +-
.../Advanced/GuardedWitness{T}.cs | 7 +-
.../Advanced/ImmutableReturnRxVoidSignal.cs | 22 +-
.../Advanced/IsEmptySignal{T}.cs | 9 +-
.../Advanced/IsEmptyWitness{T}.cs | 2 +-
.../Advanced/LoopSignal{T}.cs | 23 +-
.../MapIndexedSignal{TSource,TResult}.cs | 2 +-
.../Advanced/MergeCoordinator{T}.cs | 4 +-
.../Advanced/MergeSignal{T}.cs | 2 +-
.../Advanced/RecoverSignal{T,TException}.cs | 6 +-
.../Advanced/RepeatSourceCoordinator{T}.cs | 2 +-
.../Advanced/RepeatSourceWitness{T}.cs | 2 +-
.../Advanced/ResumeSignal{T}.cs | 5 +-
.../Advanced/ReturnSignal{T}.cs | 30 +--
.../Advanced/ScheduledEnumerableSignal{T}.cs | 6 +-
.../Advanced/SequenceSignal.cs | 4 +-
.../Advanced/SparkWitness.cs | 2 +-
src/Primitives.Shared/Advanced/StartSignal.cs | 4 +-
.../Advanced/StartSignal{T}.cs | 4 +-
.../Advanced/SubscriptionScheduling.cs | 13 +-
.../SwitchMapSignal{TSource,TResult}.cs | 4 +-
.../Advanced/SwitchWitness{T}.cs | 10 +-
.../Advanced/SynchronizeObjectSignal{T}.cs | 2 +-
.../Advanced/TaskAnyWitness{T}.cs | 4 +-
.../Advanced/TaskChainCoordinatorState.cs | 6 +-
.../Advanced/TaskChainSignal{T}.cs | 2 +-
.../Advanced/TaskCountWitness{T}.cs | 4 +-
.../Advanced/ThrowSignal{T}.cs | 32 +--
.../Advanced/TimeIntervalWitness.cs | 2 +-
src/Primitives.Shared/Advanced/TimerSlot.cs | 8 +-
.../Advanced/UnsparkWitness.cs | 2 +-
.../Advanced/WitnessLifetime.cs | 6 +-
.../Advanced/WitnessOnSignal{T}.cs | 73 +++--
.../Concurrency/SequencerWorkItem.cs | 2 +-
.../Concurrency/SequencerWorkItemDisposal.cs | 6 +-
.../ConnectableSignalRxNameExtensions.cs | 8 +-
src/Primitives.Shared/Core/Spark.cs | 6 +-
.../Core/SparkCompositeFormats.cs | 5 +-
src/Primitives.Shared/Core/Spark{T}.cs | 20 +-
src/Primitives.Shared/Handle.cs | 3 +-
src/Primitives.Shared/LinqExtensions.cs | 42 ++-
.../SignalOperatorMixins.BlendUnique.cs | 13 +-
.../SignalOperatorMixins.BufferSignals.cs | 6 +-
...SignalOperatorMixins.ChooseSwitchSelect.cs | 5 +-
.../SignalOperatorMixins.CollectionSignals.cs | 16 +-
...lOperatorMixins.CombineLatest.WideArity.cs | 14 +-
.../SignalOperatorMixins.CombineLatest.cs | 26 +-
.../SignalOperatorMixins.Coordinators.cs | 21 +-
.../SignalOperatorMixins.SchedulerSignals.cs | 10 +-
.../SignalOperatorMixins.StatefulSignals.cs | 38 +--
.../SignalOperatorMixins.Switch.cs | 2 +-
...alOperatorMixins.SyncLatest.MultiSource.cs | 4 +-
...gnalOperatorMixins.SyncLatest.WideArity.cs | 14 +-
src/Primitives.Shared/SignalOperatorMixins.cs | 13 +-
...alOperatorParityMixins.AggregateHelpers.cs | 14 +-
...OperatorParityMixins.AwaitableTerminals.cs | 8 +-
...atorParityMixins.BooleanTerminalHelpers.cs | 4 +-
.../SignalOperatorParityMixins.FlatMap.cs | 13 +-
.../SignalOperatorParityMixins.ForkJoin.cs | 2 +-
.../SignalOperatorParityMixins.Helpers.cs | 22 +-
.../SignalOperatorParityMixins.Probe.cs | 14 +-
...ixins.RxNames.CombineLatest.Collections.cs | 7 +-
...rityMixins.RxNames.CombineLatest.Tuples.cs | 2 +-
...yMixins.RxNames.CombineLatest.WideArity.cs | 20 +-
...ratorParityMixins.RxNames.CombineLatest.cs | 4 +-
...ratorParityMixins.RxNames.SubscribeSafe.cs | 20 +-
.../SignalOperatorParityMixins.RxNames.cs | 18 +-
.../SignalOperatorParityMixins.cs | 18 +-
.../Signals/PrioritySemaphoreSignal{T}.cs | 15 +-
.../Signals/ReplaySignal{T}.cs | 129 ++++-----
.../Signals/ScheduledSignal{T}.cs | 20 +-
.../Signals/SignalExtensions{Create}.cs | 6 +-
.../Signals/SignalExtensions{FromTask}.cs | 44 ++-
.../Signals/SignalExtensions{GetAwaiter}.cs | 10 +-
.../Signals/SignalExtensions{Recover}.cs | 6 +-
.../Signals/Signal{Collect}.cs | 6 +-
.../Signals/Signal{Create}.cs | 98 +++----
src/Primitives.Shared/Signals/Signal{Emit}.cs | 40 +--
.../Signals/Signal{Factories}.cs | 145 +++++-----
src/Primitives.Shared/Signals/Signal{Fail}.cs | 56 ++--
.../Signals/Signal{FromTask}.cs | 125 ++++-----
.../Signals/Signal{GetAwaiter}.cs | 38 +--
src/Primitives.Shared/Signals/Signal{None}.cs | 43 ++-
.../Signals/Signal{RxAliases}.cs | 15 +-
.../Signals/Signal{Silent}.cs | 17 +-
.../Signals/Signal{Subjects}.cs | 3 +-
src/Primitives.Shared/Signals/TaskSignal.cs | 48 ++--
.../Signals/TaskSignal{T}.cs | 56 ++--
.../Signals/TaskTerminalCompletion{T}.cs | 19 +-
.../Advanced/AsyncSerialGate.cs | 24 +-
.../Advanced/BackgroundJobSignal{T}.cs | 7 +-
.../Advanced/CallbackWitnessAsync.cs | 3 +-
.../Advanced/FlatMapCoordinator{TResult}.cs | 44 +--
.../Advanced/ForwardingWitnessAsync.cs | 5 +-
.../Advanced/IReentrantAsyncDisposable.cs | 7 +-
.../Advanced/IntervalSignal.cs | 6 +-
.../Advanced/IntervalSubscription.cs | 3 +-
.../Advanced/LeadSubscription{T}.cs | 6 +-
.../Advanced/PooledDelaySource.cs | 44 +--
.../Advanced/SingleElementWitness.cs | 12 +-
.../Advanced/StartSignal{TResult}.cs | 4 +-
.../Advanced/StartSubscription{TResult}.cs | 4 +-
...T1,T2,T3,T4,T5,T6,T7,T8,T9,T10,TResult}.cs | 5 +-
...2,T3,T4,T5,T6,T7,T8,T9,T10,T11,TResult}.cs | 5 +-
...,T4,T5,T6,T7,T8,T9,T10,T11,T12,TResult}.cs | 5 +-
...T5,T6,T7,T8,T9,T10,T11,T12,T13,TResult}.cs | 5 +-
...6,T7,T8,T9,T10,T11,T12,T13,T14,TResult}.cs | 5 +-
...,T8,T9,T10,T11,T12,T13,T14,T15,TResult}.cs | 5 +-
...T9,T10,T11,T12,T13,T14,T15,T16,TResult}.cs | 5 +-
.../SyncLatest2Coordinator{T1,T2,TResult}.cs | 5 +-
...yncLatest3Coordinator{T1,T2,T3,TResult}.cs | 5 +-
...Latest4Coordinator{T1,T2,T3,T4,TResult}.cs | 5 +-
...est5Coordinator{T1,T2,T3,T4,T5,TResult}.cs | 5 +-
...6Coordinator{T1,T2,T3,T4,T5,T6,TResult}.cs | 5 +-
...ordinator{T1,T2,T3,T4,T5,T6,T7,TResult}.cs | 5 +-
...inator{T1,T2,T3,T4,T5,T6,T7,T8,TResult}.cs | 5 +-
...tor{T1,T2,T3,T4,T5,T6,T7,T8,T9,TResult}.cs | 5 +-
.../Advanced/SyncLatestCoordinatorBase.cs | 28 +-
...tEnumerableCoordinator{TSource,TResult}.cs | 10 +-
.../Advanced/SyncLatestIndexedWitness.cs | 14 +-
.../Advanced/SyncLatestLifecycle.cs | 18 +-
.../Advanced/TakeUntilLifecycle.cs | 23 +-
.../Advanced/TakeUntilSourceWitness.cs | 6 +-
.../Advanced/TaskResultCompletionSource.cs | 2 +-
.../Advanced/TaskResultWitnessAsyncBase.cs | 4 +-
.../Advanced/TaskSignalSubscription{T}.cs | 40 ++-
.../ConcurrencyRaceHelpers.cs | 26 +-
.../ConcurrentWitnessCallsException.cs | 8 +-
.../ConnectableSignalAsync.cs | 27 +-
.../ConnectableSignalAsyncHelper.cs | 5 +-
.../ConnectableSignalAsyncState{T}.cs | 13 +-
.../Disposables/DisposableAsync.cs | 23 +-
.../Disposables/DisposableAsyncSlot.cs | 16 +-
.../Disposables/MultipleDisposableAsync.cs | 55 ++--
.../SingleAssignmentDisposableAsync.cs | 55 ++--
.../SingleReplaceableDisposableAsync.cs | 29 +-
.../Helpers/DisposalHelper.cs | 4 +-
.../Helpers/FireAndForgetHelper.cs | 9 +-
.../Helpers/SubscriptionHelper.cs | 5 +-
.../IObservableAsync.cs | 14 +-
.../IObserverAsync.cs | 25 +-
.../Mixins/DisposableAsyncExtensions.cs | 17 +-
.../Mixins/SignalExtensions.cs | 27 +-
.../Observables/Create.cs | 20 +-
.../Observables/Defer.cs | 33 +--
.../Observables/Empty.cs | 17 +-
.../Observables/FromAsync.cs | 13 +-
.../Observables/Interval.cs | 18 +-
.../Observables/Never.cs | 9 +-
.../Observables/Range.cs | 11 +-
.../Observables/Return.cs | 17 +-
.../Observables/Throw.cs | 15 +-
.../Observables/Timer.cs | 4 +-
.../Observables/ToAsyncSignal.cs | 32 +--
.../Operators/AnyAllAsync.cs | 9 +-
.../Operators/Blend.cs | 76 ++----
.../Operators/Cast.cs | 13 +-
.../Operators/Chain.cs | 17 +-
.../Operators/ChainEnumerableSignal{T}.cs | 21 +-
.../Operators/ChainSignalSourcesSignal{T}.cs | 13 +-
.../Operators/CompletionSignalDelegate.cs | 4 +-
.../Operators/ContainsAsync.cs | 3 -
.../Operators/CountAsync.cs | 3 -
.../Operators/Distinct.cs | 24 +-
.../Operators/Expire.cs | 70 +++--
.../Operators/FirstAsync.cs | 11 +-
.../Operators/FirstOrDefaultAsync.cs | 4 -
.../Operators/FlatMap.cs | 7 +-
.../Operators/Fold.cs | 18 +-
.../Operators/ForEachAsync.cs | 19 +-
.../Operators/GroupBy.cs | 19 +-
.../Operators/GroupedAsyncSignal.cs | 4 +-
.../Operators/Keep.cs | 24 +-
.../Operators/KeepType.cs | 10 +-
.../Operators/LastAsync.cs | 13 +-
.../Operators/LastOrDefaultAsync.cs | 3 -
.../Operators/Lead.cs | 8 +-
.../Operators/LongCountAsync.cs | 3 -
.../Operators/Map.cs | 25 +-
.../Operators/Multicast.cs | 61 ++---
.../Operators/OnDispose.cs | 18 +-
.../Operators/OnErrorResumeAsFailure.cs | 8 +-
.../Operators/Pair.cs | 21 +-
.../Operators/ParityHelpers.FilterFusions.cs | 42 +--
.../ParityHelpers.OperatorFusions.cs | 85 ++----
.../Operators/ParityHelpers.Partition.cs | 34 +--
.../Operators/ParityHelpers.Start.cs | 5 +-
.../Operators/ParityHelpers.cs | 11 +-
.../Operators/Reattempt.cs | 2 -
.../Operators/Recover.cs | 26 +-
.../Operators/ReduceAsync.cs | 15 +-
.../Operators/RefCount.cs | 13 +-
.../Operators/Shift.cs | 9 +-
.../SignalAsyncExtensions.Catch.Rx.cs | 5 +-
...AsyncExtensions.DistinctUntilChanged.Rx.cs | 8 +-
.../SignalAsyncExtensions.Timeout.Rx.cs | 20 +-
.../Operators/SingleAsync.cs | 22 +-
.../Operators/SingleOrDefaultAsync.cs | 21 +-
.../Operators/Skip.cs | 7 +-
.../Operators/SkipWhile.cs | 15 +-
.../Operators/StartWith.cs | 12 +-
.../Operators/SubscribeAsync.cs | 21 +-
.../Operators/SwitchTo.cs | 3 -
.../Operators/SwitchToSignal.cs | 8 +-
.../Operators/SyncLatest10.cs | 6 +-
.../Operators/SyncLatest11.cs | 6 +-
.../Operators/SyncLatest12.cs | 6 +-
.../Operators/SyncLatest13.cs | 6 +-
.../Operators/SyncLatest14.cs | 6 +-
.../Operators/SyncLatest15.cs | 6 +-
.../Operators/SyncLatest16.cs | 6 +-
.../Operators/SyncLatest8.cs | 6 +-
.../Operators/SyncLatest9.cs | 6 +-
.../Operators/SyncLatestEnumerable.cs | 16 +-
.../Operators/Take.cs | 12 +-
.../Operators/TakeUntil.PredicateSignals.cs | 6 +-
.../Operators/TakeUntil.cs | 32 +--
.../Operators/TakeUntilOptions.cs | 6 +-
.../Operators/TakeWhile.cs | 11 +-
.../Operators/Tap.cs | 3 -
.../UnhandledExceptionHandler.cs | 26 +-
.../WitnessAsync.cs | 208 ++++++---------
.../Advanced/AggregateWitness.cs | 7 +-
.../Advanced/AllPredicateWitness.cs | 2 +-
.../Advanced/AnyPredicateWitness.cs | 2 +-
.../Advanced/AnyWitness.cs | 2 +-
.../Advanced/AppendDelegateWitness.cs | 2 +-
.../Advanced/AppendWitness.cs | 4 +-
.../Advanced/AsyncEnumerableSignal{T}.cs | 24 +-
.../Advanced/BufferWitness.cs | 14 +-
.../Advanced/CallbackWitness.cs | 1 -
.../Advanced/CollectArrayWitness.cs | 2 +-
.../Advanced/CollectListWitness.cs | 2 +-
.../Advanced/ContainsWitness.cs | 2 +-
.../Advanced/DefaultIfEmptyWitness.cs | 4 +-
.../Advanced/DisposedMarker.cs | 2 +-
.../Advanced/DisposedWitness{T}.cs | 2 +-
.../Advanced/DistinctByCountAggregator.cs | 4 +-
.../Advanced/DistinctByLongCountAggregator.cs | 4 +-
.../Advanced/DistinctByWitness.cs | 4 +-
.../Advanced/DistinctWitness.cs | 6 +-
.../Advanced/EmptyWitness{T}.cs | 10 +-
.../Advanced/FoldWitness.cs | 2 +-
.../Advanced/FromEnumerableSignal{T}.cs | 28 +-
.../Advanced/IAggregator.cs | 6 +-
.../Advanced/IInlineSignal{T}.cs | 14 +-
.../Advanced/IRequireCurrentThread.cs | 10 +-
.../Advanced/IgnoreValuesWitness.cs | 2 +-
.../Advanced/ImmediateReturnSignal{T}.cs | 27 +-
.../Advanced/ImmediateThrowSignal{T}.cs | 25 +-
.../Advanced/ImmutableEmptySignal{T}.cs | 28 +-
.../Advanced/ImmutableNeverSignal{T}.cs | 17 +-
.../Advanced/ImmutableReturnFalseSignal.cs | 25 +-
.../Advanced/ImmutableReturnInt32Signal.cs | 39 ++-
.../Advanced/ImmutableReturnTrueSignal.cs | 25 +-
.../Advanced/KeepNotNullWitness.cs | 2 +-
.../Advanced/KeepTypeWitness.cs | 2 +-
.../Advanced/RangeConcatSignal.cs | 2 +-
.../Advanced/RangeSignal.cs | 28 +-
.../Advanced/RangeZipSignal{TResult}.cs | 32 +--
.../Advanced/ReduceWitness.cs | 2 +-
.../Advanced/RepeatSignal{T}.cs | 34 +--
.../Advanced/SinkSubscription.cs | 9 +-
.../Advanced/SinkTerminal.cs | 13 +-
.../Advanced/SkipWhileWitness.cs | 4 +-
.../Advanced/SkipWitness.cs | 2 +-
.../Advanced/StatefulWitness.cs | 1 -
.../Advanced/SubscriptionSlots.cs | 2 +-
.../Advanced/SynchronizeWitness.cs | 9 +-
.../Advanced/TakeWhileWitness.cs | 2 +-
.../Advanced/TakeWitness.cs | 2 +-
.../Advanced/TapWitness.cs | 2 +-
.../Advanced/UnfoldSignal{TState,TResult}.cs | 22 +-
.../Advanced/UniqueByWitness.cs | 6 +-
.../Advanced/UniqueWitness.cs | 6 +-
.../Advanced/UseSignal{TResource,T}.cs | 21 +-
.../Advanced/Witness.cs | 4 +-
.../Concurrency/ScheduledItem.cs | 54 +---
.../Concurrency/SequencerQueue.cs | 8 +-
.../ConnectableSignal{T}.cs | 6 +-
.../ExceptionExtensions.cs | 2 +-
src/ReactiveUI.Primitives.Core/Optional.cs | 28 +-
src/ReactiveUI.Primitives.Core/Result.cs | 29 +-
.../ActivationExtensionsEmitter.cs | 4 +-
.../CodeGeneration/ActivationSource.cs | 18 +-
.../CodeGeneration/GeneratedNames.cs | 12 +-
.../CodeGeneration/InstanceWrapperEmitter.cs | 4 -
.../CodeGeneration/PooledStringBuilder.cs | 17 +-
.../CodeGeneration/SourceFileWriter.cs | 5 +-
.../CodeGeneration/StaticEventsEmitter.cs | 3 +-
.../Constants.cs | 7 +-
.../EventGenerator.cs | 31 +--
.../GeneratorStepNames.cs | 6 +-
.../Concurrency/WasmScheduler.cs | 53 ++--
.../Advanced/DispatchSequencerState.cs | 25 +-
.../Concurrency/CurrentThreadSequencer.cs | 14 +-
.../Concurrency/ImmediateSequencer.cs | 2 +-
.../Concurrency/Sequencer.Simple.cs | 11 +-
.../Concurrency/SequencerExtensions.cs | 63 +----
.../SynchronizationContextSequencer.cs | 3 +-
.../Concurrency/SynchronizedSequencerQueue.cs | 5 +-
.../Concurrency/TaskPoolSequencer.cs | 5 +-
.../Concurrency/ThreadPoolSequencer.cs | 38 ++-
.../AsyncPrimitiveContractTests.cs | 21 +-
.../AsyncRenameCoverageTests.cs | 4 +-
.../CombiningOperatorTests.Blend.cs | 59 ++--
.../ConcurrentSignalBaseTests.cs | 20 +-
.../CustomTaskScheduler.cs | 38 +++
.../DisposableTests.cs | 15 +-
.../ErrorHandlingOperatorTests.cs | 10 +-
.../FactorySignalTests.cs | 47 ++--
.../NewThreadTaskScheduler.cs | 33 ---
.../ObserverAsyncDisposeTests.cs | 45 ++--
.../ParityHelpersOperatorFusionsTests.cs | 23 +-
...keUntilOperatorTests.CompletionDelegate.cs | 85 +++---
.../TimeBasedOperatorTests.cs | 22 +-
.../AvaloniaSchedulerTests.cs | 54 ++--
.../Operators/DropIfBusyObservableTests.cs | 31 +--
.../OperatorAfterTerminalGuardTests.cs | 36 +--
.../Operators/ScanWithInitialTests.cs | 64 ++---
.../SelectAsyncConcurrentObservableTests.cs | 33 ++-
.../SelectAsyncSequentialObservableTests.cs | 33 ++-
.../SelectLatestAsyncObservableTests.cs | 51 ++--
.../SubscribeAsyncObservableTests.cs | 51 ++--
.../SynchronizeAsyncObservableTests.cs | 5 -
.../ThrottleUntilTrueObservableTests.cs | 92 ++++---
.../Operators/WhileObservableTests.cs | 83 +++---
.../ReactiveExtensionsPortedTests.cs | 3 +-
.../ReactiveExtensionsTests.Misc.cs | 42 ++-
.../ReactiveExtensionsTests.Scheduling.cs | 16 +-
.../ReactiveExtensionsTests.SelectAsync.cs | 15 +-
.../ReactiveExtensionsTests.Synchronize.cs | 137 ++++------
.../ReactiveExtensionsTests.Throttle.cs | 57 ++--
.../MauiDispatcherSequencerTests.cs | 11 +-
.../WasmSchedulerTests.cs | 251 +++++++-----------
.../Common/AwaitableWitness.cs | 93 +++++++
.../Common/CancellationTokenExtensions.cs | 27 ++
.../RxNamesTests.cs | 77 +++---
.../SignalAliasCoverageTests.cs | 8 +-
.../SignalFactoriesTests.Behavior.cs | 22 +-
.../SignalFactoriesTests.Composition.cs | 6 +-
.../SignalFactoriesTests.DefaultSequencer.cs | 36 ++-
.../SignalFactoriesTests.Scheduling.cs | 6 +-
.../SignalFactoriesTests.cs | 66 +++--
.../SignalOperatorMixinsTests.cs | 21 +-
399 files changed, 3238 insertions(+), 4390 deletions(-)
create mode 100644 src/tests/ReactiveUI.Primitives.Async.Tests/CustomTaskScheduler.cs
delete mode 100644 src/tests/ReactiveUI.Primitives.Async.Tests/NewThreadTaskScheduler.cs
create mode 100644 src/tests/ReactiveUI.Primitives.Tests/Common/AwaitableWitness.cs
create mode 100644 src/tests/ReactiveUI.Primitives.Tests/Common/CancellationTokenExtensions.cs
diff --git a/src/Polyfills/ArgumentExceptionHelper.cs b/src/Polyfills/ArgumentExceptionHelper.cs
index d7df3bde..d154c660 100644
--- a/src/Polyfills/ArgumentExceptionHelper.cs
+++ b/src/Polyfills/ArgumentExceptionHelper.cs
@@ -8,9 +8,8 @@
namespace ReactiveUI.Primitives.Internal;
///
-/// Polyfill for ArgumentNullException.ThrowIfNull on target frameworks (net462-net481) that predate it.
-/// On net8.0 and later this type is not compiled; consuming projects alias the ArgumentExceptionHelper
-/// identifier directly to so the call sites bind to the BCL method.
+/// Polyfill for ArgumentNullException.ThrowIfNull on net462-net481; net8.0 and later compile no copy of this
+/// type and alias the ArgumentExceptionHelper identifier to instead.
///
[ExcludeFromCodeCoverage]
internal static class ArgumentExceptionHelper
diff --git a/src/Polyfills/ArgumentOutOfRangeExceptionHelper.cs b/src/Polyfills/ArgumentOutOfRangeExceptionHelper.cs
index 65a5d9c1..825fc1e9 100644
--- a/src/Polyfills/ArgumentOutOfRangeExceptionHelper.cs
+++ b/src/Polyfills/ArgumentOutOfRangeExceptionHelper.cs
@@ -70,9 +70,7 @@ internal static void ThrowIfLessThan(
"Design",
"SST2318:Members should not have identical bodies",
Justification =
- "A type-specialized polyfill overload. The bounds-check body is identical to the int overload only "
- + "because the guard shape is the same; the two operate on different value types (int vs TimeSpan) and "
- + "cannot forward to one another. This mirrors the BCL's per-type ThrowIfLessThan overloads.")]
+ "The int and TimeSpan overloads guard different value types and cannot forward to one another.")]
internal static void ThrowIfLessThan(
TimeSpan value,
TimeSpan other,
diff --git a/src/Polyfills/ObjectDisposedExceptionHelper.cs b/src/Polyfills/ObjectDisposedExceptionHelper.cs
index 8867323b..c74b6361 100644
--- a/src/Polyfills/ObjectDisposedExceptionHelper.cs
+++ b/src/Polyfills/ObjectDisposedExceptionHelper.cs
@@ -7,9 +7,8 @@
namespace ReactiveUI.Primitives.Internal;
///
-/// Polyfill for ObjectDisposedException.ThrowIf on target frameworks (net462-net481) that predate it.
-/// On net8.0 and later this type is not compiled; consuming projects alias the ObjectDisposedExceptionHelper
-/// identifier directly to so the call sites bind to the BCL method.
+/// Polyfill for ObjectDisposedException.ThrowIf on net462-net481; net8.0 and later compile no copy of this
+/// type and alias the ObjectDisposedExceptionHelper identifier to instead.
///
[ExcludeFromCodeCoverage]
internal static class ObjectDisposedExceptionHelper
diff --git a/src/Polyfills/TaskCompletionSource.cs b/src/Polyfills/TaskCompletionSource.cs
index b5b10cc2..615dcae8 100644
--- a/src/Polyfills/TaskCompletionSource.cs
+++ b/src/Polyfills/TaskCompletionSource.cs
@@ -19,7 +19,7 @@ internal sealed class TaskCompletionSource
/// Initializes a new instance of the class.
[SuppressMessage("Concurrency", "PSH1302", Justification =
- "BCL-parity polyfill; must match the framework ctor's TaskCreationOptions.None default, not force async continuations.")]
+ "Matches the framework constructor's TaskCreationOptions.None default.")]
public TaskCompletionSource() => _inner = new();
/// Transitions the underlying task to the state.
@@ -47,9 +47,7 @@ internal sealed class TaskCompletionSource
"Design",
"SST2318:Members should not have identical bodies",
Justification =
- "Distinct BCL-parity surface members with different contracts and return types: SetCanceled is void and "
- + "TrySetCanceled returns bool. They share the same backing call here but must track the framework surface "
- + "independently, so they are deliberately kept as separate members rather than one forwarding to the other.")]
+ "SetCanceled returns void and TrySetCanceled returns bool; each tracks its own framework contract.")]
internal bool TrySetCanceled() => _inner.TrySetCanceled();
/// Attempts to transition the underlying task to the state for the specified token.
diff --git a/src/Primitives.Extensions.Shared/Internal/InvalidOperationExceptionHelper.cs b/src/Primitives.Extensions.Shared/Internal/InvalidOperationExceptionHelper.cs
index 235386f3..11581b65 100644
--- a/src/Primitives.Extensions.Shared/Internal/InvalidOperationExceptionHelper.cs
+++ b/src/Primitives.Extensions.Shared/Internal/InvalidOperationExceptionHelper.cs
@@ -8,23 +8,16 @@
namespace ReactiveUI.Primitives.Extensions;
///
-/// Provides helper methods for throwing when
-/// constructor-supplied state on an operator is missing at the time it is consumed.
-/// The thrown message is composed of the captured member name and the caller member
-/// (typically Subscribe or the enclosing type), so call sites just pass the
-/// field being validated.
+/// Throws when constructor-supplied operator state is missing at the
+/// point it is consumed, naming the captured member and the calling member in the message.
///
[ExcludeFromCodeCoverage]
internal static class InvalidOperationExceptionHelper
{
- ///
- /// Throws an if is null.
- /// The exception message is composed from the captured argument expression and the
- /// caller member, e.g. "'source' was not supplied to 'Subscribe'.".
- ///
+ /// Throws when is null.
/// The reference type field to validate as non-null.
- /// The validated member's name, captured from the expression via .
- /// The void-throwing caller's name, captured via .
+ /// The validated member's name, captured from the expression.
+ /// The calling member's name.
/// is .
internal static void ThrowIfNull(
[NotNull] object? argument,
@@ -42,13 +35,13 @@ internal static void ThrowIfNull(
}
///
- /// Validates an argument and returns it if it is not null, otherwise throws an .
- /// Designed for use in primary constructor initializers.
+ /// Returns when non-null and otherwise throws, shaped for use in a primary
+ /// constructor initializer.
///
/// The type of the argument.
/// The argument to validate.
- /// The validated reference-type argument's name, captured from the expression via .
- /// The reference-type-checking caller's name, captured via .
+ /// The validated argument's name, captured from the expression.
+ /// The calling member's name.
/// The non-null argument.
/// is .
internal static T Check(
@@ -68,12 +61,12 @@ internal static T Check(
}
///
- /// Validates a string argument and returns it if it is not null or empty, otherwise throws an .
- /// Designed for use in primary constructor initializers.
+ /// Returns when it is neither null nor empty and otherwise throws, shaped for use
+ /// in a primary constructor initializer.
///
/// The argument to validate.
- /// The validated string argument's name, captured from the expression via .
- /// The string-checking caller's name, captured via .
+ /// The validated argument's name, captured from the expression.
+ /// The calling member's name.
/// The non-null, non-empty argument.
/// is or empty.
internal static string Check(
diff --git a/src/Primitives.Extensions.Shared/Internal/ScheduledDrainState.cs b/src/Primitives.Extensions.Shared/Internal/ScheduledDrainState.cs
index 89c6ef95..7271810a 100644
--- a/src/Primitives.Extensions.Shared/Internal/ScheduledDrainState.cs
+++ b/src/Primitives.Extensions.Shared/Internal/ScheduledDrainState.cs
@@ -12,14 +12,10 @@ namespace ReactiveUI.Primitives.Extensions.Internal;
#endif
///
-/// Shared queue-and-single-drain marshaller used by the synchronous scheduler-marshalling operator
-/// sinks (ObserveOn, Conflate). Each of those sinks previously hand-rolled the same gate,
-/// FIFO queue, drain-in-flight flag, terminal flag, enqueue-and-schedule logic, and dequeue loop on top
-/// of identical fields; this helper centralises that machinery so the per-sink class only carries the
-/// operator-specific notification handling. Notifications are enqueued and a single drain pass is
-/// scheduled per burst (rather than one scheduled action per item), and the drain callback carries no
-/// captures — the sink is passed through as an . Sinks compose one instance
-/// and forward to it; there is no base class and no per-item virtual dispatch.
+/// Queue-and-single-drain marshaller composed by the synchronous scheduler-marshalling operator sinks
+/// (ObserveOn, Conflate): notifications are enqueued under the sink's gate and one drain pass is
+/// scheduled per burst rather than one scheduled action per item. The drain callback is capture-free, taking
+/// the sink through as an .
///
/// The element type carried by notifications.
/// The scheduler each drain pass runs on.
@@ -42,8 +38,7 @@ internal sealed class ScheduledDrainState(ISequencer scheduler, IDrainTarget
/// Set to once a terminal notification has been delivered or the sink disposed.
private bool _done;
- /// Gets a value indicating whether the sink has reached a terminal state. Read inside
- /// the sink's gate by callers that need to short-circuit once terminated.
+ /// Gets a value indicating whether the sink has reached a terminal state; read it inside the sink's gate.
internal bool Done => _done;
/// Enqueues an OnNext notification and schedules a drain pass if one isn't already running.
@@ -60,7 +55,7 @@ internal sealed class ScheduledDrainState(ISequencer scheduler, IDrainTarget
[MethodImpl(MethodImplOptions.AggressiveInlining)]
internal void EnqueueCompleted() => Enqueue(new(DrainNotificationKind.Completed, default!, null));
- /// Records the upstream subscription, or disposes it immediately if the sink is already done.
+ /// Records the upstream subscription, or disposes it immediately when the sink has terminated.
/// The upstream subscription handle.
internal void Attach(IDisposable subscription)
{
@@ -109,9 +104,8 @@ internal void Terminate()
/// the still-queued notifications are abandoned because checks the done flag first.
internal void MarkDoneLocked() => _done = true;
- /// Begins disposal under the gate: marks the sink done, clears the queue, and detaches
- /// the upstream subscription — returned to the caller so it is disposed outside the gate. Returns
- /// when already disposed.
+ /// Begins disposal under the gate, returning the upstream subscription so the caller disposes it
+ /// outside the gate, or when the sink has terminated.
/// The upstream subscription to dispose outside the gate, or .
internal IDisposable? BeginDispose()
{
@@ -121,10 +115,9 @@ internal void Terminate()
}
}
- /// Marks the sink done, clears the queue, and detaches the upstream subscription, returning it for
- /// disposal outside the gate. Caller must hold the gate and have confirmed is
- /// . Lets a composing sink dispose its own scheduled-work slot atomically with the
- /// done transition under the same lock.
+ /// Marks the sink done, clears the queue, and returns the upstream subscription for disposal outside
+ /// the gate. Caller must hold the gate and have confirmed is , which
+ /// lets it retire its own scheduled-work slot atomically with the done transition.
/// The upstream subscription to dispose outside the gate, or .
internal IDisposable? BeginDisposeLocked()
{
diff --git a/src/Primitives.Extensions.Shared/Internal/SequencerPeriodicExtensions.cs b/src/Primitives.Extensions.Shared/Internal/SequencerPeriodicExtensions.cs
index 6af78f2c..e355ba5c 100644
--- a/src/Primitives.Extensions.Shared/Internal/SequencerPeriodicExtensions.cs
+++ b/src/Primitives.Extensions.Shared/Internal/SequencerPeriodicExtensions.cs
@@ -11,7 +11,7 @@ namespace ReactiveUI.Primitives.Extensions.Reactive.Internal;
namespace ReactiveUI.Primitives.Extensions.Internal;
#endif
-/// Periodic scheduling helpers used by migrated extension operators.
+/// Periodic scheduling helpers for .
internal static class SequencerPeriodicExtensions
{
/// Periodic scheduling helpers for a sequencer.
@@ -67,10 +67,7 @@ private static PeriodicSubscription SchedulePeriodic(
return subscription;
}
- ///
- /// Disposable state for one periodic schedule. Internal (rather than private) so coverage tests can
- /// drive directly instead of via reflection.
- ///
+ /// Disposable state for one periodic schedule, re-arming itself after each tick until disposed.
/// The state type.
/// The scheduler used for each tick.
/// The state passed to each tick.
@@ -127,7 +124,7 @@ internal void ScheduleNext(TimeSpan dueTime)
});
}
- /// Runs a tick and schedules the next one when still active.
+ /// Runs one tick and schedules the next, unless disposed.
internal void Tick()
{
if (Volatile.Read(ref _disposed) != 0)
diff --git a/src/Primitives.Extensions.Shared/ObservableSubscriptionExtensions.cs b/src/Primitives.Extensions.Shared/ObservableSubscriptionExtensions.cs
index 1165906c..48ea0223 100644
--- a/src/Primitives.Extensions.Shared/ObservableSubscriptionExtensions.cs
+++ b/src/Primitives.Extensions.Shared/ObservableSubscriptionExtensions.cs
@@ -12,9 +12,8 @@ namespace ReactiveUI.Primitives.Extensions;
#endif
///
-/// Provides extension methods for subscribing to and handling reactive sequences
-/// in a synchronous or blocking manner. These methods offer utility functions
-/// to retrieve emitted values, handle completion, and capture errors from observables.
+/// Provides extension methods that subscribe to a sequence and block the calling thread until it produces a
+/// value or terminates, returning the emitted value or the captured error.
///
public static class ObservableSubscriptionExtensions
{
@@ -240,8 +239,8 @@ private static void WaitForCompletionCore(IObservable source, ISequencer
}
///
- /// Subscribes to the specified observable using the provided .
- /// If a scheduler is specified, the subscription is scheduled; otherwise, the subscription occurs immediately.
+ /// Subscribes to on , or
+ /// inline when no scheduler is supplied.
///
/// The type of the elements in .
/// The observable to subscribe to.
@@ -360,9 +359,8 @@ private sealed class BlockingValueWitness(ManualResetEventSlim done) : IObser
"Design",
"SST2318:Members should not have identical bodies",
Justification =
- "This blocking witness treats completion and error identically: either terminal signal releases the "
- + "gate. OnError and OnCompleted are distinct IObserver channels that share this by design, not a "
- + "copy that was meant to differ.")]
+ "Completion and error are distinct IObserver channels that deliberately share one action: "
+ + "releasing the gate.")]
public void OnCompleted() => done.Set();
}
diff --git a/src/Primitives.Extensions.Shared/Operators/AsSignalObservable.cs b/src/Primitives.Extensions.Shared/Operators/AsSignalObservable.cs
index a9826886..25ad98c5 100644
--- a/src/Primitives.Extensions.Shared/Operators/AsSignalObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/AsSignalObservable.cs
@@ -11,10 +11,8 @@ namespace ReactiveUI.Primitives.Extensions.Operators;
#endif
///
-/// Projection operator that emits for every source
-/// element. Replaces the source.Select(_ => RxVoid.Default) pattern,
-/// avoiding the per-subscription closure allocation that the projection lambda
-/// would otherwise capture.
+/// Projection operator that emits for every source element without allocating a
+/// projection closure.
///
/// The element type of the source observable (ignored).
/// The source observable whose values are ignored.
@@ -28,7 +26,7 @@ public IDisposable Subscribe(IObserver observer)
return source.Subscribe(new AsSignalWitness(observer));
}
- /// Forwarding observer that replaces every value with . Error and completion signals pass through unchanged.
+ /// Forwards every source value as , passing error and completion through unchanged.
/// The downstream observer.
private sealed class AsSignalWitness(IObserver downstream) : IObserver
{
diff --git a/src/Primitives.Extensions.Shared/Operators/BufferUntilIdleObservable.cs b/src/Primitives.Extensions.Shared/Operators/BufferUntilIdleObservable.cs
index 524306f5..42a334d3 100644
--- a/src/Primitives.Extensions.Shared/Operators/BufferUntilIdleObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/BufferUntilIdleObservable.cs
@@ -35,10 +35,7 @@ public IDisposable Subscribe(IObserver> observer)
return new DisposableBag(subscription, sink);
}
- ///
- /// Sink that manages the buffer and idle timer. Composes for
- /// the shared gate / timer / done-flag plumbing so this class only carries the buffer logic.
- ///
+ /// Sink that accumulates elements and emits the buffer once the idle timer elapses.
/// The downstream observer.
/// The idle time period.
/// The scheduler.
diff --git a/src/Primitives.Extensions.Shared/Operators/ConflateObservable.cs b/src/Primitives.Extensions.Shared/Operators/ConflateObservable.cs
index 0421684d..3960cbf3 100644
--- a/src/Primitives.Extensions.Shared/Operators/ConflateObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/ConflateObservable.cs
@@ -34,11 +34,9 @@ public IDisposable Subscribe(IObserver observer)
}
///
- /// Single observer that combines two previously-distinct concerns into one allocation:
- /// (1) marshals upstream notifications onto the scheduler thread — delegated to the shared
- /// FIFO queue and scheduled drain — and (2) applies the conflate
- /// time-window throttle to each notification. End-user-observable
- /// semantics are unchanged from the prior two-observer implementation.
+ /// Single observer that both marshals upstream notifications onto the scheduler thread, through the
+ /// queue and scheduled drain, and applies the conflate time-window
+ /// throttle to each notification.
///
internal sealed class ConflateSink : IObserver, IDisposable, IDrainTarget
{
@@ -149,10 +147,8 @@ void IDrainTarget.Drain()
[MethodImpl(MethodImplOptions.AggressiveInlining)]
internal void AttachSourceSubscription(IDisposable subscription) => _state.Attach(subscription);
- /// Applies the throttle-window decision to a dequeued value and either emits inline or
- /// schedules a deferred emission. The emission bodies live in covered helpers; only this
- /// race-guarded shell (whose already-done early-out is reachable only when a concurrent dispose
- /// flips the flag between the drain dequeue and this gate acquisition) is excluded.
+ /// Applies the throttle-window decision to a dequeued value, emitting it inline or scheduling a
+ /// deferred emission at the end of the window.
/// The value to forward.
[System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
private void ProcessNext(T value)
@@ -162,6 +158,7 @@ private void ProcessNext(T value)
lock (_gate)
{
+ // Race-only: a concurrent dispose can flip the done flag between the drain dequeue and this gate.
if (_state.Done)
{
return;
@@ -228,13 +225,12 @@ private void EmitInline(T value)
/// Forwards an error to downstream and terminates the sink.
/// The error to forward.
- /// The already-terminated early-out is reachable only when a concurrent dispose flips the
- /// flag between the drain dequeue and this gate acquisition; excluded as race-only.
[System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
private void ForwardError(Exception error)
{
lock (_gate)
{
+ // Race-only: a concurrent dispose can flip the done flag between the drain dequeue and this gate.
if (_state.Done)
{
return;
@@ -247,14 +243,13 @@ private void ForwardError(Exception error)
_downstream.OnError(error);
}
- /// Forwards completion, deferring if a throttled emission is still scheduled.
- /// The already-terminated early-out is reachable only when a concurrent dispose flips the
- /// flag between the drain dequeue and this gate acquisition; excluded as race-only.
+ /// Forwards completion, deferring it when a throttled emission is scheduled.
[System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
private void ForwardCompleted()
{
lock (_gate)
{
+ // Race-only: a concurrent dispose can flip the done flag between the drain dequeue and this gate.
if (_state.Done)
{
return;
diff --git a/src/Primitives.Extensions.Shared/Operators/DebounceImmediateObservable.cs b/src/Primitives.Extensions.Shared/Operators/DebounceImmediateObservable.cs
index cbd0270a..63703687 100644
--- a/src/Primitives.Extensions.Shared/Operators/DebounceImmediateObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/DebounceImmediateObservable.cs
@@ -32,7 +32,7 @@ public IDisposable Subscribe(IObserver observer)
return new DisposableBag(subscription, sink);
}
- /// Sink for the debounce immediate observable.
+ /// Sink that forwards the first value inline and debounces every later value by the due time.
/// The downstream observer.
/// The debounce duration.
/// The scheduler to use for timing.
diff --git a/src/Primitives.Extensions.Shared/Operators/DebounceUntilObservable.cs b/src/Primitives.Extensions.Shared/Operators/DebounceUntilObservable.cs
index 266b6170..e92e72d6 100644
--- a/src/Primitives.Extensions.Shared/Operators/DebounceUntilObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/DebounceUntilObservable.cs
@@ -35,10 +35,7 @@ public IDisposable Subscribe(IObserver observer)
return new DisposableBag(subscription, sink);
}
- ///
- /// Sink for the debounce-until observable. Composes for the
- /// shared gate / timer / done-flag plumbing so this class only carries the OnNext logic.
- ///
+ /// Sink that forwards a value inline when the condition holds and otherwise after the debounce window.
/// The downstream observer.
/// The debounce duration.
/// The condition.
diff --git a/src/Primitives.Extensions.Shared/Operators/DetectStaleObservable.cs b/src/Primitives.Extensions.Shared/Operators/DetectStaleObservable.cs
index 60a46e1b..bb69cb0d 100644
--- a/src/Primitives.Extensions.Shared/Operators/DetectStaleObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/DetectStaleObservable.cs
@@ -34,10 +34,7 @@ public IDisposable Subscribe(IObserver> observer)
return sink;
}
- ///
- /// Sink that manages staleness detection. Composes for the
- /// shared gate / timer / done-flag plumbing so this class only carries the OnNext / schedule logic.
- ///
+ /// Sink that re-arms the staleness timer on each upstream value and emits a stale marker when the window elapses.
/// The downstream observer.
/// The staleness period.
/// The scheduler.
@@ -52,7 +49,7 @@ private sealed class DetectStaleSink(
/// Shared timer / done-flag plumbing.
private readonly TimerSinkState> _state = new(downstream);
- /// Upstream subscription handle, set once via so the sink can tear it down on dispose without a wrapper bag.
+ /// Upstream subscription handle, set once via and disposed with the sink.
private IDisposable? _sourceSubscription;
/// Records the upstream subscription for disposal.
@@ -119,9 +116,7 @@ public void Dispose()
Interlocked.Exchange(ref _sourceSubscription, null)?.Dispose();
}
- /// Schedules the staleness notification. Uses the state-carrying scheduler
- /// overload with a static lambda so no per-reschedule closure capturing this is
- /// allocated (the timer re-arms on every upstream emission).
+ /// Arms the staleness timer, which re-arms on every upstream emission without allocating a per-arm closure.
private void ScheduleStale() =>
_state.Timer.Disposable =
scheduler.Schedule(this, stalenessPeriod, static (_, self) => self.OnStaleTimer());
diff --git a/src/Primitives.Extensions.Shared/Operators/ForEachObservable.cs b/src/Primitives.Extensions.Shared/Operators/ForEachObservable.cs
index 3c8570bd..529016f3 100644
--- a/src/Primitives.Extensions.Shared/Operators/ForEachObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/ForEachObservable.cs
@@ -11,11 +11,8 @@ namespace ReactiveUI.Primitives.Extensions.Operators;
#endif
///
-/// Flattening operator that subscribes to an upstream sequence of
-/// batches and emits each contained element
-/// individually. Replaces the
-/// Observable.Create<T>(o => source.ObserveOnSafe(s).Subscribe(v => o.FastForEach(v)))
-/// pattern with a single operator + observer pair.
+/// Flattening operator that subscribes to an upstream sequence of batches and emits
+/// each contained element individually.
///
/// The element type emitted to the downstream observer.
/// The source observable of enumerables.
diff --git a/src/Primitives.Extensions.Shared/Operators/FromArrayObservable.cs b/src/Primitives.Extensions.Shared/Operators/FromArrayObservable.cs
index e37f7674..b8a06fde 100644
--- a/src/Primitives.Extensions.Shared/Operators/FromArrayObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/FromArrayObservable.cs
@@ -11,11 +11,8 @@ namespace ReactiveUI.Primitives.Extensions.Operators;
#endif
///
-/// Factory operator that emits each element of an
-/// to a single subscriber, then completes. The emission can optionally be
-/// marshalled through a scheduler. Replaces the
-/// Observable.Create<T>(o => scheduler.ScheduleSafe(() => o.FastForEach(source)))
-/// pattern with a dedicated implementation.
+/// Factory operator that emits each element of an to a single subscriber and then
+/// completes, optionally marshalling the emission through a scheduler.
///
/// The element type emitted to the downstream observer.
/// The source enumerable whose elements are pumped on subscription.
diff --git a/src/Primitives.Extensions.Shared/Operators/HeartbeatObservable.cs b/src/Primitives.Extensions.Shared/Operators/HeartbeatObservable.cs
index e9027274..3c73db56 100644
--- a/src/Primitives.Extensions.Shared/Operators/HeartbeatObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/HeartbeatObservable.cs
@@ -34,7 +34,7 @@ public IDisposable Subscribe(IObserver> observer)
return sink;
}
- /// The sink for the heartbeat operator.
+ /// Sink that forwards upstream values and emits a heartbeat whenever the period elapses without one.
/// The downstream observer.
/// The period between heartbeats.
/// The scheduler to run the heartbeat timer on.
@@ -49,7 +49,7 @@ private sealed class HeartbeatSink(
/// The subscription to the periodic heartbeat timer.
private readonly MutableDisposable _timerSubscription = new();
- /// Upstream subscription handle; set once via so the sink can tear it down in without needing a wrapper bag.
+ /// Upstream subscription handle, set once via and torn down in .
private IDisposable? _sourceSubscription;
/// Whether the sink has completed or been disposed.
diff --git a/src/Primitives.Extensions.Shared/Operators/ObserveOnIfObservable.cs b/src/Primitives.Extensions.Shared/Operators/ObserveOnIfObservable.cs
index 270dcb71..574dc407 100644
--- a/src/Primitives.Extensions.Shared/Operators/ObserveOnIfObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/ObserveOnIfObservable.cs
@@ -46,7 +46,7 @@ public IDisposable Subscribe(IObserver observer)
return new DisposableBag(sourceSub, conditionSub, sink);
}
- /// Sinks the source observable and conditionally observes on different schedulers.
+ /// Sink that forwards each value on whichever scheduler the latest condition value selects.
/// The downstream observer.
/// The scheduler to use when condition is true.
/// The scheduler to use when condition is false.
diff --git a/src/Primitives.Extensions.Shared/Operators/ObserveOnObservable.cs b/src/Primitives.Extensions.Shared/Operators/ObserveOnObservable.cs
index 610a58c3..f17f3740 100644
--- a/src/Primitives.Extensions.Shared/Operators/ObserveOnObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/ObserveOnObservable.cs
@@ -11,12 +11,9 @@ namespace ReactiveUI.Primitives.Extensions.Operators;
#endif
///
-/// Marshals every source notification onto the supplied , preserving order.
-/// Replaces the System.Reactive.Linq.Observable.ObserveOn delegation behind the sync
-/// ObserveOnSafe / ObserveOnIf helpers with our own queue-and-single-drain marshaller:
-/// notifications are enqueued and a single drain pass is scheduled per burst (rather than one
-/// scheduled action per item). The shared queue / gate / drain machinery lives in
-/// ; this sink only carries the forward-everything drain handling.
+/// Marshals every source notification onto the supplied , preserving order: notifications
+/// are enqueued in and one drain pass is scheduled per burst rather than one
+/// scheduled action per item.
///
/// The element type of the source sequence.
/// The source observable.
diff --git a/src/Primitives.Extensions.Shared/Operators/RetryBackoffPolicy.cs b/src/Primitives.Extensions.Shared/Operators/RetryBackoffPolicy.cs
index 9171802f..2e67051b 100644
--- a/src/Primitives.Extensions.Shared/Operators/RetryBackoffPolicy.cs
+++ b/src/Primitives.Extensions.Shared/Operators/RetryBackoffPolicy.cs
@@ -9,9 +9,8 @@ namespace ReactiveUI.Primitives.Extensions.Operators;
#endif
///
-/// Bundled retry configuration for : retry count,
-/// delay schedule, scheduler, and an optional error sink. A
-/// so it stays allocation-free and keeps the constructors that consume it to a short parameter list.
+/// Bundled retry configuration for : retry count, delay schedule,
+/// scheduler, and an optional error sink.
///
/// Maximum number of retries.
/// Delay before the first retry.
diff --git a/src/Primitives.Extensions.Shared/Operators/RetryWithDelayObservable.cs b/src/Primitives.Extensions.Shared/Operators/RetryWithDelayObservable.cs
index 9592b8fb..30f2c0ff 100644
--- a/src/Primitives.Extensions.Shared/Operators/RetryWithDelayObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/RetryWithDelayObservable.cs
@@ -16,10 +16,12 @@ namespace ReactiveUI.Primitives.Extensions.Operators;
/// The source observable.
/// The maximum number of retries.
/// A function to select the delay for each retry attempt.
+/// The sequencer timing retry delays; null uses the default sequencer.
internal sealed class RetryWithDelayObservable(
IObservable source,
int retryCount,
- Func delaySelector) : IObservable
+ Func delaySelector,
+ ISequencer? sequencer = null) : IObservable
{
///
public IDisposable Subscribe(IObserver observer)
@@ -28,7 +30,7 @@ public IDisposable Subscribe(IObserver observer)
InvalidOperationExceptionHelper.ThrowIfNull(delaySelector);
ArgumentExceptionHelper.ThrowIfNull(observer);
- RetryWithDelaySink sink = new(observer, source, retryCount, delaySelector, Sequencer.Default);
+ RetryWithDelaySink sink = new(observer, source, retryCount, delaySelector, sequencer ?? Sequencer.Default);
sink.Run();
return sink;
}
diff --git a/src/Primitives.Extensions.Shared/Operators/RunAllObservable.cs b/src/Primitives.Extensions.Shared/Operators/RunAllObservable.cs
index e687c2a7..05d4b096 100644
--- a/src/Primitives.Extensions.Shared/Operators/RunAllObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/RunAllObservable.cs
@@ -17,12 +17,6 @@ namespace ReactiveUI.Primitives.Extensions.Operators;
/// Errors from any observable propagate to the downstream observer.
///
/// The list of one-shot observables to run in order.
-///
-/// Replaces patterns like sources.Concat().LastOrDefaultAsync() with a single
-/// operator that subscribes sequentially. Uses an iterative loop with a sync-completion
-/// flag to avoid stack overflow when sources complete synchronously during
-/// Subscribe.
-///
internal sealed class RunAllObservable(IReadOnlyList> sources) : IObservable
{
///
@@ -43,10 +37,9 @@ public IDisposable Subscribe(IObserver observer)
}
///
- /// Stateful observer that walks the source list sequentially. The sink subscribes itself
- /// directly to each source — its own sets a
- /// per-iteration flag the surrounding loop reads to decide whether to advance. This
- /// replaces the previous probe-observer-per-iteration allocation pattern.
+ /// Stateful observer that walks the source list sequentially: the sink subscribes itself to each source, and its
+ /// own sets a per-iteration flag the surrounding loop reads to
+ /// decide whether to advance.
///
/// The downstream observer.
/// The source list to walk.
@@ -66,10 +59,9 @@ private sealed class Sink(
/// Guards against re-entrant calls.
private bool _looping;
- /// Per-iteration latch (0 = pending, 1 = terminated). Set by
- /// when a source terminates synchronously during Subscribe; read by the surrounding
- /// loop in . Accessed via so it crosses the
- /// method boundary safely without needing a separate probe-observer allocation per iteration.
+ /// Per-iteration latch (0 = pending, 1 = terminated), set by when a source
+ /// terminates synchronously during Subscribe and read by the surrounding loop in
+ /// .
private int _iterationTerminated;
///
@@ -146,12 +138,10 @@ internal void RunNext()
}
/// Emits the terminal and completes once all sources have run.
- /// The already-done early-out is only reachable when a concurrent dispose latches between the
- /// loop exit and this call; this small completion shell is excluded from coverage as race-only while the
- /// trampoline loop in stays covered.
[System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
private void CompleteRun()
{
+ // Race-only: a concurrent dispose can latch the done flag between the loop exit and this call.
if (Interlocked.Exchange(ref _done, 1) != 0)
{
return;
diff --git a/src/Primitives.Extensions.Shared/Operators/ScheduleConfig.cs b/src/Primitives.Extensions.Shared/Operators/ScheduleConfig.cs
index 28b7181a..fa93ee26 100644
--- a/src/Primitives.Extensions.Shared/Operators/ScheduleConfig.cs
+++ b/src/Primitives.Extensions.Shared/Operators/ScheduleConfig.cs
@@ -10,9 +10,7 @@ namespace ReactiveUI.Primitives.Extensions.Operators;
///
/// Bundled scheduling configuration shared by and
-/// . Carrying these parameters as a single readonly record
-/// struct keeps observable/observer constructor parameter counts low, avoids SA1117-style parameter
-/// soup, and lets the compiler copy the configuration into capture-free scheduler lambdas.
+/// , copied by value into capture-free scheduler lambdas.
///
/// The element type emitted by the configured observable.
/// The scheduler on which each emission is dispatched.
diff --git a/src/Primitives.Extensions.Shared/Operators/ScheduledSourceObservable.cs b/src/Primitives.Extensions.Shared/Operators/ScheduledSourceObservable.cs
index 7de6946b..8d8832d7 100644
--- a/src/Primitives.Extensions.Shared/Operators/ScheduledSourceObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/ScheduledSourceObservable.cs
@@ -11,21 +11,14 @@ namespace ReactiveUI.Primitives.Extensions.Operators;
#endif
///
-/// Source-driven scheduled observable. Subscribes to an upstream
-/// and, for every emitted value, schedules a
-/// callback on the supplied that applies an optional
-/// side-effect and/or transform
-/// before forwarding the value to the downstream observer. Replaces the
-/// Observable.Create<T>(o => source.Subscribe(v => scheduler.Schedule(...)))
-/// family of source-driven Schedule overloads.
+/// Source-driven scheduled observable: for every upstream value, schedules a callback on the supplied
+/// that applies an optional side-effect and/or
+/// transform before forwarding the value downstream.
///
/// The element type of the source observable.
///
-/// To match the original source.Subscribe(Action<T>) semantics, this
-/// operator only forwards . Source errors and
-/// completion are intentionally not propagated to the downstream observer; that
-/// preserves the historical behaviour of Observable.Create + a
-/// next-only subscription.
+/// Only is forwarded: source errors and completion never reach the downstream
+/// observer, so a subscriber sees no terminal notification.
///
internal sealed class ScheduledSourceObservable : IObservable
{
@@ -56,11 +49,8 @@ public IDisposable Subscribe(IObserver observer)
}
///
- /// Carries the per-emission state into the scheduled callback so the
- /// scheduler lambda does not capture any fields. A
- /// so it rides inside the
- /// scheduler's work item by value rather than as a separate per-emission heap
- /// allocation.
+ /// Carries the per-emission state by value into the scheduled callback so the scheduler lambda captures
+ /// nothing.
///
/// The downstream observer.
/// The value to emit.
@@ -129,15 +119,13 @@ public void OnNext(T value)
///
public void OnError(Exception error)
{
- // Intentionally not forwarded: original Observable.Create + Subscribe(Action)
- // pattern silently dropped source errors. Preserving that behaviour.
+ // Not forwarded: this operator is next-only by contract.
}
///
public void OnCompleted()
{
- // Intentionally not forwarded: original Observable.Create + Subscribe(Action)
- // pattern silently dropped completion. Preserving that behaviour.
+ // Not forwarded: this operator is next-only by contract.
}
}
}
diff --git a/src/Primitives.Extensions.Shared/Operators/ScheduledValueObservable.cs b/src/Primitives.Extensions.Shared/Operators/ScheduledValueObservable.cs
index 4d9dc857..d5a61fa6 100644
--- a/src/Primitives.Extensions.Shared/Operators/ScheduledValueObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/ScheduledValueObservable.cs
@@ -11,16 +11,15 @@ namespace ReactiveUI.Primitives.Extensions.Operators;
#endif
///
-/// Single-value scheduled observable. On subscription, schedules a callback on
-/// the supplied that applies an optional
-/// side-effect and/or an optional
-/// transform to the captured value, calls
-/// once, then . Replaces the
-/// Observable.Create<T>(o => scheduler.Schedule[Safe](due, () => o.OnNext(...)))
-/// family of single-value Schedule overloads with one dedicated type
-/// that captures only the fields each overload actually uses.
+/// Single-value scheduled observable: on subscription, schedules a callback on the supplied
+/// that applies an optional side-effect and/or an optional
+/// transform to the captured value and calls once.
///
/// The value type emitted to the downstream observer.
+///
+/// The sequence never completes on its own: no terminal notification is sent, so the subscriber ends it by
+/// disposing the subscription.
+///
internal sealed class ScheduledValueObservable : IObservable
{
/// The value to emit.
@@ -121,13 +120,9 @@ private sealed class EmitState(
Func? transform,
Action? action)
{
- /// Applies the optional side-effect and transform, then emits the value followed by completion to the captured observer.
+ /// Applies the optional side-effect and transform, then emits the value to the captured observer.
public void Emit()
{
- // Preserves the original Observable.Create-based semantics: the
- // scheduled callback only emits OnNext. The sequence completes
- // when downstream subscribers dispose; we do not auto-call
- // OnCompleted here.
try
{
action?.Invoke(value);
diff --git a/src/Primitives.Extensions.Shared/Operators/SyncTimerObservable.cs b/src/Primitives.Extensions.Shared/Operators/SyncTimerObservable.cs
index e683b50a..fe782db6 100644
--- a/src/Primitives.Extensions.Shared/Operators/SyncTimerObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/SyncTimerObservable.cs
@@ -12,12 +12,7 @@ namespace ReactiveUI.Primitives.Extensions.Reactive.Operators;
namespace ReactiveUI.Primitives.Extensions.Operators;
#endif
-///
-/// Optimized operator that shares a single timer per (TimeSpan, ISequencer) key.
-/// Replaces the manual ConcurrentDictionary<..., Lazy<SharedTimer>> shape with a stateful
-/// ConcurrentDictionary.GetOrAdd overload that doesn't allocate a
-/// or its factory delegate on the hot path.
-///
+/// Caches and shares one running timer per (TimeSpan, ISequencer) key.
internal static class SyncTimerObservable
{
/// The timer cache, keyed by (TimeSpan, ISequencer).
@@ -40,9 +35,8 @@ internal static IObservable Get(TimeSpan timeSpan, ISequencer schedule
}
///
- /// A manual implementation of a connectable timer that minimizes allocations and unrolls Rx chains.
- /// Tick uses a swap-on-write array so the read path is allocation-free
- /// and lock-free; subscribe / unsubscribe takes the gate and publishes a fresh array.
+ /// Connectable timer that fans each tick out to its observers: the tick path reads a swap-on-write observer
+ /// array lock-free, while subscribe and unsubscribe take the gate and publish a fresh array.
///
/// The period.
/// The scheduler.
@@ -55,9 +49,8 @@ private sealed class SharedTimer(TimeSpan timeSpan, ISequencer scheduler) : IObs
private readonly Lock _gate = new();
///
- /// Snapshot of currently active observers. Replaced (not mutated) on subscribe / unsubscribe under
- /// . The tick path reads this via Volatile.Read with no lock and no
- /// allocation.
+ /// Snapshot of active observers, replaced rather than mutated on subscribe and unsubscribe under
+ /// so the tick path can read it without the lock.
///
private IObserver[] _observers = _emptyObservers;
@@ -87,9 +80,7 @@ public IDisposable Subscribe(IObserver observer)
return new TimerSubscription(this, observer);
}
- /// Ticks every currently-subscribed observer with the scheduler's current time.
- /// The empty-array short-circuit lives in
- /// (excluded from coverage) so this hot path stays branchless on the steady state.
+ /// Ticks every currently-subscribed observer with the scheduler's current time.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private void Tick() =>
ObserverArrayHelpers.Broadcast(Volatile.Read(ref _observers), scheduler.Now.DateTime);
@@ -100,29 +91,21 @@ private void Remove(IObserver observer)
{
lock (_gate)
{
- // TimerSubscription.Dispose's Interlocked guard ensures Remove is called at most
- // once per subscription, and each subscription's observer was placed in _observers
- // under this same lock before the disposable was returned — so RemoveOrNull always
- // locates the observer by construction.
+ // Never null: Dispose's Interlocked guard admits one Remove per subscription, and the
+ // observer was added under this same lock before the disposable was handed out.
var updated = ObserverArrayHelpers.RemoveOrNull(_observers, observer, _emptyObservers)!;
Volatile.Write(ref _observers, updated);
if (ReferenceEquals(updated, _emptyObservers))
{
- // Subscribe sets _timerSubscription on first add, before the disposable is
- // returned; if we reach the "all observers gone" branch, at least one Subscribe
- // ran, so _timerSubscription is non-null by construction.
+ // Never null: reaching an empty set means Subscribe ran, which arms the timer.
_timerSubscription!.Dispose();
_timerSubscription = null;
}
}
}
- ///
- /// Per-subscribe disposable. Holding (parent, observer) as fields instead of capturing them in
- /// a lambda removes the per-subscribe closure allocation that would
- /// have required.
- ///
+ /// Per-subscribe disposable that detaches its observer from the owning timer exactly once.
/// The owning timer.
/// The observer to remove on dispose.
private sealed class TimerSubscription(SharedTimer parent, IObserver observer) : IDisposable
diff --git a/src/Primitives.Extensions.Shared/Operators/ThrottleDistinctObservable.cs b/src/Primitives.Extensions.Shared/Operators/ThrottleDistinctObservable.cs
index 673cfe60..6a4fcecf 100644
--- a/src/Primitives.Extensions.Shared/Operators/ThrottleDistinctObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/ThrottleDistinctObservable.cs
@@ -27,18 +27,13 @@ public IDisposable Subscribe(IObserver observer)
InvalidOperationExceptionHelper.ThrowIfNull(scheduler);
ArgumentExceptionHelper.ThrowIfNull(observer);
- // Implementation of .DistinctUntilChanged().Throttle(throttle, scheduler).DistinctUntilChanged()
- // But fused into a single sink to avoid multiple operator allocations and observer chains.
+ // Equivalent to DistinctUntilChanged().Throttle(throttle, scheduler).DistinctUntilChanged(), fused into one sink.
ThrottleDistinctSink sink = new(observer, throttle, scheduler);
var subscription = source.Subscribe(sink);
return new DisposableBag(subscription, sink);
}
- ///
- /// Sink that implements the throttle distinct logic. Composes
- /// for the shared gate / timer / done-flag plumbing so this class only carries the throttle
- /// and distinct-value tracking.
- ///
+ /// Sink that emits the latest value once the throttle window elapses, skipping it when it equals the last emitted value.
/// The observer to forward elements to.
/// The throttle duration.
/// The scheduler to use for timing.
@@ -115,17 +110,14 @@ public void Dispose()
}
}
- /// Emits the last received value if it differs from the last emitted value.
- /// Marked [ExcludeFromCodeCoverage] because the in-lock
- /// race-loser branch (sink done or no buffered value) is only reachable when the
- /// scheduled callback fires concurrently with Dispose / OnCompleted, which the
- /// single-threaded test harness cannot trigger.
+ /// Emits the last received value when it differs from the last emitted value.
[System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
private void Emit()
{
T? toEmit;
lock (_gate)
{
+ // Race-only: reachable when this scheduled callback overlaps Dispose or a terminal notification.
if (_state.Done || !_hasLastReceived)
{
return;
diff --git a/src/Primitives.Extensions.Shared/Operators/ThrottleObservable.cs b/src/Primitives.Extensions.Shared/Operators/ThrottleObservable.cs
index 317ac0fc..c54836b6 100644
--- a/src/Primitives.Extensions.Shared/Operators/ThrottleObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/ThrottleObservable.cs
@@ -145,12 +145,8 @@ public void Dispose()
}
///
- /// Emits the buffered value if it is still current (i.e. no newer
- /// arrived after this emission was scheduled).
- /// Marked [ExcludeFromCodeCoverage] because the in-lock
- /// race-loser branch (sink done, emission superseded, value already drained) is only
- /// reachable when the scheduled callback fires concurrently with Dispose / OnCompleted,
- /// which the single-threaded test harness cannot trigger.
+ /// Emits the buffered value when it is still current, meaning no newer arrived after
+ /// this emission was scheduled.
///
/// The emission id this callback was scheduled for.
[System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
@@ -159,6 +155,7 @@ private void Emit(long id)
T value;
lock (_gate)
{
+ // Race-only: reachable when this scheduled callback overlaps Dispose or a terminal notification.
if (_done || id != _emissionId || !_hasValue)
{
return;
diff --git a/src/Primitives.Extensions.Shared/Operators/ThrottleUntilTrueObservable.cs b/src/Primitives.Extensions.Shared/Operators/ThrottleUntilTrueObservable.cs
index 0d344b82..04946c35 100644
--- a/src/Primitives.Extensions.Shared/Operators/ThrottleUntilTrueObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/ThrottleUntilTrueObservable.cs
@@ -15,10 +15,12 @@ namespace ReactiveUI.Primitives.Extensions.Operators;
/// The source observable.
/// The throttle duration.
/// The predicate to determine if an element should be emitted immediately or throttled.
+/// The sequencer timing throttled emissions; null uses the default sequencer.
internal sealed class ThrottleUntilTrueObservable(
IObservable source,
TimeSpan throttle,
- Func predicate) : IObservable
+ Func predicate,
+ ISequencer? sequencer = null) : IObservable
{
/// The source observable.
private readonly IObservable _source = InvalidOperationExceptionHelper.Check(source);
@@ -29,17 +31,20 @@ internal sealed class ThrottleUntilTrueObservable(
/// The predicate to determine if an element should be emitted immediately or throttled.
private readonly Func _predicate = InvalidOperationExceptionHelper.Check(predicate);
+ /// The sequencer timing throttled emissions.
+ private readonly ISequencer _sequencer = sequencer ?? Sequencer.Default;
+
///
public IDisposable Subscribe(IObserver observer)
{
ArgumentExceptionHelper.ThrowIfNull(observer);
- ThrottleUntilTrueSink sink = new(observer, _throttle, _predicate, Sequencer.Default);
+ ThrottleUntilTrueSink sink = new(observer, _throttle, _predicate, _sequencer);
var subscription = _source.Subscribe(sink);
return new DisposableBag(subscription, sink);
}
- /// Sinks the source observable and throttles elements until a predicate is true.
+ /// Sink that forwards a value inline when the predicate holds and otherwise after the throttle window.
/// The downstream observer.
/// The throttle duration.
/// The predicate.
diff --git a/src/Primitives.Extensions.Shared/Operators/UsingActionObservable.cs b/src/Primitives.Extensions.Shared/Operators/UsingActionObservable.cs
index b1a3b22d..cdc2e1e2 100644
--- a/src/Primitives.Extensions.Shared/Operators/UsingActionObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/UsingActionObservable.cs
@@ -11,11 +11,8 @@ namespace ReactiveUI.Primitives.Extensions.Operators;
#endif
///
-/// Resource-bound factory observable that runs an optional
-/// against a captured
-/// resource, emits , completes, and finally disposes
-/// the resource. Replaces the legacy
-/// Observable.Using(() => obj, id => Observable.Start(...)) pattern.
+/// Resource-bound factory observable that runs an optional against a captured
+/// resource, emits , completes, and disposes the resource.
///
/// The disposable resource type.
/// The resource to use during the operation and dispose at the end.
@@ -55,7 +52,7 @@ public IDisposable Subscribe(IObserver observer)
"Design",
"SST1429:Handle, rethrow, or narrow this catch; an empty catch of the base exception hides failures",
Justification =
- "A secondary dispose failure during error handling is intentionally swallowed so the primary exception sent to the observer wins; this assembly has no unhandled-exception sink.")]
+ "A secondary dispose failure is swallowed so the primary exception reaching the observer wins.")]
private static void Run(IObserver observer, T resource, Action? action)
{
try
diff --git a/src/Primitives.Extensions.Shared/Operators/UsingFuncObservable.cs b/src/Primitives.Extensions.Shared/Operators/UsingFuncObservable.cs
index 768da676..14755c3c 100644
--- a/src/Primitives.Extensions.Shared/Operators/UsingFuncObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/UsingFuncObservable.cs
@@ -11,10 +11,8 @@ namespace ReactiveUI.Primitives.Extensions.Operators;
#endif
///
-/// Resource-bound factory observable that runs a
-/// against a captured resource, emits the result,
-/// completes, and finally disposes the resource. Replaces the legacy
-/// Observable.Using(() => obj, id => Observable.Start(() => func(id))) pattern.
+/// Resource-bound factory observable that runs a against a captured
+/// resource, emits the result, completes, and disposes the resource.
///
/// The disposable resource type.
/// The result type emitted to the downstream observer.
@@ -59,7 +57,7 @@ public IDisposable Subscribe(IObserver observer)
"Design",
"SST1429:Handle, rethrow, or narrow this catch; an empty catch of the base exception hides failures",
Justification =
- "A secondary dispose failure during error handling is intentionally swallowed so the primary exception sent to the observer wins; this assembly has no unhandled-exception sink.")]
+ "A secondary dispose failure is swallowed so the primary exception reaching the observer wins.")]
private static void Run(IObserver observer, T resource, Func function)
{
TResult result;
diff --git a/src/Primitives.Extensions.Shared/Operators/WhileObservable.cs b/src/Primitives.Extensions.Shared/Operators/WhileObservable.cs
index ae87e75b..214ce2d9 100644
--- a/src/Primitives.Extensions.Shared/Operators/WhileObservable.cs
+++ b/src/Primitives.Extensions.Shared/Operators/WhileObservable.cs
@@ -12,12 +12,9 @@ namespace ReactiveUI.Primitives.Extensions.Operators;
#endif
///
-/// Loops the supplied on the supplied
-/// (or inline when no scheduler is provided), emitting
-/// after each iteration, for as long as
-/// returns true. Replaces the legacy
-/// Observable.While(condition, Observable.Start(action, scheduler))
-/// pattern.
+/// Loops the supplied on the supplied , or inline when no scheduler is
+/// provided, emitting after each iteration for as long as
+/// returns true.
///
/// The loop predicate. Evaluated before each iteration.
/// The action to invoke per iteration.
diff --git a/src/Primitives.Extensions.Shared/ReactiveExtensions.cs b/src/Primitives.Extensions.Shared/ReactiveExtensions.cs
index 7ad51784..33feaa0b 100644
--- a/src/Primitives.Extensions.Shared/ReactiveExtensions.cs
+++ b/src/Primitives.Extensions.Shared/ReactiveExtensions.cs
@@ -20,7 +20,7 @@ namespace ReactiveUI.Primitives.Extensions;
[SuppressMessage(
"Roslynator",
"RCS1047:Non-asynchronous method name should not end with \'Async\'",
- Justification = "Existing API")]
+ Justification = "The SubscribeAsync overloads are named for the asynchronous handler they invoke and return a subscription handle rather than a task.")]
public static partial class ReactiveExtensions
{
/// Default match timeout for regex filters created from string patterns.
@@ -42,16 +42,16 @@ public IObservable Start(ISequencer? scheduler) =>
/// Condition to evaluate.
extension(Func condition)
{
- /// While construct.
+ /// Runs the action inline, emitting after each iteration for as long as the condition holds.
/// Action to execute.
/// Observable representing the loop.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public IObservable While(Action action) =>
While(condition, action, null);
- /// While construct.
+ /// Runs the action on the scheduler, emitting after each iteration for as long as the condition holds.
/// Action to execute.
- /// Scheduler.
+ /// Scheduler each iteration runs on; null runs inline.
/// Observable representing the loop.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public IObservable While(Action action, ISequencer? scheduler) =>
@@ -62,13 +62,13 @@ public IObservable While(Action action, ISequencer? scheduler) =>
/// The sources.
extension(IEnumerable> sources)
{
- /// Latest values of each sequence are all false.
+ /// Combines the latest value of every source and reports whether they are all false.
/// A sequence that emits true when all latest booleans are false.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public IObservable CombineLatestValuesAreAllFalse() =>
new BooleanReduceObservable(sources, false);
- /// Latest values of each sequence are all true.
+ /// Combines the latest value of every source and reports whether they are all true.
/// A sequence that emits true when all latest booleans are true.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public IObservable CombineLatestValuesAreAllTrue() =>
@@ -415,7 +415,7 @@ public IObservable SwitchIfEmpty(IObservable fallback) =>
[SuppressMessage(
"Roslynator",
"RCS1047:Non-asynchronous method name should not end with \'Async\'",
- Justification = "This is an existing method")]
+ Justification = "The name describes the asynchronous handler this overload invokes; it returns a subscription handle rather than a task.")]
public IDisposable SubscribeAsync(Func onNext) =>
new SubscribeAsyncObservable(source, onNext, null, null);
@@ -429,7 +429,7 @@ public IDisposable SubscribeAsync(Func onNext) =>
[SuppressMessage(
"Roslynator",
"RCS1047:Non-asynchronous method name should not end with \'Async\'",
- Justification = "This is an existing method")]
+ Justification = "The name describes the asynchronous handler this overload invokes; it returns a subscription handle rather than a task.")]
public IDisposable SubscribeAsync(Func onNext, Action onCompleted) =>
new SubscribeAsyncObservable(source, onNext, null, onCompleted);
@@ -443,7 +443,7 @@ public IDisposable SubscribeAsync(Func onNext, Action onCompleted)
[SuppressMessage(
"Roslynator",
"RCS1047:Non-asynchronous method name should not end with \'Async\'",
- Justification = "This is an existing method")]
+ Justification = "The name describes the asynchronous handler this overload invokes; it returns a subscription handle rather than a task.")]
public IDisposable SubscribeAsync(Func onNext, Action onError) =>
new SubscribeAsyncObservable(source, onNext, onError, null);
@@ -458,7 +458,7 @@ public IDisposable SubscribeAsync(Func onNext, Action o
[SuppressMessage(
"Roslynator",
"RCS1047:Non-asynchronous method name should not end with \'Async\'",
- Justification = "This is an existing method")]
+ Justification = "The name describes the asynchronous handler this overload invokes; it returns a subscription handle rather than a task.")]
public IDisposable SubscribeAsync(
Func onNext,
Action onError,
diff --git a/src/Primitives.Shared/Advanced/AfterSignal.cs b/src/Primitives.Shared/Advanced/AfterSignal.cs
index 65f681a6..20ef9072 100644
--- a/src/Primitives.Shared/Advanced/AfterSignal.cs
+++ b/src/Primitives.Shared/Advanced/AfterSignal.cs
@@ -20,12 +20,12 @@ public sealed class AfterSignal : IRequireCurrentThread
/// The recurring period after the first tick, when this is a periodic timer.
private readonly TimeSpan? _period;
- /// The sequencer used to schedule the tick.
+ /// The sequencer that schedules the tick.
private readonly ISequencer _scheduler;
/// Initializes a new instance of the class.
/// The delay before the single tick.
- /// The sequencer used to schedule the tick.
+ /// The sequencer that schedules the tick.
public AfterSignal(TimeSpan dueTime, ISequencer scheduler)
{
_dueTime = dueTime;
@@ -35,7 +35,7 @@ public AfterSignal(TimeSpan dueTime, ISequencer scheduler)
/// Initializes a new instance of the class.
/// The delay before the first tick.
/// The period between subsequent ticks.
- /// The sequencer used to schedule ticks.
+ /// The sequencer that schedules ticks.
public AfterSignal(TimeSpan dueTime, TimeSpan period, ISequencer scheduler)
{
ArgumentOutOfRangeExceptionHelper.ThrowIfLessThan(period, TimeSpan.Zero);
diff --git a/src/Primitives.Shared/Advanced/AfterSubscription.cs b/src/Primitives.Shared/Advanced/AfterSubscription.cs
index 1b789c87..533b2aab 100644
--- a/src/Primitives.Shared/Advanced/AfterSubscription.cs
+++ b/src/Primitives.Shared/Advanced/AfterSubscription.cs
@@ -16,7 +16,7 @@ public sealed class AfterSubscription : IDisposable
{
/// Initializes a new instance of the class.
/// The downstream observer.
- /// The scheduler used to emit ticks.
+ /// The scheduler that emits ticks.
/// The delay before the first tick.
/// The period between subsequent ticks, or for one-shot timers.
public AfterSubscription(IObserver observer, ISequencer scheduler, TimeSpan dueTime, TimeSpan? period)
@@ -30,7 +30,7 @@ public AfterSubscription(IObserver observer, ISequencer scheduler, TimeSpa
/// Gets the downstream observer.
private IObserver Observer { get; }
- /// Gets the scheduler used to emit ticks.
+ /// Gets the scheduler that emits ticks.
private ISequencer Scheduler { get; }
/// Gets the delay before the first tick.
diff --git a/src/Primitives.Shared/Advanced/AsyncSubscriptionLifetime.cs b/src/Primitives.Shared/Advanced/AsyncSubscriptionLifetime.cs
index 76b8e231..020b85ba 100644
--- a/src/Primitives.Shared/Advanced/AsyncSubscriptionLifetime.cs
+++ b/src/Primitives.Shared/Advanced/AsyncSubscriptionLifetime.cs
@@ -47,7 +47,7 @@ public sealed class AsyncSubscriptionLifetime : IDisposable
public void SetSubscription(IDisposable? disposable) =>
_subscription.Create(disposable ?? EmptyDisposable.Instance);
- /// Marks asynchronous setup complete and releases the cancellation source when still owned here.
+ /// Marks asynchronous setup complete and releases the cancellation source while it is owned here.
public void Complete() => _ = TryComplete();
///
@@ -71,7 +71,7 @@ public void Dispose()
_cts.Dispose();
}
- /// Attempts to mark asynchronous setup complete and release the cancellation source when still owned here.
+ /// Attempts to mark asynchronous setup complete and release the cancellation source while it is owned here.
/// when this call completed the lifetime.
internal bool TryComplete()
{
@@ -100,7 +100,7 @@ private static void CancelIgnoringDisposed(CancellationTokenSource cts)
}
catch (ObjectDisposedException)
{
- // Completion can release the CTS concurrently; disposal still continues with the inner subscription.
+ // Completion can release the CTS concurrently; disposal continues with the inner subscription.
}
}
}
diff --git a/src/Primitives.Shared/Advanced/AutoShareSignal{T}.cs b/src/Primitives.Shared/Advanced/AutoShareSignal{T}.cs
index c461fe9f..a2a3e03c 100644
--- a/src/Primitives.Shared/Advanced/AutoShareSignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/AutoShareSignal{T}.cs
@@ -95,10 +95,9 @@ internal void Release(IDisposable subscription)
/// Connects the source outside and publishes or drops the connection.
/// The inner source subscription owned by the connecting observer.
///
- /// Connecting runs outside the lock so a synchronous source cannot drive user callbacks while the
- /// gate is held. A re-entrant or concurrent can drop the subscriber count to
- /// zero before the connection is published; in that case the freshly returned connection is orphaned
- /// and is disposed here rather than stored.
+ /// Connecting runs outside the lock so a synchronous source cannot drive user callbacks while the gate is
+ /// held, which lets a re-entrant or concurrent drain the subscriber count before the
+ /// connection is published.
///
private void ConnectOutsideGate(IDisposable subscription)
{
diff --git a/src/Primitives.Shared/Advanced/BufferSignal{T}.cs b/src/Primitives.Shared/Advanced/BufferSignal{T}.cs
index a8a5117b..80e958b1 100644
--- a/src/Primitives.Shared/Advanced/BufferSignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/BufferSignal{T}.cs
@@ -16,7 +16,7 @@ public sealed class BufferSignal : IObservable>
/// Initializes a new instance of the class.
/// The source observable.
/// The buffer window duration.
- /// The sequencer used to schedule buffer flushes.
+ /// The sequencer that schedules buffer flushes.
/// or is .
public BufferSignal(IObservable source, TimeSpan timeSpan, ISequencer sequencer)
{
@@ -31,7 +31,7 @@ public BufferSignal(IObservable source, TimeSpan timeSpan, ISequencer sequenc
/// Gets the buffer window duration.
private TimeSpan Window { get; }
- /// Gets the sequencer used to schedule buffer flushes.
+ /// Gets the sequencer that schedules buffer flushes.
private ISequencer Scheduler { get; }
///
diff --git a/src/Primitives.Shared/Advanced/CatchSignal{T}.cs b/src/Primitives.Shared/Advanced/CatchSignal{T}.cs
index 5be0be31..66e650ed 100644
--- a/src/Primitives.Shared/Advanced/CatchSignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/CatchSignal{T}.cs
@@ -10,75 +10,74 @@ namespace ReactiveUI.Primitives.Reactive.Advanced;
namespace ReactiveUI.Primitives.Advanced;
#endif
-/// Represents the CatchSignal class.
-/// The T type.
+/// Subscribes to each source in turn, moving to the next one whenever a source errors.
+/// The value type.
internal sealed class CatchSignal : IRequireCurrentThread
{
- /// Stores state for the signal implementation.
+ /// The sources tried in order.
private readonly IEnumerable> _sources;
/// Initializes a new instance of the class.
- /// The sources value.
+ /// The sources to try in order.
public CatchSignal(IEnumerable> sources) => _sources = sources;
- /// Executes the IsRequiredSubscribeOnCurrentThread operation.
- /// The result.
+ /// Reports that subscription runs on the calling thread.
+ /// Always .
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool IsRequiredSubscribeOnCurrentThread() => true;
- /// Executes the Subscribe operation.
- /// The observer value.
- /// The result.
+ /// Subscribes the observer and starts walking the sources.
+ /// The downstream observer.
+ /// The disposable that tears the walk down.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public IDisposable Subscribe(IObserver observer) =>
SignalSubscription.Subscribe(observer, true, SubscribeCore);
- /// Executes the SubscribeCore operation.
- /// The observer value.
- /// The cancel value.
- /// The result.
+ /// Creates the handler that walks the sources and starts it.
+ /// The downstream observer.
+ /// The outer subscription handle.
+ /// The disposable that tears the walk down.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private IDisposable SubscribeCore(IObserver observer, IDisposable cancel) =>
new Catch(this, observer, cancel).Run();
- /// Represents the Catch class.
+ /// Walks the source sequence, advancing on each error and forwarding the last error if none succeed.
private sealed class Catch : IObserver, IDisposable
{
- /// Stores state for the signal implementation.
+ /// The signal supplying the sources.
private readonly CatchSignal _parent;
- /// Stores the downstream observer.
+ /// The downstream observer.
private readonly IObserver _observer;
- /// Executes the new operation.
- /// The result.
+ /// Serializes advancing the enumerator against teardown.
private readonly Lock _gate = new();
- /// Stores the upstream subscription.
+ /// The outer subscription handle released on teardown.
private IDisposable? _cancel;
/// Disposed latch; 0 when alive, 1 once disposed.
private int _disposed;
- /// Stores state for the signal implementation.
+ /// Set under once teardown ran, so no further source is subscribed.
private bool _isDisposed;
- /// Stores state for the signal implementation.
+ /// The enumerator over the sources.
private IEnumerator>? _e;
- /// Stores state for the signal implementation.
+ /// The slot holding the current source subscription.
private SingleReplaceableDisposable? _subscription;
- /// Stores state for the signal implementation.
+ /// The error raised by the most recent source.
private Exception? _lastException;
- /// Stores state for the signal implementation.
+ /// The recursive continuation that advances to the next source.
private Action? _nextSelf;
/// Initializes a new instance of the class.
- /// The parent value.
- /// The observer value.
- /// The cancel value.
+ /// The signal supplying the sources.
+ /// The downstream observer.
+ /// The outer subscription handle.
/// is .
public Catch(CatchSignal parent, IObserver observer, IDisposable cancel)
{
@@ -87,8 +86,8 @@ public Catch(CatchSignal parent, IObserver observer, IDisposable cancel)
_parent = parent;
}
- /// Executes the Run operation.
- /// The result.
+ /// Starts the walk on the immediate sequencer.
+ /// The disposable that releases the enumerator and the current source subscription.
public MultipleDisposable Run()
{
_isDisposed = false;
@@ -108,20 +107,20 @@ public MultipleDisposable Run()
}));
}
- /// Executes the OnNext operation.
- /// The value.
+ /// Forwards a value downstream.
+ /// The value to forward.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void OnNext(T value) => _observer.OnNext(value);
- /// Executes the OnError operation.
- /// The error value.
+ /// Records the error and advances to the next source instead of terminating.
+ /// The error raised by the current source.
public void OnError(Exception error)
{
_lastException = error;
_nextSelf!();
}
- /// Executes the OnCompleted operation.
+ /// Completes downstream and tears the walk down.
public void OnCompleted()
{
try
@@ -134,7 +133,7 @@ public void OnCompleted()
}
}
- /// Executes the Dispose operation.
+ /// Releases the enumerator, the current source subscription and the outer handle.
public void Dispose()
{
_e?.Dispose();
@@ -144,8 +143,8 @@ public void Dispose()
_ = WitnessTeardown.Dispose(ref _disposed, ref _cancel);
}
- /// Executes the RecursiveRun operation.
- /// The self value.
+ /// Subscribes to the next source, or terminates once the sequence is exhausted.
+ /// The continuation that re-enters this method for the following source.
private void RecursiveRun(Action self)
{
lock (_gate)
@@ -172,7 +171,7 @@ private void RecursiveRun(Action self)
}
}
- /// Advances the handler sequence to the next source. Call while holding the gate.
+ /// Advances the enumerator to the next source. Call while holding .
/// The next source, or once the sequence is exhausted.
/// The exception the sequence raised, when it raised one.
/// when the sequence advanced without raising.
@@ -203,7 +202,7 @@ private bool TryMoveToNextSource(out IObservable? next, out Exception? error)
}
}
- /// Forwards an error downstream and tears the handler down.
+ /// Forwards an error downstream and tears the walk down.
/// The error to forward.
private void FailAndDispose(Exception error)
{
diff --git a/src/Primitives.Shared/Advanced/CollectSignal{T}.cs b/src/Primitives.Shared/Advanced/CollectSignal{T}.cs
index 7d157de7..f5594754 100644
--- a/src/Primitives.Shared/Advanced/CollectSignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/CollectSignal{T}.cs
@@ -16,7 +16,7 @@ public sealed class CollectSignal : IObservable>
/// Initializes a new instance of the class.
/// The source observable.
/// The buffer window duration.
- /// The sequencer used to schedule buffer flushes.
+ /// The sequencer that schedules buffer flushes.
/// or is .
public CollectSignal(IObservable source, TimeSpan timeSpan, ISequencer sequencer)
{
@@ -31,7 +31,7 @@ public CollectSignal(IObservable source, TimeSpan timeSpan, ISequencer sequen
/// Gets the buffer window duration.
private TimeSpan TimeSpan { get; }
- /// Gets the sequencer used to schedule buffer flushes.
+ /// Gets the sequencer that schedules buffer flushes.
private ISequencer Sequencer { get; }
///
diff --git a/src/Primitives.Shared/Advanced/CollectWitness{T}.cs b/src/Primitives.Shared/Advanced/CollectWitness{T}.cs
index 0c688716..eaaf6a54 100644
--- a/src/Primitives.Shared/Advanced/CollectWitness{T}.cs
+++ b/src/Primitives.Shared/Advanced/CollectWitness{T}.cs
@@ -18,7 +18,7 @@ public sealed class CollectWitness : IObserver, IDisposable
/// Serializes access to buffered values and terminal state.
private readonly Lock _gate = new();
- /// Whether a flush has already been scheduled for the current window.
+ /// Whether a flush is scheduled for the current window.
private bool _flushScheduled;
/// Whether the source has terminated.
@@ -33,7 +33,7 @@ public CollectWitness(IObserver> observer) =>
/// Initializes a new instance of the class.
/// The downstream observer.
/// The buffer window duration.
- /// The sequencer used to schedule flushes.
+ /// The sequencer that schedules flushes.
/// or is .
public CollectWitness(IObserver> observer, TimeSpan timeSpan, ISequencer sequencer)
{
@@ -48,7 +48,7 @@ public CollectWitness(IObserver> observer, TimeSpan timeSpan, ISequence
/// Gets the buffer window duration.
private TimeSpan TimeSpan { get; }
- /// Gets the sequencer used to schedule flushes.
+ /// Gets the sequencer that schedules flushes.
private ISequencer? Sequencer { get; }
/// Gets the source subscription and scheduled flushes.
@@ -132,7 +132,7 @@ public void OnNext(T value)
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void SetSubscription(IDisposable subscription) => Disposables.Add(subscription);
- /// Flushes the current window if it still has buffered values.
+ /// Flushes the current window when it holds buffered values.
private void Flush()
{
var batch = TakeScheduledBatch();
@@ -177,7 +177,7 @@ private bool TryRecord(T value)
}
}
- /// Marks the observer as stopped if it has not already stopped.
+ /// Transitions the observer to the stopped state.
/// when this call stopped the observer.
private bool TryMarkStopped()
{
diff --git a/src/Primitives.Shared/Advanced/CreateSafeSignal{T}.cs b/src/Primitives.Shared/Advanced/CreateSafeSignal{T}.cs
index 6fea332c..0fc3c76d 100644
--- a/src/Primitives.Shared/Advanced/CreateSafeSignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/CreateSafeSignal{T}.cs
@@ -10,37 +10,38 @@ namespace ReactiveUI.Primitives.Reactive.Advanced;
namespace ReactiveUI.Primitives.Advanced;
#endif
-/// Represents the CreateSafeSignal class.
-/// The T type.
+/// Creates a signal from a subscribe delegate whose sink releases the subscription when a downstream
+/// OnNext throws.
+/// The value type.
internal sealed class CreateSafeSignal : IRequireCurrentThread
{
- /// Stores state for the signal implementation.
+ /// The delegate invoked for each subscription.
private readonly Func, IDisposable> _subscribe;
/// Whether subscription must be dispatched through the current-thread sequencer.
private readonly bool _isRequiredSubscribeOnCurrentThread;
/// Initializes a new instance of the class.
- /// The subscribe value.
+ /// The delegate invoked for each subscription.
public CreateSafeSignal(Func, IDisposable> subscribe) => _subscribe = subscribe;
/// Initializes a new instance of the class.
- /// The subscribe value.
- /// The isRequiredSubscribeOnCurrentThread value.
+ /// The delegate invoked for each subscription.
+ /// Whether subscription must be dispatched through the current-thread sequencer.
public CreateSafeSignal(Func, IDisposable> subscribe, bool isRequiredSubscribeOnCurrentThread)
{
_subscribe = subscribe;
_isRequiredSubscribeOnCurrentThread = isRequiredSubscribeOnCurrentThread;
}
- /// Executes the IsRequiredSubscribeOnCurrentThread operation.
- /// The result.
+ /// Reports whether subscription must be dispatched through the current-thread sequencer.
+ /// when current-thread dispatch is required.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool IsRequiredSubscribeOnCurrentThread() => _isRequiredSubscribeOnCurrentThread;
- /// Executes the Subscribe operation.
- /// The observer value.
- /// The result.
+ /// Invokes the subscribe delegate with a sink wrapping the observer.
+ /// The downstream observer.
+ /// The disposable that releases the subscription.
public IDisposable Subscribe(IObserver observer)
{
ArgumentExceptionHelper.ThrowIfNull(observer);
@@ -55,10 +56,10 @@ public IDisposable Subscribe(IObserver observer)
return sink;
}
- /// Executes the SubscribeCore operation.
- /// The observer value.
- /// The cancel value.
- /// The result.
+ /// Invokes the subscribe delegate with a sink that owns .
+ /// The downstream observer.
+ /// The outer subscription handle.
+ /// The disposable returned by the subscribe delegate.
private IDisposable SubscribeCore(IObserver observer, IDisposable cancel)
{
CreateSink sink = new(observer, cancel, true);
diff --git a/src/Primitives.Shared/Advanced/CreateSignal{T,TState}.cs b/src/Primitives.Shared/Advanced/CreateSignal{T,TState}.cs
index d7c276b7..b67b18fb 100644
--- a/src/Primitives.Shared/Advanced/CreateSignal{T,TState}.cs
+++ b/src/Primitives.Shared/Advanced/CreateSignal{T,TState}.cs
@@ -10,23 +10,23 @@ namespace ReactiveUI.Primitives.Reactive.Advanced;
namespace ReactiveUI.Primitives.Advanced;
#endif
-/// Represents the CreateSignal class.
-/// The T type.
-/// The TState type.
+/// Creates a signal from a subscribe delegate that receives a caller-supplied state value.
+/// The value type.
+/// The state type handed to the subscribe delegate.
internal sealed class CreateSignal : IRequireCurrentThread
{
- /// Stores state for the signal implementation.
+ /// The state handed to the subscribe delegate.
private readonly TState _state;
- /// Stores state for the signal implementation.
+ /// The delegate invoked for each subscription.
private readonly Func, IDisposable> _subscribe;
- /// Stores state for the signal implementation.
+ /// Whether subscription must be dispatched through the current-thread sequencer.
private readonly bool _currentThreadRequired;
/// Initializes a new instance of the class.
- /// The state value.
- /// The subscribe value.
+ /// The state handed to the subscribe delegate.
+ /// The delegate invoked for each subscription.
public CreateSignal(TState state, Func, IDisposable> subscribe)
{
_state = state;
@@ -34,9 +34,9 @@ public CreateSignal(TState state, Func, IDisposable> subscr
}
/// Initializes a new instance of the class.
- /// The state value.
- /// The subscribe value.
- /// The isRequiredSubscribeOnCurrentThread value.
+ /// The state handed to the subscribe delegate.
+ /// The delegate invoked for each subscription.
+ /// Whether subscription must be dispatched through the current-thread sequencer.
public CreateSignal(
TState state,
Func, IDisposable> subscribe,
@@ -47,43 +47,43 @@ public CreateSignal(
_currentThreadRequired = isRequiredSubscribeOnCurrentThread;
}
- /// Executes the IsRequiredSubscribeOnCurrentThread operation.
- /// The result.
+ /// Reports whether subscription must be dispatched through the current-thread sequencer.
+ /// when current-thread dispatch is required.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool IsRequiredSubscribeOnCurrentThread() => _currentThreadRequired;
- /// Executes the Subscribe operation.
- /// The observer value.
- /// The result.
+ /// Invokes the subscribe delegate with the state and a wrapper around the observer.
+ /// The downstream observer.
+ /// The disposable that releases the subscription.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public IDisposable Subscribe(IObserver observer) =>
SignalSubscription.Subscribe(observer, _currentThreadRequired, SubscribeCore);
- /// Executes the SubscribeCore operation.
- /// The observer value.
- /// The cancel value.
- /// The result.
+ /// Invokes the subscribe delegate with a wrapper that owns .
+ /// The downstream observer.
+ /// The outer subscription handle.
+ /// The disposable returned by the subscribe delegate.
private IDisposable SubscribeCore(IObserver observer, IDisposable cancel)
{
observer = new Create(observer, cancel);
return _subscribe(_state, observer) ?? EmptyDisposable.Instance;
}
- /// Represents the Create class.
+ /// Forwards notifications downstream and releases the subscription on termination.
private sealed class Create : IObserver, IDisposable
{
- /// Stores the downstream observer.
+ /// The downstream observer.
private readonly IObserver _observer;
- /// Stores the upstream subscription.
+ /// The outer subscription handle released on teardown.
private IDisposable? _cancel;
/// Disposed latch; 0 when alive, 1 once disposed.
private int _disposed;
/// Initializes a new instance of the class.
- /// The observer value.
- /// The cancel value.
+ /// The downstream observer.
+ /// The outer subscription handle.
/// is .
public Create(IObserver observer, IDisposable cancel)
{
@@ -91,13 +91,13 @@ public Create(IObserver observer, IDisposable cancel)
_observer = observer;
}
- /// Executes the OnNext operation.
- /// The value.
+ /// Forwards a value downstream.
+ /// The value to forward.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void OnNext(T value) => _observer.OnNext(value);
- /// Executes the OnError operation.
- /// The error value.
+ /// Forwards the error downstream and releases the subscription.
+ /// The error to forward.
public void OnError(Exception error)
{
try
@@ -110,7 +110,7 @@ public void OnError(Exception error)
}
}
- /// Executes the OnCompleted operation.
+ /// Completes downstream and releases the subscription.
public void OnCompleted()
{
try
@@ -123,7 +123,7 @@ public void OnCompleted()
}
}
- /// Executes the Dispose operation.
+ /// Releases the outer subscription handle once.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void Dispose() => WitnessTeardown.Dispose(ref _disposed, ref _cancel);
}
diff --git a/src/Primitives.Shared/Advanced/CreateSignal{T}.cs b/src/Primitives.Shared/Advanced/CreateSignal{T}.cs
index 6b6f49d9..cd12bbe7 100644
--- a/src/Primitives.Shared/Advanced/CreateSignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/CreateSignal{T}.cs
@@ -10,37 +10,37 @@ namespace ReactiveUI.Primitives.Reactive.Advanced;
namespace ReactiveUI.Primitives.Advanced;
#endif
-/// Represents the CreateSignal class.
-/// The T type.
+/// Creates a signal from a caller-supplied subscribe delegate.
+/// The value type.
internal sealed class CreateSignal : IRequireCurrentThread
{
- /// Stores state for the signal implementation.
+ /// The delegate invoked for each subscription.
private readonly Func, IDisposable> _subscribe;
- /// Stores state for the signal implementation.
+ /// Whether subscription must be dispatched through the current-thread sequencer.
private readonly bool _currentThreadRequired;
/// Initializes a new instance of the class.
- /// The subscribe value.
+ /// The delegate invoked for each subscription.
public CreateSignal(Func, IDisposable> subscribe) => _subscribe = subscribe;
/// Initializes a new instance of the class.
- /// The subscribe value.
- /// The isRequiredSubscribeOnCurrentThread value.
+ /// The delegate invoked for each subscription.
+ /// Whether subscription must be dispatched through the current-thread sequencer.
public CreateSignal(Func, IDisposable> subscribe, bool isRequiredSubscribeOnCurrentThread)
{
_subscribe = subscribe;
_currentThreadRequired = isRequiredSubscribeOnCurrentThread;
}
- /// Executes the IsRequiredSubscribeOnCurrentThread operation.
- /// The result.
+ /// Reports whether subscription must be dispatched through the current-thread sequencer.
+ /// when current-thread dispatch is required.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool IsRequiredSubscribeOnCurrentThread() => _currentThreadRequired;
- /// Executes the Subscribe operation.
- /// The observer value.
- /// The result.
+ /// Invokes the subscribe delegate with a sink wrapping the observer.
+ /// The downstream observer.
+ /// The disposable that releases the subscription.
public IDisposable Subscribe(IObserver observer)
{
ArgumentExceptionHelper.ThrowIfNull(observer);
@@ -55,10 +55,10 @@ public IDisposable Subscribe(IObserver observer)
return sink;
}
- /// Executes the SubscribeCore operation.
- /// The observer value.
- /// The cancel value.
- /// The result.
+ /// Invokes the subscribe delegate with a sink that owns .
+ /// The downstream observer.
+ /// The outer subscription handle.
+ /// The disposable returned by the subscribe delegate.
private IDisposable SubscribeCore(IObserver observer, IDisposable cancel)
{
CreateSink sink = new(observer, cancel, false);
diff --git a/src/Primitives.Shared/Advanced/CreateSink{T}.cs b/src/Primitives.Shared/Advanced/CreateSink{T}.cs
index 21ebdf29..234c4735 100644
--- a/src/Primitives.Shared/Advanced/CreateSink{T}.cs
+++ b/src/Primitives.Shared/Advanced/CreateSink{T}.cs
@@ -11,10 +11,9 @@ namespace ReactiveUI.Primitives.Advanced;
#endif
///
-/// The observer handed to create-style subscription factories: forwards until terminated, owns the factory's
-/// cancel resource, and optionally releases it when a downstream OnNext throws (the safe-create contract).
-/// The shared sink behind and , usable by any
-/// implementation that hands an observer to a caller-supplied subscribe delegate.
+/// The observer handed to create-style subscription factories. It forwards notifications until terminated and owns
+/// the cancel resource the factory returns; when constructed with disposeOnNextThrow it also releases that
+/// resource before rethrowing if a downstream OnNext throws.
///
/// The value type.
[System.Diagnostics.DebuggerDisplay("CreateSink: Stopped = {_stopped}, Observer = {_observer}")]
@@ -60,7 +59,7 @@ public CreateSink(IObserver observer, IDisposable cancel, bool disposeOnNextT
_disposeOnNextThrow = disposeOnNextThrow;
}
- /// Assigns the cancellation resource, releasing it immediately when already stopped.
+ /// Assigns the cancellation resource, releasing it immediately when the sink has stopped.
/// Cancellation resource.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void SetCancel(IDisposable cancel) =>
diff --git a/src/Primitives.Shared/Advanced/DeferSignal{T}.cs b/src/Primitives.Shared/Advanced/DeferSignal{T}.cs
index 91db3726..b0dec20d 100644
--- a/src/Primitives.Shared/Advanced/DeferSignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/DeferSignal{T}.cs
@@ -10,33 +10,33 @@ namespace ReactiveUI.Primitives.Reactive.Advanced;
namespace ReactiveUI.Primitives.Advanced;
#endif
-/// Represents the DeferSignal class.
-/// The T type.
+/// Invokes a factory per subscription and subscribes the observer to the observable it returns.
+/// The value type.
internal sealed class DeferSignal : IRequireCurrentThread
{
- /// Stores state for the signal implementation.
+ /// The factory invoked for each subscription.
private readonly Func> _observableFactory;
/// Initializes a new instance of the class.
- /// The observableFactory value.
+ /// The factory invoked for each subscription.
public DeferSignal(Func> observableFactory) => _observableFactory = observableFactory;
- /// Executes the IsRequiredSubscribeOnCurrentThread operation.
- /// The result.
+ /// Reports that subscription needs no current-thread dispatch.
+ /// Always .
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool IsRequiredSubscribeOnCurrentThread() => false;
- /// Executes the Subscribe operation.
- /// The observer value.
- /// The result.
+ /// Subscribes the observer to the observable produced for this subscription.
+ /// The downstream observer.
+ /// The disposable that releases the subscription.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public IDisposable Subscribe(IObserver observer) =>
SignalSubscription.Subscribe(observer, false, SubscribeCore);
- /// Executes the SubscribeCore operation.
- /// The observer value.
- /// The cancel value.
- /// The result.
+ /// Invokes the factory and subscribes, turning a factory throw into an error notification.
+ /// The downstream observer.
+ /// The outer subscription handle.
+ /// The subscription to the produced observable.
private IDisposable SubscribeCore(IObserver observer, IDisposable cancel)
{
observer = new GuardedWitness(observer, cancel);
diff --git a/src/Primitives.Shared/Advanced/EmitIfQuietSignal{T}.cs b/src/Primitives.Shared/Advanced/EmitIfQuietSignal{T}.cs
index 7d313418..3b037e8a 100644
--- a/src/Primitives.Shared/Advanced/EmitIfQuietSignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/EmitIfQuietSignal{T}.cs
@@ -16,7 +16,7 @@ public sealed class EmitIfQuietSignal : IObservable
/// Initializes a new instance of the class.
/// The source observable.
/// The quiet period before emitting the latest value.
- /// The sequencer used to schedule delayed emissions.
+ /// The sequencer that schedules delayed emissions.
/// or is .
public EmitIfQuietSignal(IObservable source, TimeSpan dueTime, ISequencer sequencer)
{
@@ -31,7 +31,7 @@ public EmitIfQuietSignal(IObservable source, TimeSpan dueTime, ISequencer seq
/// Gets the quiet period before emitting the latest value.
private TimeSpan DueTime { get; }
- /// Gets the sequencer used to schedule delayed emissions.
+ /// Gets the sequencer that schedules delayed emissions.
private ISequencer Sequencer { get; }
///
diff --git a/src/Primitives.Shared/Advanced/EmitIfQuietWitness{T}.cs b/src/Primitives.Shared/Advanced/EmitIfQuietWitness{T}.cs
index 0aa0e3e9..2db5c71f 100644
--- a/src/Primitives.Shared/Advanced/EmitIfQuietWitness{T}.cs
+++ b/src/Primitives.Shared/Advanced/EmitIfQuietWitness{T}.cs
@@ -21,7 +21,7 @@ public sealed class EmitIfQuietWitness : IObserver, IDisposable
/// The latest observed value.
private T? _latest;
- /// Monotonic version used to suppress obsolete scheduled emissions.
+ /// Monotonic version that suppresses obsolete scheduled emissions.
private long _version;
/// Whether a latest value is pending emission.
@@ -33,7 +33,7 @@ public sealed class EmitIfQuietWitness : IObserver, IDisposable
/// Initializes a new instance of the class.
/// The downstream observer.
/// The quiet period before emitting the latest value.
- /// The sequencer used to schedule delayed emissions.
+ /// The sequencer that schedules delayed emissions.
/// or is .
public EmitIfQuietWitness(IObserver observer, TimeSpan dueTime, ISequencer sequencer)
{
@@ -48,7 +48,7 @@ public EmitIfQuietWitness(IObserver observer, TimeSpan dueTime, ISequencer se
/// Gets the quiet period before emitting the latest value.
private TimeSpan DueTime { get; }
- /// Gets the sequencer used to schedule delayed emissions.
+ /// Gets the sequencer that schedules delayed emissions.
private ISequencer Sequencer { get; }
/// Gets the source subscription and scheduled delayed emissions.
@@ -125,7 +125,7 @@ public void OnCompleted()
}
}
- /// Emits the latest value if the scheduled version is still current.
+ /// Emits the latest value when the scheduled version is the current one.
/// The version captured when the emission was scheduled.
private void EmitIfLatest(long scheduledVersion)
{
@@ -158,7 +158,7 @@ private bool TryRecord(T value, out long currentVersion)
}
}
- /// Marks the observer as stopped if it has not already stopped.
+ /// Transitions the observer to the stopped state.
/// when this call stopped the observer.
private bool TryMarkStopped()
{
diff --git a/src/Primitives.Shared/Advanced/EmptySignal{T}.cs b/src/Primitives.Shared/Advanced/EmptySignal{T}.cs
index 8451a01f..bf749610 100644
--- a/src/Primitives.Shared/Advanced/EmptySignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/EmptySignal{T}.cs
@@ -10,34 +10,34 @@ namespace ReactiveUI.Primitives.Reactive.Advanced;
namespace ReactiveUI.Primitives.Advanced;
#endif
-/// Represents the EmptySignal class.
-/// The T type.
+/// Completes without emitting a value, delivering completion on the supplied sequencer.
+/// The value type.
[System.Diagnostics.DebuggerDisplay("EmptySignal: Scheduler = {_scheduler}")]
public sealed class EmptySignal : IRequireCurrentThread
{
- /// Stores state for the signal implementation.
+ /// The sequencer that delivers completion.
private readonly ISequencer _scheduler;
/// Initializes a new instance of the class.
- /// The scheduler value.
+ /// The sequencer that delivers completion.
public EmptySignal(ISequencer scheduler) => _scheduler = scheduler;
- /// Executes the IsRequiredSubscribeOnCurrentThread operation.
- /// The result.
+ /// Reports that subscription needs no current-thread dispatch.
+ /// Always .
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool IsRequiredSubscribeOnCurrentThread() => false;
- /// Executes the Subscribe operation.
- /// The observer value.
- /// The result.
+ /// Subscribes the observer and arranges its completion.
+ /// The downstream observer.
+ /// The disposable that cancels a completion not yet delivered.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public IDisposable Subscribe(IObserver observer) =>
SignalSubscription.Subscribe(observer, false, SubscribeCore);
- /// Executes the SubscribeCore operation.
- /// The observer value.
- /// The cancel value.
- /// The result.
+ /// Completes the observer inline on the immediate sequencer, otherwise schedules the completion.
+ /// The downstream observer.
+ /// The outer subscription handle.
+ /// The disposable that cancels a scheduled completion.
private IDisposable SubscribeCore(IObserver observer, IDisposable cancel)
{
observer = new GuardedWitness(observer, cancel);
diff --git a/src/Primitives.Shared/Advanced/EverySignal.cs b/src/Primitives.Shared/Advanced/EverySignal.cs
index cb313d55..458ea6ca 100644
--- a/src/Primitives.Shared/Advanced/EverySignal.cs
+++ b/src/Primitives.Shared/Advanced/EverySignal.cs
@@ -10,19 +10,16 @@ namespace ReactiveUI.Primitives.Reactive.Advanced;
namespace ReactiveUI.Primitives.Advanced;
#endif
-///
-/// Dedicated signal for the interval timer factory (Every), replacing the self-referencing
-/// CreateSafe closure with a coordinator that reschedules itself through a method group.
-///
+/// Emits an incrementing tick at a fixed interval for the Every factory.
/// The interval between ticks.
-/// The sequencer used to schedule ticks.
+/// The sequencer that schedules ticks.
[System.Diagnostics.DebuggerDisplay("EverySignal: Period = {_period}, Scheduler = {_scheduler}")]
public sealed class EverySignal(TimeSpan period, ISequencer scheduler) : IRequireCurrentThread
{
/// The interval between ticks.
private readonly TimeSpan _period = period;
- /// The sequencer used to schedule ticks.
+ /// The sequencer that schedules ticks.
private readonly ISequencer _scheduler = scheduler;
///
@@ -50,13 +47,13 @@ public IDisposable Subscribe(IObserver observer)
return subscription;
}
- /// Reschedules the recurring tick without a captured closure.
+ /// Emits each tick and re-arms the schedule for the following one.
private sealed class EveryCoordinator : IDisposable
{
/// The downstream observer.
private readonly IObserver _observer;
- /// The sequencer used to schedule ticks.
+ /// The sequencer that schedules ticks.
private readonly ISequencer _scheduler;
/// The interval between ticks.
@@ -73,7 +70,7 @@ private sealed class EveryCoordinator : IDisposable
/// Initializes a new instance of the class.
/// The downstream observer.
- /// The sequencer used to schedule ticks.
+ /// The sequencer that schedules ticks.
/// The interval between ticks.
internal EveryCoordinator(IObserver observer, ISequencer scheduler, TimeSpan period)
{
@@ -100,14 +97,9 @@ internal EveryCoordinator Run()
private void ScheduleNext() => TimerSlot.Arm(_slot, _scheduler, _period, _tickAction);
/// Emits the current tick and reschedules unless cancelled.
- ///
- /// Disposal is tested before the emit as well as after it. A cancelled subscription must not deliver another
- /// value, and the sequencer's own cancellation check happens before the item is invoked, not before the
- /// observer is called. The second test covers an observer that disposes the subscription from inside
- /// , which must stop the recurring schedule rather than re-arm it.
- ///
private void Tick()
{
+ // The sequencer checks cancellation before invoking this item, not before the observer is called.
if (_slot.IsDisposed)
{
return;
@@ -116,6 +108,8 @@ private void Tick()
var tick = _tick;
_tick++;
_observer.OnNext(tick);
+
+ // An observer that disposes from inside OnNext must stop the schedule rather than re-arm it.
if (_slot.IsDisposed)
{
return;
diff --git a/src/Primitives.Shared/Advanced/ExpireCoordinator{T}.cs b/src/Primitives.Shared/Advanced/ExpireCoordinator{T}.cs
index 2aba4628..f96303f0 100644
--- a/src/Primitives.Shared/Advanced/ExpireCoordinator{T}.cs
+++ b/src/Primitives.Shared/Advanced/ExpireCoordinator{T}.cs
@@ -22,7 +22,7 @@ public sealed class ExpireCoordinator : IObserver, IDisposable
/// The timeout period.
private readonly TimeSpan _dueTime;
- /// The sequencer used to schedule the timeout.
+ /// The sequencer that schedules the timeout.
private readonly ISequencer _sequencer;
/// The downstream observer.
@@ -32,35 +32,33 @@ public sealed class ExpireCoordinator : IObserver, IDisposable
[System.Diagnostics.CodeAnalysis.SuppressMessage(
"Usage",
"CA2213:Disposable fields should be disposed",
- Justification =
- "Disposed via the thread-safe Interlocked.Exchange teardown in Dispose; CA2213 does not recognize disposal of a field through Interlocked.Exchange.")]
+ Justification = "Disposed through Interlocked.Exchange in Dispose.")]
private IDisposable? _subscription;
/// The active timeout timer.
[System.Diagnostics.CodeAnalysis.SuppressMessage(
"Usage",
"CA2213:Disposable fields should be disposed",
- Justification =
- "Disposed via the thread-safe Interlocked.Exchange teardown in Dispose; CA2213 does not recognize disposal of a field through Interlocked.Exchange.")]
+ Justification = "Disposed through Interlocked.Exchange in Dispose.")]
private IDisposable? _timer;
/// A value indicating whether the timeout or source has terminated.
private int _done;
- /// Monotonic version used to suppress timeouts superseded by a newer value.
+ /// Monotonic version that suppresses timeouts superseded by a newer value.
private long _epoch;
///
- /// The instant, on the sequencer's own clock, at which the current inactivity window closes. Read and written
- /// under . Starts at so a window that has not been
- /// published yet can never expire a value.
+ /// The instant on the sequencer's clock at which the current inactivity window closes, read and written under
+ /// . It starts at so an unpublished window never
+ /// expires a value.
///
private DateTimeOffset _deadline = DateTimeOffset.MaxValue;
/// Initializes a new instance of the class.
/// The source observable.
/// The timeout period.
- /// The sequencer used to schedule the timeout.
+ /// The sequencer that schedules the timeout.
/// The downstream observer.
public ExpireCoordinator(IObservable source, TimeSpan dueTime, ISequencer sequencer, IObserver observer)
{
@@ -140,11 +138,9 @@ public void OnError(Exception error)
///
///
- /// A value is on time only when it arrives before the current inactivity window closes, which is a question for
- /// the sequencer's clock — not for whether the armed timer has run yet. The timer is dispatched by the sequencer,
- /// and a thread-pool sequencer whose pool is saturated can dispatch it arbitrarily late while a source on another
- /// thread keeps producing. Forwarding a value in that gap would deliver a value the operator has already promised
- /// to time out, so a value that arrives after its deadline expires the sequence here instead.
+ /// A value is on time only when it arrives before the inactivity window closes on the sequencer's clock, not
+ /// merely before the armed timer has run. A value that arrives past its deadline — which a saturated thread-pool
+ /// sequencer can allow — terminates the sequence with instead of being forwarded.
///
public void OnNext(T value)
{
@@ -204,10 +200,10 @@ public ExpireCoordinator Run()
}
/// Schedules a fresh inactivity timer for the given epoch and discards the in-flight one.
- /// The version this timer must still match to fire.
- /// Scheduled outside the gate to avoid reentrant acquisition on a synchronous
- /// sequencer; the publish is re-checked under the gate so a timer never survives a terminal notification, and
- /// so a superseded arm cannot publish its older deadline and timer over a newer value's.
+ /// The version this timer must match to fire.
+ /// Scheduling happens outside the gate so a synchronous sequencer cannot re-enter .
+ /// The publish is re-checked under the gate, so neither a terminal notification nor a newer value's window can be
+ /// overwritten by a superseded arm.
private void ArmTimer(long epoch)
{
var deadline = Deadline();
@@ -232,13 +228,11 @@ private void ArmTimer(long epoch)
previous?.Dispose();
}
- /// Computes the instant the inactivity window opened now would close at.
+ /// Computes the closing instant of an inactivity window opened at the current time.
/// The deadline on the sequencer's clock, saturated instead of overflowing.
///
- /// A due time that normalizes to zero is scheduled as immediate work rather than timed work, so it has no clock
- /// window: the timeout is ordered by the sequencer's queue, and a synchronous value that arrives before the queue
- /// drains still wins. That branch mirrors the scheduling extension's own zero-due-time path, keeping the deadline
- /// in lockstep with how the timer was actually scheduled. Only a positive due time opens a window on the clock.
+ /// A due time that normalizes to zero is queued as immediate work, so it opens no clock window and a synchronous
+ /// value arriving before the queue drains wins; only a positive due time yields a real deadline.
///
private DateTimeOffset Deadline()
{
@@ -252,7 +246,7 @@ private DateTimeOffset Deadline()
return DateTimeOffset.MaxValue - now <= dueTime ? DateTimeOffset.MaxValue : now + dueTime;
}
- /// Emits the timeout error when the firing timer is still current.
+ /// Emits the timeout error when the firing timer is the current one.
/// The version captured when the firing timer was armed.
/// An empty disposable.
private EmptyDisposable EmitTimeout(long epoch)
diff --git a/src/Primitives.Shared/Advanced/ExpireSignal{T}.cs b/src/Primitives.Shared/Advanced/ExpireSignal{T}.cs
index 9011b5c1..a4c31459 100644
--- a/src/Primitives.Shared/Advanced/ExpireSignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/ExpireSignal{T}.cs
@@ -8,7 +8,7 @@ namespace ReactiveUI.Primitives.Reactive.Advanced;
namespace ReactiveUI.Primitives.Advanced;
#endif
-/// Timeout signal with a direct subscription path.
+/// Terminates with a when the source stays quiet longer than the timeout period.
/// The source value type.
[System.Diagnostics.DebuggerDisplay("ExpireSignal: DueTime = {_dueTime}, Source = {_source}")]
public sealed class ExpireSignal : IRequireCurrentThread
@@ -19,13 +19,13 @@ public sealed class ExpireSignal : IRequireCurrentThread
/// The timeout period.
private readonly TimeSpan _dueTime;
- /// The sequencer used to schedule the timeout.
+ /// The sequencer that schedules the timeout.
private readonly ISequencer _sequencer;
/// Initializes a new instance of the class.
/// The source observable.
/// The timeout period.
- /// The sequencer used to schedule the timeout.
+ /// The sequencer that schedules the timeout.
public ExpireSignal(IObservable source, TimeSpan dueTime, ISequencer sequencer)
{
ArgumentExceptionHelper.ThrowIfNull(source);
diff --git a/src/Primitives.Shared/Advanced/FinallySignal{T}.cs b/src/Primitives.Shared/Advanced/FinallySignal{T}.cs
index 8e739889..7a339c83 100644
--- a/src/Primitives.Shared/Advanced/FinallySignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/FinallySignal{T}.cs
@@ -10,58 +10,58 @@ namespace ReactiveUI.Primitives.Reactive.Advanced;
namespace ReactiveUI.Primitives.Advanced;
#endif
-/// Represents the FinallySignal class.
-/// The T type.
-/// The source value.
-/// The finallyAction value.
+/// Runs an action once the subscription ends, whether it terminated or was disposed.
+/// The value type.
+/// The source observable.
+/// The action run when the subscription ends.
[System.Diagnostics.DebuggerDisplay("FinallySignal: Source = {_source}")]
public sealed class FinallySignal(IObservable source, Action finallyAction) : IRequireCurrentThread
{
- /// Stores state for the signal implementation.
+ /// The source observable.
private readonly IObservable _source = source;
- /// Stores state for the signal implementation.
+ /// The action run when the subscription ends.
private readonly Action _finallyAction = finallyAction;
- /// Executes the IsRequiredSubscribeOnCurrentThread operation.
- /// The result.
+ /// Reports that subscription runs on the calling thread.
+ /// Always .
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool IsRequiredSubscribeOnCurrentThread() => true;
- /// Executes the Subscribe operation.
- /// The observer value.
- /// The result.
+ /// Subscribes the observer and attaches the end-of-subscription action.
+ /// The downstream observer.
+ /// The disposable that releases the subscription and runs the action.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public IDisposable Subscribe(IObserver observer) =>
SignalSubscription.Subscribe(observer, true, SubscribeCore);
- /// Executes the SubscribeCore operation.
- /// The observer value.
- /// The cancel value.
- /// The result.
+ /// Creates the handler that forwards notifications and owns the end-of-subscription action.
+ /// The downstream observer.
+ /// The outer subscription handle.
+ /// The disposable that releases the subscription and runs the action.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private IDisposable SubscribeCore(IObserver observer, IDisposable cancel) =>
new Finally(this, observer, cancel).Run();
- /// Represents the Finally class.
+ /// Forwards notifications downstream and pairs the subscription with the end-of-subscription action.
private sealed class Finally : IObserver, IDisposable
{
- /// Stores state for the signal implementation.
+ /// The signal supplying the source and the action.
private readonly FinallySignal _parent;
- /// Stores the downstream observer.
+ /// The downstream observer.
private readonly IObserver _observer;
- /// Stores the upstream subscription.
+ /// The outer subscription handle released on teardown.
private IDisposable? _cancel;
/// Disposed latch; 0 when alive, 1 once disposed.
private int _disposed;
/// Initializes a new instance of the class.
- /// The parent value.
- /// The observer value.
- /// The cancel value.
+ /// The signal supplying the source and the action.
+ /// The downstream observer.
+ /// The outer subscription handle.
/// is .
public Finally(FinallySignal parent, IObserver observer, IDisposable cancel)
{
@@ -70,8 +70,8 @@ public Finally(FinallySignal parent, IObserver observer, IDisposable cance
_parent = parent;
}
- /// Executes the Run operation.
- /// The result.
+ /// Subscribes to the source, running the action immediately if subscription throws.
+ /// The disposable that releases the source subscription and then runs the action.
public MultipleDisposable Run()
{
IDisposable subscription;
@@ -88,13 +88,13 @@ public MultipleDisposable Run()
return new(subscription, new ActionDisposable(() => _parent._finallyAction()));
}
- /// Executes the OnNext operation.
- /// The value.
+ /// Forwards a value downstream.
+ /// The value to forward.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void OnNext(T value) => _observer.OnNext(value);
- /// Executes the OnError operation.
- /// The error value.
+ /// Forwards the error downstream and releases the subscription.
+ /// The error to forward.
public void OnError(Exception error)
{
try
@@ -107,7 +107,7 @@ public void OnError(Exception error)
}
}
- /// Executes the OnCompleted operation.
+ /// Completes downstream and releases the subscription.
public void OnCompleted()
{
try
@@ -120,7 +120,7 @@ public void OnCompleted()
}
}
- /// Executes the Dispose operation.
+ /// Releases the outer subscription handle once.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void Dispose() => WitnessTeardown.Dispose(ref _disposed, ref _cancel);
}
diff --git a/src/Primitives.Shared/Advanced/FromAsyncSubscription{T}.cs b/src/Primitives.Shared/Advanced/FromAsyncSubscription{T}.cs
index 615fd62e..8bf1e500 100644
--- a/src/Primitives.Shared/Advanced/FromAsyncSubscription{T}.cs
+++ b/src/Primitives.Shared/Advanced/FromAsyncSubscription{T}.cs
@@ -110,7 +110,7 @@ internal IDisposable Start()
return this;
}
- /// Forwards a task that has already reached a terminal state.
+ /// Forwards a task that has reached a terminal state.
/// The task to observe.
/// The downstream observer.
/// The subscription lifetime.
@@ -120,7 +120,7 @@ internal IDisposable Start()
[System.Diagnostics.CodeAnalysis.SuppressMessage(
"Concurrency",
"PSH1315:A blocking wait on an awaitable that may not be done",
- Justification = "Synchronous read is limited to the already-completed task fast path.")]
+ Justification = "Synchronous read is limited to a task in a terminal state.")]
private static bool TryCompleteSynchronously(
Task task,
IObserver observer,
@@ -145,7 +145,7 @@ private static bool TryCompleteSynchronously(
: task.IsFaulted && FaultSynchronously(task, observer, lifetime, externalCancellation, linkedSource);
}
- /// Forwards an already-successful task result.
+ /// Forwards a successful task result.
/// The task result.
/// The downstream observer.
/// The subscription lifetime.
@@ -171,7 +171,7 @@ private static bool CompleteSynchronously(
return true;
}
- /// Forwards an already-canceled task result.
+ /// Forwards a canceled task result.
/// The task to observe.
/// The downstream observer.
/// The subscription lifetime.
@@ -202,7 +202,7 @@ private static bool CancelSynchronously(
return true;
}
- /// Forwards an already-faulted task result.
+ /// Forwards a faulted task result.
/// The task to observe.
/// The downstream observer.
/// The subscription lifetime.
diff --git a/src/Primitives.Shared/Advanced/FromEventConversionSignal{TEventHandler,TCallback,TResult}.cs b/src/Primitives.Shared/Advanced/FromEventConversionSignal{TEventHandler,TCallback,TResult}.cs
index 7f78774d..fb04d930 100644
--- a/src/Primitives.Shared/Advanced/FromEventConversionSignal{TEventHandler,TCallback,TResult}.cs
+++ b/src/Primitives.Shared/Advanced/FromEventConversionSignal{TEventHandler,TCallback,TResult}.cs
@@ -9,9 +9,9 @@ namespace ReactiveUI.Primitives.Advanced;
#endif
///
-/// Bridges an event whose handler is built by a caller-supplied conversion, so the handler can do
-/// synchronous work inside the event before the notification is published downstream. Supplying the
-/// conversion removes the need to derive a delegate reflectively, keeping the bridge trim- and AOT-safe.
+/// Bridges an event whose handler is built by a caller-supplied conversion, letting the handler do synchronous work
+/// inside the event before the notification is published downstream. The conversion replaces reflective delegate
+/// creation, so the bridge is trim- and AOT-safe.
///
/// The delegate type used by the event.
/// The notification callback type handed to the conversion.
diff --git a/src/Primitives.Shared/Advanced/GuardedWitness{T}.cs b/src/Primitives.Shared/Advanced/GuardedWitness{T}.cs
index 80542e27..0f283e61 100644
--- a/src/Primitives.Shared/Advanced/GuardedWitness{T}.cs
+++ b/src/Primitives.Shared/Advanced/GuardedWitness{T}.cs
@@ -12,17 +12,16 @@ namespace ReactiveUI.Primitives.Advanced;
///
/// Forwarding observer that releases its upstream cancel resource when the sequence terminates or a downstream
-/// OnNext throws. The shared guard behind the scheduled factory signals (Empty, Return, Throw, Defer),
-/// usable by any signal implementation that needs terminate-and-release semantics around a downstream observer.
+/// OnNext throws, rethrowing the latter after release.
///
/// The value type.
[System.Diagnostics.DebuggerDisplay("GuardedWitness: Disposed = {_disposed}, Observer = {_observer}")]
public sealed class GuardedWitness : IObserver, IDisposable
{
- /// Stores the downstream observer.
+ /// The downstream observer.
private readonly IObserver _observer;
- /// Stores the upstream subscription.
+ /// The upstream cancel resource released on termination.
private IDisposable? _cancel;
/// Disposed latch; 0 when alive, 1 once disposed.
diff --git a/src/Primitives.Shared/Advanced/ImmutableReturnRxVoidSignal.cs b/src/Primitives.Shared/Advanced/ImmutableReturnRxVoidSignal.cs
index fc74becb..2f08e582 100644
--- a/src/Primitives.Shared/Advanced/ImmutableReturnRxVoidSignal.cs
+++ b/src/Primitives.Shared/Advanced/ImmutableReturnRxVoidSignal.cs
@@ -10,7 +10,7 @@ namespace ReactiveUI.Primitives.Reactive.Advanced;
namespace ReactiveUI.Primitives.Advanced;
#endif
-/// An allocation-free observable that emits a single value and then completes; the concrete singleton backing for the RxVoid emit path.
+/// Emits inline and completes, without allocating a subscription.
public sealed class ImmutableReturnRxVoidSignal : IRequireCurrentThread, IInlineSignal
{
/// The shared singleton instance.
@@ -21,14 +21,14 @@ private ImmutableReturnRxVoidSignal()
{
}
- /// Executes the IsRequiredSubscribeOnCurrentThread operation.
- /// The result.
+ /// Reports that subscription needs no current-thread dispatch.
+ /// Always .
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool IsRequiredSubscribeOnCurrentThread() => false;
- /// Executes the Subscribe operation.
- /// The observer value.
- /// The result.
+ /// Emits and completes before returning.
+ /// The downstream observer.
+ /// An empty disposable; the sequence has ended and there is nothing to cancel.
public IDisposable Subscribe(IObserver observer)
{
observer.OnNext(RxVoid.Default);
@@ -36,11 +36,11 @@ public IDisposable Subscribe(IObserver observer)
return EmptyDisposable.Instance;
}
- /// Executes the Subscribe operation.
- /// The onNext value.
- /// The onError value.
- /// The onCompleted value.
- /// The result.
+ /// Invokes with and then .
+ /// The value callback.
+ /// The error callback, which is never invoked.
+ /// The completion callback.
+ /// An empty disposable; the sequence has ended and there is nothing to cancel.
public IDisposable Subscribe(Action onNext, Action onError, Action onCompleted)
{
onNext(RxVoid.Default);
diff --git a/src/Primitives.Shared/Advanced/IsEmptySignal{T}.cs b/src/Primitives.Shared/Advanced/IsEmptySignal{T}.cs
index 27fe3ebe..4ebd7d17 100644
--- a/src/Primitives.Shared/Advanced/IsEmptySignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/IsEmptySignal{T}.cs
@@ -48,11 +48,10 @@ public IDisposable Subscribe(IObserver observer)
return EmptyDisposable.Instance;
}
- // The first value settles this operator, so it must be able to dispose the source as soon as one arrives.
- // A current-thread source runs its work on the trampoline of whichever call enters it first; if that call is
- // the source's own Subscribe, the source drains the trampoline before the sink is handed the subscription,
- // and an endless source therefore never stops. Entering the trampoline here means the source only queues its
- // first tick and returns, so the sink owns the subscription before that tick is delivered.
+ // The first value settles this operator, so the sink must own the subscription before any value arrives. A
+ // current-thread source drains the trampoline inside whichever call enters it first, so letting the source's
+ // own Subscribe enter it would deliver values before the sink holds the subscription, and an endless source
+ // would never stop. Entering the trampoline here leaves the source only able to queue its first tick.
if (!IsRequiredSubscribeOnCurrentThread() || !CurrentThreadSequencer.IsScheduleRequired)
{
return SubscribeCore(observer);
diff --git a/src/Primitives.Shared/Advanced/IsEmptyWitness{T}.cs b/src/Primitives.Shared/Advanced/IsEmptyWitness{T}.cs
index e844fb89..640df4ca 100644
--- a/src/Primitives.Shared/Advanced/IsEmptyWitness{T}.cs
+++ b/src/Primitives.Shared/Advanced/IsEmptyWitness{T}.cs
@@ -15,7 +15,7 @@ namespace ReactiveUI.Primitives.Advanced;
[System.Diagnostics.DebuggerDisplay("IsEmptyWitness: Stopped = {_stopped}, Subscription = {Subscription}")]
public sealed class IsEmptyWitness : IObserver, IDisposable
{
- /// Stores the stopped flag for interlocked/ref helper calls.
+ /// Non-zero once the result has been emitted or the sink disposed.
private int _stopped;
/// Initializes a new instance of the class.
diff --git a/src/Primitives.Shared/Advanced/LoopSignal{T}.cs b/src/Primitives.Shared/Advanced/LoopSignal{T}.cs
index 23c50a2d..4fadc5f8 100644
--- a/src/Primitives.Shared/Advanced/LoopSignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/LoopSignal{T}.cs
@@ -10,7 +10,7 @@ namespace ReactiveUI.Primitives.Reactive.Advanced;
namespace ReactiveUI.Primitives.Advanced;
#endif
-/// Represents an infinite repetition signal.
+/// Emits the same value endlessly until the subscription is disposed.
/// The value type.
[System.Diagnostics.DebuggerDisplay("LoopSignal: Value = {Value}")]
public sealed class LoopSignal : IRequireCurrentThread
@@ -29,14 +29,14 @@ public LoopSignal(T value)
/// Gets the repeated value.
internal T Value { get; }
- /// Executes the IsRequiredSubscribeOnCurrentThread operation.
- /// The result.
+ /// Reports that subscription runs on the current-thread sequencer.
+ /// Always .
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool IsRequiredSubscribeOnCurrentThread() => _currentThreadRequired;
- /// Executes the Subscribe operation.
- /// The observer value.
- /// The result.
+ /// Subscribes the observer and starts the endless emission loop.
+ /// The downstream observer.
+ /// The disposable that stops the loop.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public IDisposable Subscribe(IObserver observer) =>
SignalSubscription.Subscribe(observer, _currentThreadRequired, SubscribeCore);
@@ -47,16 +47,15 @@ public IDisposable Subscribe(IObserver observer) =>
private static bool IsDisposed(IDisposable disposable) =>
disposable is IsDisposed state && state.IsDisposed;
- /// Executes the SubscribeCore operation.
- /// The observer value.
- /// The cancel value.
- /// The result.
+ /// Schedules the recursive emission loop, which stops as soon as the handle is disposed.
+ /// The downstream observer.
+ /// The outer subscription handle.
+ /// The disposable that cancels the scheduled loop.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
[System.Diagnostics.CodeAnalysis.SuppressMessage(
"Allocations",
"PSH1011",
- Justification =
- "No state-taking recursive Schedule overload exists; needs new scheduling API or a rule exclusion.")]
+ Justification = "No state-taking recursive Schedule overload exists.")]
private IDisposable SubscribeCore(IObserver observer, IDisposable cancel) =>
Sequencer.CurrentThread.Schedule(self =>
{
diff --git a/src/Primitives.Shared/Advanced/MapIndexedSignal{TSource,TResult}.cs b/src/Primitives.Shared/Advanced/MapIndexedSignal{TSource,TResult}.cs
index 1b2362dd..182c2070 100644
--- a/src/Primitives.Shared/Advanced/MapIndexedSignal{TSource,TResult}.cs
+++ b/src/Primitives.Shared/Advanced/MapIndexedSignal{TSource,TResult}.cs
@@ -10,7 +10,7 @@ namespace ReactiveUI.Primitives.Reactive.Advanced;
namespace ReactiveUI.Primitives.Advanced;
#endif
-/// Indexed map signal.
+/// Projects each source value together with its zero-based position in the sequence.
/// The source value type.
/// The projected value type.
/// The source observable.
diff --git a/src/Primitives.Shared/Advanced/MergeCoordinator{T}.cs b/src/Primitives.Shared/Advanced/MergeCoordinator{T}.cs
index 407d1e5a..2b8b016b 100644
--- a/src/Primitives.Shared/Advanced/MergeCoordinator{T}.cs
+++ b/src/Primitives.Shared/Advanced/MergeCoordinator{T}.cs
@@ -85,8 +85,8 @@ public void OnSource(IObservable? source)
}
// A source is free to signal completion more than once. Latch per inner so a repeat cannot decrement
- // the active count a second time - that would drop the count on behalf of a sibling that is still
- // running and complete the merge early, losing everything the sibling had left to produce.
+ // the active count on behalf of a running sibling, which would complete the merge early and drop
+ // everything that sibling had left to produce.
var completed = 0;
Subscriptions.Add(source.Subscribe(
OnInnerNext,
diff --git a/src/Primitives.Shared/Advanced/MergeSignal{T}.cs b/src/Primitives.Shared/Advanced/MergeSignal{T}.cs
index b2c94a23..27c3eb03 100644
--- a/src/Primitives.Shared/Advanced/MergeSignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/MergeSignal{T}.cs
@@ -8,7 +8,7 @@ namespace ReactiveUI.Primitives.Reactive.Advanced;
namespace ReactiveUI.Primitives.Advanced;
#endif
-/// Concrete signal for Rx-named merge overloads.
+/// Merges concurrent sources into one sequence, optionally bounding the active subscription count.
/// The value type.
[System.Diagnostics.DebuggerDisplay("MergeSignal: Sources = {Sources}, MaxConcurrent = {MaxConcurrent}")]
public sealed class MergeSignal : IObservable
diff --git a/src/Primitives.Shared/Advanced/RecoverSignal{T,TException}.cs b/src/Primitives.Shared/Advanced/RecoverSignal{T,TException}.cs
index 8d5f6f3f..52b55198 100644
--- a/src/Primitives.Shared/Advanced/RecoverSignal{T,TException}.cs
+++ b/src/Primitives.Shared/Advanced/RecoverSignal{T,TException}.cs
@@ -11,9 +11,9 @@ namespace ReactiveUI.Primitives.Advanced;
#endif
///
-/// Dedicated cold signal for Recover/Resume (catch a typed error and switch to a
-/// handler-selected sequence). Replaces the witness-framework subject with a lightweight sink that
-/// holds its source and fallback subscriptions in two interlocked slots, with no composite disposable.
+/// Cold signal that mirrors the source and, on an error of type , switches to
+/// the sequence the handler selects for it. Any other error, and an error thrown by the handler itself, is
+/// forwarded downstream unchanged.
///
/// The value type.
/// The handled exception type.
diff --git a/src/Primitives.Shared/Advanced/RepeatSourceCoordinator{T}.cs b/src/Primitives.Shared/Advanced/RepeatSourceCoordinator{T}.cs
index a52696e0..07353aed 100644
--- a/src/Primitives.Shared/Advanced/RepeatSourceCoordinator{T}.cs
+++ b/src/Primitives.Shared/Advanced/RepeatSourceCoordinator{T}.cs
@@ -25,7 +25,7 @@ public sealed class RepeatSourceCoordinator : IDisposable
/// The active source subscription or queued resubscription.
private readonly SingleReplaceableDisposable _active = new();
- /// Guards synchronous completion while a subscription is still being assigned.
+ /// Guards completion raised synchronously while a subscription is being assigned.
private readonly Lock _gate = new();
/// The remaining number of finite subscriptions.
diff --git a/src/Primitives.Shared/Advanced/RepeatSourceWitness{T}.cs b/src/Primitives.Shared/Advanced/RepeatSourceWitness{T}.cs
index 3e0f1725..2725a093 100644
--- a/src/Primitives.Shared/Advanced/RepeatSourceWitness{T}.cs
+++ b/src/Primitives.Shared/Advanced/RepeatSourceWitness{T}.cs
@@ -19,7 +19,7 @@ public sealed class RepeatSourceWitness : IObserver
/// The source subscription generation.
private readonly int _generation;
- /// Tracks whether this source subscription has already terminated.
+ /// Latches to 1 once this source subscription has terminated.
private int _terminated;
/// Initializes a new instance of the class.
diff --git a/src/Primitives.Shared/Advanced/ResumeSignal{T}.cs b/src/Primitives.Shared/Advanced/ResumeSignal{T}.cs
index d3e9f7d9..3ce907c8 100644
--- a/src/Primitives.Shared/Advanced/ResumeSignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/ResumeSignal{T}.cs
@@ -11,9 +11,8 @@ namespace ReactiveUI.Primitives.Advanced;
#endif
///
-/// Dedicated cold signal for Resume (continue with a fixed fallback sequence after any error). Holds the
-/// fallback observable directly so no per-subscription closure is allocated, mirroring the slot-based subscription
-/// management of .
+/// Cold signal that mirrors the source and, when it produces any error, subscribes the fallback sequence in its
+/// place. The error is not forwarded downstream.
///
/// The value type.
/// The source observable.
diff --git a/src/Primitives.Shared/Advanced/ReturnSignal{T}.cs b/src/Primitives.Shared/Advanced/ReturnSignal{T}.cs
index 13658ac5..5ea4ccb1 100644
--- a/src/Primitives.Shared/Advanced/ReturnSignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/ReturnSignal{T}.cs
@@ -10,23 +10,23 @@ namespace ReactiveUI.Primitives.Reactive.Advanced;
namespace ReactiveUI.Primitives.Advanced;
#endif
-/// An observable that emits a single value then completes on the supplied scheduler; the concrete backing for the scheduled emit path.
+/// Emits a single value and then completes, on the supplied sequencer.
/// The emitted value type.
[System.Diagnostics.DebuggerDisplay("ReturnSignal: Value = {_value}, Scheduler = {_scheduler}")]
public sealed class ReturnSignal : IRequireCurrentThread
{
- /// Stores state for the signal implementation.
+ /// The value to emit.
private readonly T _value;
- /// Stores state for the signal implementation.
+ /// The sequencer that emits the value and completion.
private readonly ISequencer _scheduler;
- /// Stores state for the signal implementation.
+ /// Whether subscription must be dispatched through the current-thread sequencer.
private readonly bool _currentThreadRequired;
/// Initializes a new instance of the class.
- /// The value.
- /// The scheduler value.
+ /// The value to emit.
+ /// The sequencer that emits the value and completion.
public ReturnSignal(T value, ISequencer scheduler)
{
_value = value;
@@ -34,22 +34,22 @@ public ReturnSignal(T value, ISequencer scheduler)
_currentThreadRequired = scheduler == Sequencer.CurrentThread;
}
- /// Executes the IsRequiredSubscribeOnCurrentThread operation.
- /// The result.
+ /// Gets whether subscription has to be dispatched through the current-thread sequencer.
+ /// when the supplied sequencer is the current-thread sequencer.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool IsRequiredSubscribeOnCurrentThread() => _currentThreadRequired;
- /// Executes the Subscribe operation.
- /// The observer value.
- /// The result.
+ /// Subscribes an observer that receives the value followed by completion.
+ /// The downstream observer.
+ /// A disposable that cancels the emission when it has not run yet.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public IDisposable Subscribe(IObserver observer) =>
SignalSubscription.Subscribe(observer, _currentThreadRequired, SubscribeCore);
- /// Executes the SubscribeCore operation.
- /// The observer value.
- /// The cancel value.
- /// The result.
+ /// Emits the value and completion inline for the immediate sequencer, otherwise on the sequencer.
+ /// The downstream observer.
+ /// The subscription handle the guard checks before forwarding.
+ /// The disposable that cancels the scheduled emission.
private IDisposable SubscribeCore(IObserver observer, IDisposable cancel)
{
observer = new GuardedWitness(observer, cancel);
diff --git a/src/Primitives.Shared/Advanced/ScheduledEnumerableSignal{T}.cs b/src/Primitives.Shared/Advanced/ScheduledEnumerableSignal{T}.cs
index 94fc4418..c13b942d 100644
--- a/src/Primitives.Shared/Advanced/ScheduledEnumerableSignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/ScheduledEnumerableSignal{T}.cs
@@ -8,14 +8,14 @@ namespace ReactiveUI.Primitives.Reactive.Advanced;
namespace ReactiveUI.Primitives.Advanced;
#endif
-/// Scheduled enumerable-backed signal used by observable conversion overloads.
+/// Emits the values of an enumerable on the supplied sequencer, then completes.
/// The value type.
[System.Diagnostics.DebuggerDisplay("ScheduledEnumerableSignal: Values = {Values}, Scheduler = {Scheduler}")]
public sealed class ScheduledEnumerableSignal : IObservable
{
/// Initializes a new instance of the class.
/// The values to emit.
- /// The scheduler used to enumerate and emit the values.
+ /// The sequencer that enumerates and emits the values.
public ScheduledEnumerableSignal(IEnumerable values, ISequencer scheduler)
{
ArgumentExceptionHelper.ThrowIfNull(values);
@@ -29,7 +29,7 @@ public ScheduledEnumerableSignal(IEnumerable values, ISequencer scheduler)
/// Gets the values to emit.
private IEnumerable Values { get; }
- /// Gets the scheduler used to enumerate and emit the values.
+ /// Gets the sequencer that enumerates and emits the values.
private ISequencer Scheduler { get; }
///
diff --git a/src/Primitives.Shared/Advanced/SequenceSignal.cs b/src/Primitives.Shared/Advanced/SequenceSignal.cs
index f75ef73c..899143a4 100644
--- a/src/Primitives.Shared/Advanced/SequenceSignal.cs
+++ b/src/Primitives.Shared/Advanced/SequenceSignal.cs
@@ -15,7 +15,7 @@ public sealed class SequenceSignal : IRequireCurrentThread
/// Initializes a new instance of the class.
/// The first value to emit.
/// The number of values to emit.
- /// The scheduler used to emit values.
+ /// The sequencer that emits the values.
public SequenceSignal(int start, int count, ISequencer scheduler)
{
Start = start;
@@ -29,7 +29,7 @@ public SequenceSignal(int start, int count, ISequencer scheduler)
/// Gets the number of values to emit.
private int Count { get; }
- /// Gets the scheduler used to emit values.
+ /// Gets the sequencer that emits the values.
private ISequencer Scheduler { get; }
///
diff --git a/src/Primitives.Shared/Advanced/SparkWitness.cs b/src/Primitives.Shared/Advanced/SparkWitness.cs
index dfff98cd..dbbb1c34 100644
--- a/src/Primitives.Shared/Advanced/SparkWitness.cs
+++ b/src/Primitives.Shared/Advanced/SparkWitness.cs
@@ -37,7 +37,7 @@ public sealed class SparkWitness : IObserver, IDisposable
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void OnCompleted() => SinkTerminal.Complete(_observer, Spark.CreateOnCompleted(), this);
- /// Assigns the upstream subscription, disposing it if one is already held.
+ /// Assigns the upstream subscription, disposing the incoming one when the slot is occupied.
/// The upstream subscription.
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void SetSubscription(IDisposable subscription) => SinkSubscription.Set(ref _subscription, subscription);
diff --git a/src/Primitives.Shared/Advanced/StartSignal.cs b/src/Primitives.Shared/Advanced/StartSignal.cs
index 5a70166b..a6d89a1d 100644
--- a/src/Primitives.Shared/Advanced/StartSignal.cs
+++ b/src/Primitives.Shared/Advanced/StartSignal.cs
@@ -14,7 +14,7 @@ public sealed class StartSignal : IRequireCurrentThread
{
/// Initializes a new instance of the class.
/// The action to run.
- /// The scheduler used to run the action.
+ /// The sequencer that runs the action.
public StartSignal(Action action, ISequencer scheduler)
{
Action = action;
@@ -24,7 +24,7 @@ public StartSignal(Action action, ISequencer scheduler)
/// Gets the action to run.
private Action Action { get; }
- /// Gets the scheduler used to run the action.
+ /// Gets the sequencer that runs the action.
private ISequencer Scheduler { get; }
///
diff --git a/src/Primitives.Shared/Advanced/StartSignal{T}.cs b/src/Primitives.Shared/Advanced/StartSignal{T}.cs
index 3280e555..e1f4141d 100644
--- a/src/Primitives.Shared/Advanced/StartSignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/StartSignal{T}.cs
@@ -15,7 +15,7 @@ public sealed class StartSignal : IRequireCurrentThread
{
/// Initializes a new instance of the class.
/// The function to run.
- /// The scheduler used to run the function.
+ /// The sequencer that runs the function.
public StartSignal(Func function, ISequencer scheduler)
{
Function = function;
@@ -25,7 +25,7 @@ public StartSignal(Func function, ISequencer scheduler)
/// Gets the function to run.
private Func Function { get; }
- /// Gets the scheduler used to run the function.
+ /// Gets the sequencer that runs the function.
private ISequencer Scheduler { get; }
///
diff --git a/src/Primitives.Shared/Advanced/SubscriptionScheduling.cs b/src/Primitives.Shared/Advanced/SubscriptionScheduling.cs
index 19f4d726..5e70b6d2 100644
--- a/src/Primitives.Shared/Advanced/SubscriptionScheduling.cs
+++ b/src/Primitives.Shared/Advanced/SubscriptionScheduling.cs
@@ -9,16 +9,13 @@ namespace ReactiveUI.Primitives.Advanced;
#endif
///
-/// Shared subscription-time scheduling for advanced signals. A cold signal that must observe the
-/// current-thread sequencer cannot simply subscribe inline: when the current-thread sequencer is already
-/// draining work, the subscription has to be queued behind it so the emissions arrive in trampoline order.
-/// The two shapes below capture that dance once so each signal keeps only its own Run body.
+/// Subscription-time scheduling shared by advanced signals: a signal that observes the current-thread
+/// sequencer has to queue behind it while it is draining work, so emissions arrive in trampoline order.
///
internal static class SubscriptionScheduling
{
///
- /// Subscribes immediately when the current-thread sequencer is idle, otherwise queues the subscription
- /// on it and hands back a placeholder that is filled in once the subscription actually runs.
+ /// Subscribes inline when the current-thread sequencer is idle, otherwise queues the subscription on it.
///
/// The type of the state passed to the subscribe callback.
/// The state carried to .
@@ -44,13 +41,13 @@ internal static IDisposable OnCurrentThread(TState state, Func
/// Runs on : inline for the immediate sequencer, and
- /// queued behind the current-thread sequencer when that one is the target and is already draining work.
+ /// queued behind the current-thread sequencer when that one is the target and is draining work.
///
/// The type of the state passed to the scheduled work.
/// The sequencer that runs the work.
/// The state carried to .
/// The work to run, which emits to the downstream observer.
- /// The disposable used to cancel the scheduled work (best effort).
+ /// A disposable that cancels the scheduled work, best effort.
internal static IDisposable RunOn(ISequencer sequencer, TState state, Action run)
{
if (sequencer == Sequencer.Immediate)
diff --git a/src/Primitives.Shared/Advanced/SwitchMapSignal{TSource,TResult}.cs b/src/Primitives.Shared/Advanced/SwitchMapSignal{TSource,TResult}.cs
index c9a4101c..27a9b1aa 100644
--- a/src/Primitives.Shared/Advanced/SwitchMapSignal{TSource,TResult}.cs
+++ b/src/Primitives.Shared/Advanced/SwitchMapSignal{TSource,TResult}.cs
@@ -14,8 +14,8 @@ namespace ReactiveUI.Primitives.Advanced;
/// The source element type.
/// The element type of the projected inner observables.
///
-/// Fuses the projection into the switch: one object and one observer hop, where a projection followed by a
-/// separate switch costs two of each and an intermediate sequence of observables.
+/// The projection is fused into the switch, so it costs one object and one observer hop instead of the two of
+/// each, plus an intermediate sequence of observables, that a separate projection and switch cost.
///
[System.Diagnostics.DebuggerDisplay("SwitchMapSignal: Source = {_source}, SkipNullSources = {_skipNullSources}")]
public sealed class SwitchMapSignal : IObservable
diff --git a/src/Primitives.Shared/Advanced/SwitchWitness{T}.cs b/src/Primitives.Shared/Advanced/SwitchWitness{T}.cs
index c10b1e46..d4c5b23f 100644
--- a/src/Primitives.Shared/Advanced/SwitchWitness{T}.cs
+++ b/src/Primitives.Shared/Advanced/SwitchWitness{T}.cs
@@ -80,12 +80,10 @@ private void OnSource(IObservable source)
error => OnError(current, error),
() => OnCompleted(current));
- // Subscribing can push a value downstream synchronously, and a downstream handler is free to feed the
- // outer source again. That re-enters OnSource, installs a newer generation, and only then returns here.
- // Installing unconditionally at that point would replace the newer subscription with this stale one and
- // dispose it, leaving a subscription whose notifications are all filtered out by version - the sequence
- // would then never produce another value nor complete. Only the generation that is still current may
- // occupy the slot; a superseded one disposes itself, outside the gate.
+ // Subscribing can push a value downstream synchronously, and that handler is free to feed the outer
+ // source again, re-entering OnSource and installing a newer generation before this call returns. Only
+ // the current generation may occupy the slot: installing a superseded subscription would dispose the
+ // newer one and leave a sequence whose notifications are all filtered out by version.
var superseded = false;
lock (_gate)
{
diff --git a/src/Primitives.Shared/Advanced/SynchronizeObjectSignal{T}.cs b/src/Primitives.Shared/Advanced/SynchronizeObjectSignal{T}.cs
index 7629a5ae..dcdaaa7b 100644
--- a/src/Primitives.Shared/Advanced/SynchronizeObjectSignal{T}.cs
+++ b/src/Primitives.Shared/Advanced/SynchronizeObjectSignal{T}.cs
@@ -8,7 +8,7 @@ namespace ReactiveUI.Primitives.Reactive.Advanced;
namespace ReactiveUI.Primitives.Advanced;
#endif
-/// Object-gated