feat(primitives): add Rx compatibility APIs and advanced sinks

src/Directory.Build.props

@@ -96,6 +96,7 @@
     <Compile Include="..\Polyfills\NotNullWhenAttribute.cs" Link="Polyfills\NotNullWhenAttribute.cs" />
     <Compile Include="..\Polyfills\MemberNotNullWhenAttribute.cs" Link="Polyfills\MemberNotNullWhenAttribute.cs" />
     <Compile Include="..\Polyfills\NotNullAttribute.cs" Link="Polyfills\NotNullAttribute.cs" />
+    <Compile Include="..\Polyfills\AllowNullAttribute.cs" Link="Polyfills\AllowNullAttribute.cs" />
     <Compile Include="..\Polyfills\CallerArgumentExpressionAttribute.cs" Link="Polyfills\CallerArgumentExpressionAttribute.cs" />
     <Compile Include="..\Polyfills\IsExternalInit.cs" Link="Polyfills\IsExternalInit.cs" />
     <Compile Include="..\Polyfills\RequiredMemberAttribute.cs" Link="Polyfills\RequiredMemberAttribute.cs" />

src/Directory.Packages.props

@@ -9,7 +9,7 @@
   </ItemGroup>
   <ItemGroup>
     <PackageVersion Include="MinVer" Version="7.0.0" />
-    <PackageVersion Include="StyleSharp.Analyzers" Version="3.11.1" />
+    <PackageVersion Include="StyleSharp.Analyzers" Version="3.11.2" />
     <PackageVersion Include="Roslynator.Analyzers" Version="4.15.0" />
     <PackageVersion Include="Microsoft.CodeAnalysis.BannedApiAnalyzers" Version="5.0.0-1.25277.114" />
     <PackageVersion Include="Microsoft.CodeAnalysis.PublicApiAnalyzers" Version="5.0.0-1.25277.114" />

src/Polyfills/AllowNullAttribute.cs

@@ -0,0 +1,14 @@
+// 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.
+
+// Polyfill implementation adapted from SimonCropp/Polyfill (https://github.com/SimonCropp/Polyfill).
+#if !NETCOREAPP3_0_OR_GREATER && !NETSTANDARD2_1_OR_GREATER
+namespace System.Diagnostics.CodeAnalysis;
+
+/// <summary>Specifies that <see langword="null"/> is allowed as an input even if the corresponding type disallows it.</summary>
+[ExcludeFromCodeCoverage]
+[DebuggerNonUserCode]
+[AttributeUsage(AttributeTargets.Field | AttributeTargets.Parameter | AttributeTargets.Property)]
+internal sealed class AllowNullAttribute : Attribute;
+#endif

src/Primitives.Extensions.Shared/ReactiveExtensions.cs

@@ -22,6 +22,9 @@ public static class ReactiveExtensions
     /// <summary>Default backoff factor for <see cref="RetryWithBackoff{T}(IObservable{T}, int, TimeSpan)"/>: each retry doubles the previous delay.</summary>
     private const double DefaultBackoffFactor = 2.0;
 
+    /// <summary>Default match timeout for regex filters created from string patterns.</summary>
+    private static readonly TimeSpan DefaultRegexMatchTimeout = TimeSpan.FromSeconds(30);
+
     /// <summary>Boolean reduction operators for a set of boolean observable sources.</summary>
     /// <param name="sources">The sources.</param>
     extension(IEnumerable<IObservable<bool>> sources)
@@ -906,7 +909,7 @@ public IObservable<string> BufferUntil(char startsWith, char endsWith) =>
         /// <param name="regexPattern">Regex pattern.</param>
         /// <returns>Filtered sequence.</returns>
         public IObservable<string> Filter(string regexPattern) =>
-            source.Filter(new Regex(regexPattern, RegexOptions.None, TimeSpan.FromSeconds(1)));
+            source.Filter(new Regex(regexPattern, RegexOptions.None, DefaultRegexMatchTimeout));
 
         /// <summary>Filters strings by regex.</summary>
         /// <param name="regex">Regex.</param>

src/Primitives.Shared/Advanced/AsyncSubscriptionLifetime.cs

