feat(operators): add Repeat aliases

.editorconfig

@@ -7,6 +7,7 @@ root = true
 # Default settings
 #############################################
 [*]
+end_of_line = lf
 insert_final_newline = true
 indent_style = space
 indent_size = 4

.gitattributes

@@ -1,6 +1,6 @@
 # Auto-detect text files and normalise line endings to LF in the repository.
 # Working trees check out LF on every platform (auto, no forced CRLF).
-* text=auto
+* text eol=lf
 
 # Source code
 *.cs       text diff=csharp

src/Primitives.Shared/Advanced/RepeatSourceCoordinator{T}.cs

@@ -0,0 +1,248 @@
+// Copyright (c) 2019-2026 ReactiveUI Association Incorporated. All rights reserved.
+// ReactiveUI Association Incorporated licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for full license information.
+
+#if REACTIVE_SHIM
+namespace ReactiveUI.Primitives.Reactive.Advanced;
+#else
+namespace ReactiveUI.Primitives.Advanced;
+#endif
+
+/// <summary>Coordinates sequential subscriptions for <see cref="RepeatSourceSignal{T}"/>.</summary>
+/// <typeparam name="T">The value type.</typeparam>
+public sealed class RepeatSourceCoordinator<T> : IDisposable
+{
+    /// <summary>The source sequence.</summary>
+    private readonly IObservable<T> _source;
+
+    /// <summary>The downstream observer.</summary>
+    private readonly IObserver<T> _observer;
+
+    /// <summary>The configured repeat count.</summary>
+    private readonly int? _repeatCount;
+
+    /// <summary>The active source subscription or queued resubscription.</summary>
+    private readonly SingleReplaceableDisposable _active = new();
+
+    /// <summary>Guards synchronous completion while a subscription is still being assigned.</summary>
+    private readonly Lock _gate = new();
+
+    /// <summary>The remaining number of finite subscriptions.</summary>
+    private int _remaining;
+
+    /// <summary>Tracks whether a source subscription is currently being created.</summary>
+    private bool _subscribing;
+
+    /// <summary>Tracks synchronous completion before the subscription disposable is returned.</summary>
+    private bool _completedWhileSubscribing;
+
+    /// <summary>Tracks the current subscription generation.</summary>
+    private int _generation;
+
+    /// <summary>The generation currently allowed to forward notifications.</summary>
+    private int _activeGeneration;
+
+    /// <summary>Tracks disposal and terminal notification state.</summary>
+    private int _disposed;
+
+    /// <summary>Initializes a new instance of the <see cref="RepeatSourceCoordinator{T}"/> class.</summary>
+    /// <param name="source">The source sequence.</param>
+    /// <param name="repeatCount">The number of repetitions, or <see langword="null"/> for indefinite repetition.</param>
+    /// <param name="observer">The downstream observer.</param>
+    /// <exception cref="ArgumentNullException">
+    /// <paramref name="source"/> or <paramref name="observer"/> is <see langword="null"/>.
+    /// </exception>
+    /// <exception cref="ArgumentOutOfRangeException"><paramref name="repeatCount"/> is less than zero.</exception>
+    public RepeatSourceCoordinator(IObservable<T> source, int? repeatCount, IObserver<T> observer)
+    {
+        ArgumentExceptionHelper.ThrowIfNull(source);
+
+        if (repeatCount.HasValue)
+        {
+            ArgumentOutOfRangeExceptionHelper.ThrowIfNegative(repeatCount.GetValueOrDefault());
+        }
+
+        ArgumentExceptionHelper.ThrowIfNull(observer);
+
+        _source = source;
+        _observer = observer;
+        _repeatCount = repeatCount;
+        _remaining = repeatCount.GetValueOrDefault();
+    }
+
+    /// <inheritdoc/>
+    public void Dispose()
+    {
+        if (Interlocked.Exchange(ref _disposed, 1) != 0)
+        {
+            return;
+        }
+
+        _active.Dispose();
+    }
+
+    /// <summary>Starts the repeated subscription loop.</summary>
+    /// <returns>The coordinator that owns the subscription cleanup.</returns>
+    public RepeatSourceCoordinator<T> Run()
+    {
+        ScheduleNext();
+        return this;
+    }
+
+    /// <summary>Handles completion for the active source subscription.</summary>
+    /// <param name="generation">The source subscription generation.</param>
+    internal void OnCompleted(int generation)
+    {
+        if (!IsCurrentGeneration(generation))
+        {
+            return;
+        }
+
+        ScheduleNext();
+    }
+
+    /// <summary>Handles failure for the active source subscription.</summary>
+    /// <param name="generation">The source subscription generation.</param>
+    /// <param name="error">The source error.</param>
+    internal void OnError(int generation, Exception error)
+    {
+        if (!IsCurrentGeneration(generation) || Interlocked.Exchange(ref _disposed, 1) != 0)
+        {
+            return;
+        }
+
+        try
+        {
+            _observer.OnError(error);
+        }
+        finally
+        {
+            _active.Dispose();
+        }
+    }
+
+    /// <summary>Forwards a value from the active source subscription.</summary>
+    /// <param name="generation">The source subscription generation.</param>
+    /// <param name="value">The source value.</param>
+    internal void OnNext(int generation, T value)
+    {
+        if (IsDisposed() || !IsCurrentGeneration(generation))
+        {
+            return;
+        }
+
+        _observer.OnNext(value);
+    }
+
+    /// <summary>Completes the downstream observer once.</summary>
+    internal void Complete()
+    {
+        if (Interlocked.Exchange(ref _disposed, 1) != 0)
+        {
+            return;
+        }
+
+        try
+        {
+            _observer.OnCompleted();
+        }
+        finally
+        {
+            _active.Dispose();
+        }
+    }
+
+    /// <summary>Schedules the next source subscription on the current-thread trampoline.</summary>
+    internal void ScheduleNext()
+    {
+        if (IsDisposed())
+        {
+            return;
+        }
+
+        lock (_gate)
+        {
+            _completedWhileSubscribing |= _subscribing;
+        }
+
+        var scheduled = Sequencer.CurrentThread.Schedule(SubscribeNext);
+        if (ReferenceEquals(scheduled, EmptyDisposable.Instance) || IsDisposed())
+        {
+            return;
+        }
+
+        _active.Create(scheduled);
+    }
+
+    /// <summary>Subscribes to the source for the next repetition.</summary>
+    internal void SubscribeNext()
+    {
+        if (IsDisposed())
+        {
+            return;
+        }
+
+        if (_repeatCount is not null && _remaining == 0)
+        {
+            Complete();
+            return;
+        }
+
+        if (_repeatCount is not null)
+        {
+            _remaining--;
+        }
+
+        var generation = Interlocked.Increment(ref _generation);
+        Volatile.Write(ref _activeGeneration, generation);
+        RepeatSourceWitness<T> observer = new(this, generation);
+        IDisposable? subscription = null;
+        var completedWhileSubscribing = false;
+        lock (_gate)
+        {
+            _subscribing = true;
+            _completedWhileSubscribing = false;
+        }
+
+        try
+        {
+            subscription = _source.Subscribe(observer);
+        }
+        catch (Exception error)
+        {
+            observer.OnError(error);
+        }
+        finally
+        {
+            lock (_gate)
+            {
+                completedWhileSubscribing = _completedWhileSubscribing;
+                _completedWhileSubscribing = false;
+                _subscribing = false;
+            }
+        }
+
+        if (subscription is null)
+        {
+            return;
+        }
+
+        if (completedWhileSubscribing || IsDisposed())
+        {
+            subscription.Dispose();
+            return;
+        }
+
+        _active.Create(subscription);
+    }
+
+    /// <summary>Gets a value indicating whether the coordinator is disposed.</summary>
+    /// <returns><see langword="true"/> when the coordinator is disposed; otherwise, <see langword="false"/>.</returns>
+    private bool IsDisposed() => Volatile.Read(ref _disposed) != 0;
+
+    /// <summary>Gets a value indicating whether a notification belongs to the active generation.</summary>
+    /// <param name="generation">The source subscription generation.</param>
+    /// <returns><see langword="true"/> when the notification belongs to the active generation; otherwise, <see langword="false"/>.</returns>
+    private bool IsCurrentGeneration(int generation) =>
+        Volatile.Read(ref _activeGeneration) == generation;
+}

