CHANGELOG.md

@@ -40,7 +40,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 BigEndian/LittleEndian methods.
 - X10D: `Stream.GetHash<>` and `Stream.TryWriteHash<>` now throw ArgumentException in lieu of
 TypeInitializationException.
-- X10D: `char.IsEmoji` no longer allocates for .NET 7+.
+- X10D: `char.IsEmoji` no longer allocates for .NET 7.
 - X10D: `string.Repeat` is now more efficient.
 
 ### Removed
@@ -49,7 +49,6 @@ TypeInitializationException.
 - X10D: Removed `Endianness` enum.
 - X10D: Removed `Span<T>.Replace(T, T)` for .NET 8 target.
 - X10D: Removed .NET Standard 2.1 target.
-- X10D: Removed extensions for `Progress<T>`. These are already provided by [`Observable.FromEventPattern`](https://learn.microsoft.com/en-us/previous-versions/dotnet/reactive-extensions/hh229424(v=vs.103)).
 - X10D.Hosting: Removed .NET Standard 2.1 target.
 - X10D.DSharpPlus: Complete sunset of library. This library will not be updated to support DSharpPlus v5.0.0 (#83).
 - X10D.Unity: Complete sunset of library. This library will not be updated effective immediately (#86).

X10D.Tests/src/Reactive/ProgressTests.cs

@@ -0,0 +1,67 @@
+using NUnit.Framework;
+using X10D.Reactive;
+
+namespace X10D.Tests.Reactive;
+
+[TestFixture]
+internal class ProgressTests
+{
+    [Test]
+    public void OnProgressChanged_ShouldCallCompletionDelegate_GivenCompletionValue()
+    {
+        var subscriberWasCalled = false;
+        var completionWasCalled = false;
+
+        var progress = new Progress<float>();
+        progress.OnProgressChanged(1.0f).Subscribe(_ => subscriberWasCalled = true, () => completionWasCalled = true);
+
+        ((IProgress<float>)progress).Report(0.5f);
+        ((IProgress<float>)progress).Report(1.0f);
+
+        Thread.Sleep(1000);
+        Assert.That(subscriberWasCalled);
+        Assert.That(completionWasCalled);
+    }
+
+    [Test]
+    public void OnProgressChanged_ShouldCallSubscribers_OnProgressChanged()
+    {
+        var subscriberWasCalled = false;
+
+        var progress = new Progress<float>();
+        progress.OnProgressChanged().Subscribe(_ => subscriberWasCalled = true);
+
+        ((IProgress<float>)progress).Report(0.5f);
+
+        Thread.Sleep(1000);
+        Assert.That(subscriberWasCalled);
+    }
+
+    [Test]
+    public void OnProgressChanged_ShouldCallSubscribers_OnProgressChanged_GivenCompletionValue()
+    {
+        var subscriberWasCalled = false;
+
+        var progress = new Progress<float>();
+        progress.OnProgressChanged(1.0f).Subscribe(_ => subscriberWasCalled = true);
+
+        ((IProgress<float>)progress).Report(0.5f);
+
+        Thread.Sleep(1000);
+        Assert.That(subscriberWasCalled);
+    }
+
+    [Test]
+    public void OnProgressChanged_ShouldThrowArgumentNullException_GivenNullProgress()
+    {
+        Progress<float> progress = null!;
+        Assert.Throws<ArgumentNullException>(() => progress.OnProgressChanged());
+    }
+
+    [Test]
+    public void OnProgressChanged_ShouldThrowArgumentNullException_GivenNullProgressAndCompletionValue()
+    {
+        Progress<float> progress = null!;
+        Assert.Throws<ArgumentNullException>(() => progress.OnProgressChanged(1.0f));
+    }
+}

X10D/src/Reactive/ObservableDisposer.cs

@@ -0,0 +1,33 @@
+namespace X10D.Reactive;
+
+/// <summary>
+///     Represents a disposable that removes an observer from a collection of observers.
+/// </summary>
+internal readonly struct ObservableDisposer<T> : IDisposable
+{
+    private readonly HashSet<IObserver<T>> _observers;
+    private readonly IObserver<T> _observer;
+    private readonly Action? _additionalAction;
+
+    /// <summary>
+    ///     Initializes a new instance of the <see cref="ObservableDisposer{T}" /> struct.
+    /// </summary>
+    /// <param name="observers">A collection of observers from which to remove the specified observer.</param>
+    /// <param name="observer">The observer to remove from the collection.</param>
+    /// <param name="additionalAction">The additional action to run on dispose.</param>
+    public ObservableDisposer(HashSet<IObserver<T>> observers, IObserver<T> observer, Action? additionalAction)
+    {
+        _observers = observers ?? throw new ArgumentNullException(nameof(observers));
+        _observer = observer ?? throw new ArgumentNullException(nameof(observer));
+        _additionalAction = additionalAction;
+    }
+
+    /// <summary>
+    ///     Removes the observer from the collection of observers.
+    /// </summary>
+    public void Dispose()
+    {
+        _observers.Remove(_observer);
+        _additionalAction?.Invoke();
+    }
+}

X10D/src/Reactive/ProgressExtensions.cs

@@ -0,0 +1,89 @@
+namespace X10D.Reactive;
+
+/// <summary>
+///     Provides extension methods for <see cref="Progress{T}" />.
+/// </summary>
+public static class ProgressExtensions
+{
+    /// <summary>
+    ///     Wraps the <see cref="Progress{T}.ProgressChanged" /> event of the current <see cref="Progress{T}" /> in an
+    ///     <see cref="IObservable{T}" /> object.
+    /// </summary>
+    /// <param name="progress">The progress whose <see cref="Progress{T}.ProgressChanged" /> event to wrap.</param>
+    /// <typeparam name="T">The type of progress update value.</typeparam>
+    /// <returns>
+    ///     An <see cref="IObservable{T}" /> object that wraps the <see cref="Progress{T}.ProgressChanged" /> event of the current
+    ///     <see cref="Progress{T}" />.
+    /// </returns>
+    /// <exception cref="ArgumentNullException"><paramref name="progress" /> is <see langword="null" />.</exception>
+    public static IObservable<T> OnProgressChanged<T>(this Progress<T> progress)
+    {
+        if (progress is null)
+        {
+            throw new ArgumentNullException(nameof(progress));
+        }
+
+        var progressObservable = new ProgressObservable<T>();
+
+        void ProgressChangedHandler(object? sender, T args)
+        {
+            IObserver<T>[] observers = progressObservable.Observers;
+
+            for (var index = 0; index < observers.Length; index++)
+            {
+                observers[index].OnNext(args);
+            }
+        }
+
+        progress.ProgressChanged += ProgressChangedHandler;
+        progressObservable.OnDispose = () => progress.ProgressChanged -= ProgressChangedHandler;
+
+        return progressObservable;
+    }
+
+    /// <summary>
+    ///     Wraps the <see cref="Progress{T}.ProgressChanged" /> event of the current <see cref="Progress{T}" /> in an
+    ///     <see cref="IObservable{T}" /> object, and completes the observable when the progress reaches the specified value.
+    /// </summary>
+    /// <param name="progress">The progress whose <see cref="Progress{T}.ProgressChanged" /> event to wrap.</param>
+    /// <param name="completeValue">The value that indicates completion.</param>
+    /// <typeparam name="T">The type of progress update value.</typeparam>
+    /// <returns>
+    ///     An <see cref="IObservable{T}" /> object that wraps the <see cref="Progress{T}.ProgressChanged" /> event of the current
+    ///     <see cref="Progress{T}" />.
+    /// </returns>
+    /// <exception cref="ArgumentNullException"><paramref name="progress" /> is <see langword="null" />.</exception>
+    public static IObservable<T> OnProgressChanged<T>(this Progress<T> progress, T completeValue)
+    {
+        if (progress is null)
+        {
+            throw new ArgumentNullException(nameof(progress));
+        }
+
+        var progressObservable = new ProgressObservable<T>();
+        var comparer = EqualityComparer<T>.Default;
+
+        void ProgressChangedHandler(object? sender, T args)
+        {
+            IObserver<T>[] observers = progressObservable.Observers;
+
+            for (var index = 0; index < observers.Length; index++)
+            {
+                observers[index].OnNext(args);
+            }
+
+            if (comparer.Equals(args, completeValue))
+            {
+                for (var index = 0; index < observers.Length; index++)
+                {
+                    observers[index].OnCompleted();
+                }
+            }
+        }
+
+        progress.ProgressChanged += ProgressChangedHandler;
+        progressObservable.OnDispose = () => progress.ProgressChanged -= ProgressChangedHandler;
+
+        return progressObservable;
+    }
+}

X10D/src/Reactive/ProgressObservable.cs

@@ -0,0 +1,36 @@
+namespace X10D.Reactive;
+
+/// <summary>
+///     Represents a concrete implementation of <see cref="IObservable{T}" /> that tracks progress of a <see cref="Progress{T}"/>.
+/// </summary>
+internal sealed class ProgressObservable<T> : IObservable<T>
+{
+    private readonly HashSet<IObserver<T>> _observers = new();
+
+    /// <summary>
+    ///     Gets the observers.
+    /// </summary>
+    /// <value>The observers.</value>
+    public IObserver<T>[] Observers
+    {
+        get => _observers.ToArray();
+    }
+
+    internal Action? OnDispose { get; set; }
+
+    /// <summary>
+    ///     Subscribes the specified observer to the progress tracker.
+    /// </summary>
+    /// <param name="observer">The observer.</param>
+    /// <returns>An object which can be disposed to unsubscribe from progress tracking.</returns>
+    public IDisposable Subscribe(IObserver<T> observer)
+    {
+        if (observer is null)
+        {
+            throw new ArgumentNullException(nameof(observer));
+        }
+
+        _observers.Add(observer);
+        return new ObservableDisposer<T>(_observers, observer, OnDispose);
+    }
+}

codecov.yml

@@ -1,2 +0,0 @@
-ignore:
-  - "X10D/src/ExceptionMessages.Designer.cs"