@@ -0,0 +1,90 @@
+// 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>Owns cancellation and the eventual inner disposable for asynchronous signal subscriptions.</summary>
+/// <remarks>
+/// Call <see cref="Complete"/> exactly once after asynchronous setup has finished, even when setup faults or is canceled.
+/// </remarks>
+public sealed class AsyncSubscriptionLifetime : IDisposable
+{
+    /// <summary>The cancellation source passed to the asynchronous subscription.</summary>
+    private readonly CancellationTokenSource _cts = new();
+
+    /// <summary>The inner disposable returned by the asynchronous subscription.</summary>
+    private readonly SingleDisposable _subscription = new();
+
+    /// <summary>Non-zero once the outer subscription has been disposed.</summary>
+    private int _disposed;
+
+    /// <summary>Non-zero when disposal requested cancellation before the asynchronous subscription completed.</summary>
+    private int _canceled;
+
+    /// <summary>Non-zero once the asynchronous subscription task has completed.</summary>
+    private int _completed;
+
+    /// <summary>Gets the token supplied to the asynchronous subscription.</summary>
+    public CancellationToken Token => _cts.Token;
+
+    /// <summary>Gets a value indicating whether disposal requested cancellation before asynchronous setup completed.</summary>
+    public bool IsCancellationRequested => Volatile.Read(ref _canceled) != 0;
+
+    /// <summary>Assigns the disposable returned by asynchronous setup.</summary>
+    /// <param name="disposable">The returned disposable, or <see langword="null"/> for an empty lifetime.</param>
+    public void SetSubscription(IDisposable? disposable) =>
+        _subscription.Create(disposable ?? EmptyDisposable.Instance);
+
+    /// <summary>Marks asynchronous setup complete and releases the cancellation source when still owned here.</summary>
+    public void Complete()
+    {
+        Volatile.Write(ref _completed, 1);
+        if (Volatile.Read(ref _disposed) != 0)
+        {
+            return;
+        }
+
+        _cts.Dispose();
+    }
+
+    /// <inheritdoc/>
+    public void Dispose()
+    {
+        if (Interlocked.Exchange(ref _disposed, 1) != 0)
+        {
+            return;
+        }
+
+        if (Volatile.Read(ref _completed) != 0)
+        {
+            _subscription.Dispose();
+            _cts.Dispose();
+            return;
+        }
+
+        Volatile.Write(ref _canceled, 1);
+        CancelIgnoringDisposed(_cts);
+        _subscription.Dispose();
+        _cts.Dispose();
+    }
+
+    /// <summary>Cancels the source while tolerating a concurrent completion disposing it first.</summary>
+    /// <param name="cts">The cancellation source to cancel.</param>
+    [System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
+    private static void CancelIgnoringDisposed(CancellationTokenSource cts)
+    {
+        try
+        {
+            cts.Cancel();
+        }
+        catch (ObjectDisposedException)
+        {
+            // Completion can release the CTS concurrently; disposal still continues with the inner subscription.
+        }
+    }
+}

src/Primitives.Shared/Advanced/CurrentThreadRequirement.cs

@@ -0,0 +1,27 @@
+// 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>Shared current-thread requirement checks for advanced signals.</summary>
+internal static class CurrentThreadRequirement
+{
+    /// <summary>Determines whether a source requires current-thread subscription.</summary>
+    /// <typeparam name="T">The source value type.</typeparam>
+    /// <param name="source">The source observable.</param>
+    /// <returns><see langword="true"/> when the source requires current-thread subscription.</returns>
+    public static bool IsRequired<T>(IObservable<T> source)
+    {
+        if (source is not IRequireCurrentThread<T> currentThread)
+        {
+            return false;
+        }
+
+        return currentThread.IsRequiredSubscribeOnCurrentThread();
+    }
+}

src/Primitives.Shared/Advanced/MapIndexedSignal{TSource,TResult}.cs

@@ -0,0 +1,42 @@
+// 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>Indexed map signal.</summary>
+/// <typeparam name="TSource">The source value type.</typeparam>
+/// <typeparam name="TResult">The projected value type.</typeparam>
+public sealed class MapIndexedSignal<TSource, TResult> : IRequireCurrentThread<TResult>
+{
+    /// <summary>The source observable.</summary>
+    private readonly IObservable<TSource> _source;
+
+    /// <summary>The indexed selector.</summary>
+    private readonly Func<TSource, int, TResult> _selector;
+
+    /// <summary>Initializes a new instance of the <see cref="MapIndexedSignal{TSource, TResult}"/> class.</summary>
+    /// <param name="source">The source observable.</param>
+    /// <param name="selector">The indexed selector.</param>
+    public MapIndexedSignal(IObservable<TSource> source, Func<TSource, int, TResult> selector)
+    {
+        _source = source;
+        _selector = selector;
+    }
+
+    /// <inheritdoc/>
+    public bool IsRequiredSubscribeOnCurrentThread() =>
+        CurrentThreadRequirement.IsRequired(_source);
+
+    /// <inheritdoc/>
+    public IDisposable Subscribe(IObserver<TResult> observer)
+    {
+        ArgumentExceptionHelper.ThrowIfNull(observer);
+
+        return _source.Subscribe(new MapIndexedWitness<TSource, TResult>(observer, _selector));
+    }
+}

src/Primitives.Shared/Advanced/MapIndexedWitness{TSource,TResult}.cs

@@ -0,0 +1,82 @@
+// 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>Applies an indexed selector to source values.</summary>
+/// <typeparam name="TSource">The source value type.</typeparam>
+/// <typeparam name="TResult">The projected value type.</typeparam>
+public sealed class MapIndexedWitness<TSource, TResult> : IObserver<TSource>
+{
+    /// <summary>The downstream observer.</summary>
+    private readonly IObserver<TResult> _observer;
+
+    /// <summary>The indexed selector.</summary>
+    private readonly Func<TSource, int, TResult> _selector;
+
+    /// <summary>The next zero-based index.</summary>
+    private int _index;
+
+    /// <summary>Whether a terminal notification has been forwarded.</summary>
+    private bool _stopped;
+
+    /// <summary>Initializes a new instance of the <see cref="MapIndexedWitness{TSource, TResult}"/> class.</summary>
+    /// <param name="observer">The downstream observer.</param>
+    /// <param name="selector">The indexed selector.</param>
+    public MapIndexedWitness(IObserver<TResult> observer, Func<TSource, int, TResult> selector)
+    {
+        _observer = observer;
+        _selector = selector;
+    }
+
+    /// <inheritdoc/>
+    public void OnNext(TSource value)
+    {
+        if (_stopped)
+        {
+            return;
+        }
+
+        TResult result;
+        try
+        {
+            result = _selector(value, _index++);
+        }
+        catch (Exception error) when (!FatalExceptionHelper.IsFatal(error))
+        {
+            OnError(error);
+            return;
+        }
+
+        _observer.OnNext(result);
+    }
+
+    /// <inheritdoc/>
+    public void OnError(Exception error)
+    {
+        if (_stopped)
+        {
+            return;
+        }
+
+        _stopped = true;
+        _observer.OnError(error);
+    }
+
+    /// <inheritdoc/>
+    public void OnCompleted()
+    {
+        if (_stopped)
+        {
+            return;
+        }
+
+        _stopped = true;
+        _observer.OnCompleted();
+    }
+}