src/Primitives.Shared/Advanced/RepeatSourceSignal{T}.cs

@@ -0,0 +1,59 @@
+// Copyright (c) 2019-2026 ReactiveUI Association Incorporated. All rights reserved.
+// ReactiveUI Association Incorporated licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for full license information.
+
+#if REACTIVE_SHIM
+namespace ReactiveUI.Primitives.Reactive.Advanced;
+#else
+namespace ReactiveUI.Primitives.Advanced;
+#endif
+
+/// <summary>Repeats a source observable by resubscribing after each successful completion.</summary>
+/// <typeparam name="T">The value type.</typeparam>
+public sealed class RepeatSourceSignal<T> : IRequireCurrentThread<T>
+{
+    /// <summary>The source sequence.</summary>
+    private readonly IObservable<T> _source;
+
+    /// <summary>The number of repetitions, or <see langword="null"/> for indefinite repetition.</summary>
+    private readonly int? _repeatCount;
+
+    /// <summary>Initializes a new instance of the <see cref="RepeatSourceSignal{T}"/> class.</summary>
+    /// <param name="source">The source sequence.</param>
+    /// <param name="repeatCount">The number of repetitions, or <see langword="null"/> for indefinite repetition.</param>
+    /// <exception cref="ArgumentNullException"><paramref name="source"/> is <see langword="null"/>.</exception>
+    /// <exception cref="ArgumentOutOfRangeException"><paramref name="repeatCount"/> is less than zero.</exception>
+    public RepeatSourceSignal(IObservable<T> source, int? repeatCount)
+    {
+        ArgumentExceptionHelper.ThrowIfNull(source);
+
+        if (repeatCount.HasValue)
+        {
+            ArgumentOutOfRangeExceptionHelper.ThrowIfNegative(repeatCount.GetValueOrDefault());
+        }
+
+        _source = source;
+        _repeatCount = repeatCount;
+    }
+
+    /// <inheritdoc/>
+    public bool IsRequiredSubscribeOnCurrentThread() => true;
+
+    /// <inheritdoc/>
+    public IDisposable Subscribe(IObserver<T> observer)
+    {
+        ArgumentExceptionHelper.ThrowIfNull(observer);
+
+        return SignalSubscription.Subscribe(observer, true, SubscribeCore);
+    }
+
+    /// <summary>Starts the repeat coordinator once the subscription lifetime has been created.</summary>
+    /// <param name="observer">The downstream observer.</param>
+    /// <param name="cancel">The cancellation handle owned by the subscription helper.</param>
+    /// <returns>The repeat coordinator.</returns>
+    private RepeatSourceCoordinator<T> SubscribeCore(IObserver<T> observer, IDisposable cancel)
+    {
+        RepeatSourceCoordinator<T> coordinator = new(_source, _repeatCount, new GuardedWitness<T>(observer, cancel));
+        return coordinator.Run();
+    }
+}

src/Primitives.Shared/Advanced/RepeatSourceWitness{T}.cs

@@ -0,0 +1,68 @@
+// Copyright (c) 2019-2026 ReactiveUI Association Incorporated. All rights reserved.
+// ReactiveUI Association Incorporated licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for full license information.
+
+#if REACTIVE_SHIM
+namespace ReactiveUI.Primitives.Reactive.Advanced;
+#else
+namespace ReactiveUI.Primitives.Advanced;
+#endif
+
+/// <summary>Per-subscription witness that suppresses duplicate terminal notifications.</summary>
+/// <typeparam name="T">The value type.</typeparam>
+public sealed class RepeatSourceWitness<T> : IObserver<T>
+{
+    /// <summary>The owning repeat coordinator.</summary>
+    private readonly RepeatSourceCoordinator<T> _parent;
+
+    /// <summary>The source subscription generation.</summary>
+    private readonly int _generation;
+
+    /// <summary>Tracks whether this source subscription has already terminated.</summary>
+    private int _terminated;
+
+    /// <summary>Initializes a new instance of the <see cref="RepeatSourceWitness{T}"/> class.</summary>
+    /// <param name="parent">The owning repeat coordinator.</param>
+    /// <param name="generation">The source subscription generation.</param>
+    /// <exception cref="ArgumentNullException"><paramref name="parent"/> is <see langword="null"/>.</exception>
+    public RepeatSourceWitness(RepeatSourceCoordinator<T> parent, int generation)
+    {
+        ArgumentExceptionHelper.ThrowIfNull(parent);
+
+        _parent = parent;
+        _generation = generation;
+    }
+
+    /// <inheritdoc/>
+    public void OnCompleted()
+    {
+        if (Interlocked.Exchange(ref _terminated, 1) != 0)
+        {
+            return;
+        }
+
+        _parent.OnCompleted(_generation);
+    }
+
+    /// <inheritdoc/>
+    public void OnError(Exception error)
+    {
+        if (Interlocked.Exchange(ref _terminated, 1) != 0)
+        {
+            return;
+        }
+
+        _parent.OnError(_generation, error);
+    }
+
+    /// <inheritdoc/>
+    public void OnNext(T value)
+    {
+        if (Volatile.Read(ref _terminated) != 0)
+        {
+            return;
+        }
+
+        _parent.OnNext(_generation, value);
+    }
+}