Update SubscribeSynchronous and Add SubscribeAsync (#29)

Update Synchronous and Async Subscribe

Rename SynchronizeAsync to SynchronizeSynchronous as is blocking to the source.
Add SynchronizeAsync as a non blocking version of the same.

Author: ChrisPulman

README.md

@@ -36,4 +36,6 @@ Extensions for concerns found in System.Reactive that make consuming the library
 - OnErrorRetry
 - TakeUntil
 - SyncronizeAsync
+- SubscribeAsync
+- SyncronizeSynchronous
 - SubscribeSynchronous

src/ReactiveMarbles.Extensions.Tests/ReactiveExtensionsTests.cs

@@ -157,4 +157,96 @@ public void SyncronizeAsync_RunsWithAsyncTasksInSubscriptions()
             .Should()
             .Be(0);
     }
+
+    /// <summary>
+    /// Syncronizes the asynchronous runs with asynchronous tasks in subscriptions.
+    /// </summary>
+    [Fact]
+    public void SynchronizeSynchronous_RunsWithAsyncTasksInSubscriptions()
+    {
+        // Given, When
+        var result = 0;
+        var itterations = 0;
+        var subject = new Subject<bool>();
+        using var disposable = subject
+            .SynchronizeSynchronous()
+            .Subscribe(async x =>
+            {
+                if (x.Value)
+                {
+                    await Task.Delay(1000);
+                    result++;
+                }
+                else
+                {
+                    await Task.Delay(500);
+                    result--;
+                }
+
+                x.Sync.Dispose();
+                itterations++;
+            });
+
+        subject.OnNext(true);
+        subject.OnNext(false);
+        subject.OnNext(true);
+        subject.OnNext(false);
+        subject.OnNext(true);
+        subject.OnNext(false);
+
+        while (itterations < 6)
+        {
+            Thread.Yield();
+        }
+
+        // Then
+        result
+            .Should()
+            .Be(0);
+    }
+
+    /// <summary>
+    /// Syncronizes the asynchronous runs with asynchronous tasks in subscriptions.
+    /// </summary>
+    [Fact]
+    public void SubscribeAsync_RunsWithAsyncTasksInSubscriptions()
+    {
+        // Given, When
+        var result = 0;
+        var itterations = 0;
+        var subject = new Subject<bool>();
+        using var disposable = subject
+            .SubscribeAsync(async x =>
+            {
+                if (x)
+                {
+                    await Task.Delay(1000);
+                    result++;
+                }
+                else
+                {
+                    await Task.Delay(500);
+                    result--;
+                }
+
+                itterations++;
+            });
+
+        subject.OnNext(true);
+        subject.OnNext(false);
+        subject.OnNext(true);
+        subject.OnNext(false);
+        subject.OnNext(true);
+        subject.OnNext(false);
+
+        while (itterations < 6)
+        {
+            Thread.Yield();
+        }
+
+        // Then
+        result
+            .Should()
+            .Be(0);
+    }
 }

src/ReactiveMarbles.Extensions/Continuation.cs

@@ -30,7 +30,6 @@ public class Continuation : IDisposable
     /// </summary>
     public void Dispose()
     {
-        // Do not change this code. Put cleanup code in 'Dispose(bool disposing)' method
         Dispose(disposing: true);
         GC.SuppressFinalize(this);
     }

src/ReactiveMarbles.Extensions/ReactiveExtensions.cs

@@ -913,12 +913,8 @@ public static IObservable<TSource> TakeUntil<TSource>(this IObservable<TSource>
     /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
     /// <param name="source">The source.</param>
     /// <returns>An Observable of T and a release mechanism.</returns>
-    public static IObservable<(T Value, IDisposable Sync)> SynchronizeAsync<T>(this IObservable<T> source) =>
-        Observable.Create<(T Value, IDisposable Sync)>(observer =>
-        {
-            var gate = new object();
-            return source.Synchronize(gate).Subscribe(item => new Continuation().Lock(item, observer).Wait());
-        });
+    public static IObservable<(T Value, IDisposable Sync)> SynchronizeSynchronous<T>(this IObservable<T> source) =>
+        Observable.Create<(T Value, IDisposable Sync)>(observer => source.Subscribe(item => new Continuation().Lock(item, observer).Wait()));
 
     /// <summary>
     /// Subscribes to the specified source synchronously.
@@ -930,7 +926,7 @@ public static IObservable<TSource> TakeUntil<TSource>(this IObservable<TSource>
     /// <param name="onCompleted">The on completed.</param>
     /// <returns><see cref="IDisposable"/> object used to unsubscribe from the observable sequence.</returns>
     public static IDisposable SubscribeSynchronous<T>(this IObservable<T> source, Func<T, Task> onNext, Action<Exception> onError, Action onCompleted) =>
-        source.SynchronizeAsync().Subscribe(
+        source.SynchronizeSynchronous().Subscribe(
             async observer =>
             {
                 await onNext(observer.Value);
@@ -948,7 +944,7 @@ public static IDisposable SubscribeSynchronous<T>(this IObservable<T> source, Fu
     /// <param name="onError">Action to invoke upon exceptional termination of the observable sequence.</param>
     /// <returns><see cref="IDisposable"/> object used to unsubscribe from the observable sequence.</returns>
     public static IDisposable SubscribeSynchronous<T>(this IObservable<T> source, Func<T, Task> onNext, Action<Exception> onError) =>
-        source.SynchronizeAsync().Subscribe(
+        source.SynchronizeSynchronous().Subscribe(
             async observer =>
             {
                 await onNext(observer.Value);
@@ -966,7 +962,7 @@ public static IDisposable SubscribeSynchronous<T>(this IObservable<T> source, Fu
     /// <returns><see cref="IDisposable"/> object used to unsubscribe from the observable sequence.</returns>
     /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="onNext"/> or <paramref name="onCompleted"/> is <c>null</c>.</exception>
     public static IDisposable SubscribeSynchronous<T>(this IObservable<T> source, Func<T, Task> onNext, Action onCompleted) =>
-        source.SynchronizeAsync().Subscribe(
+        source.SynchronizeSynchronous().Subscribe(
             async observer =>
             {
                 await onNext(observer.Value);
@@ -982,13 +978,74 @@ public static IDisposable SubscribeSynchronous<T>(this IObservable<T> source, Fu
     /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
     /// <returns><see cref="IDisposable"/> object used to unsubscribe from the observable sequence.</returns>
     public static IDisposable SubscribeSynchronous<T>(this IObservable<T> source, Func<T, Task> onNext) =>
-        source.SynchronizeAsync().Subscribe(
+        source.SynchronizeSynchronous().Subscribe(
              async observer =>
              {
                  await onNext(observer.Value);
                  observer.Sync.Dispose();
              });
 
+    /// <summary>
+    /// Synchronizes the asynchronous operations in downstream operations.
+    /// Use SubscribeSynchronus instead for a simpler version.
+    /// Call Sync.Dispose() to release the lock in the downstream methods.
+    /// </summary>
+    /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
+    /// <param name="source">The source.</param>
+    /// <returns>An Observable of T and a release mechanism.</returns>
+    public static IObservable<(T Value, IDisposable Sync)> SynchronizeAsync<T>(this IObservable<T> source) =>
+        Observable.Create<(T Value, IDisposable Sync)>(observer => source.Select(item => Observable.FromAsync(() => new Continuation().Lock(item, observer))).Concat().Subscribe());
+
+    /// <summary>
+    /// Subscribes allowing asynchronous operations to be executed without blocking the source.
+    /// </summary>
+    /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
+    /// <param name="source">Observable sequence to subscribe to.</param>
+    /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
+    /// <returns><see cref="IDisposable"/> object used to unsubscribe from the observable sequence.</returns>
+    public static IDisposable SubscribeAsync<T>(this IObservable<T> source, Func<T, Task> onNext) =>
+            source.Select(o => Observable.FromAsync(() => onNext(o))).Concat().Subscribe();
+
+    /// <summary>
+    /// Subscribes allowing asynchronous operations to be executed without blocking the source.
+    /// </summary>
+    /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
+    /// <param name="source">Observable sequence to subscribe to.</param>
+    /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
+    /// <param name="onCompleted">The on completed.</param>
+    /// <returns>
+    ///   <see cref="IDisposable" /> object used to unsubscribe from the observable sequence.
+    /// </returns>
+    public static IDisposable SubscribeAsync<T>(this IObservable<T> source, Func<T, Task> onNext, Action onCompleted) =>
+            source.Select(o => Observable.FromAsync(() => onNext(o))).Concat().Subscribe(_ => { }, onCompleted);
+
+    /// <summary>
+    /// Subscribes allowing asynchronous operations to be executed without blocking the source.
+    /// </summary>
+    /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
+    /// <param name="source">Observable sequence to subscribe to.</param>
+    /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
+    /// <param name="onError">The on error.</param>
+    /// <returns>
+    ///   <see cref="IDisposable" /> object used to unsubscribe from the observable sequence.
+    /// </returns>
+    public static IDisposable SubscribeAsync<T>(this IObservable<T> source, Func<T, Task> onNext, Action<Exception> onError) =>
+            source.Select(o => Observable.FromAsync(() => onNext(o))).Concat().Subscribe(_ => { }, onError);
+
+    /// <summary>
+    /// Subscribes allowing asynchronous operations to be executed without blocking the source.
+    /// </summary>
+    /// <typeparam name="T">The type of the elements in the source sequence.</typeparam>
+    /// <param name="source">Observable sequence to subscribe to.</param>
+    /// <param name="onNext">Action to invoke for each element in the observable sequence.</param>
+    /// <param name="onError">The on error.</param>
+    /// <param name="onCompleted">The on completed.</param>
+    /// <returns>
+    ///   <see cref="IDisposable" /> object used to unsubscribe from the observable sequence.
+    /// </returns>
+    public static IDisposable SubscribeAsync<T>(this IObservable<T> source, Func<T, Task> onNext, Action<Exception> onError, Action onCompleted) =>
+            source.Select(o => Observable.FromAsync(() => onNext(o))).Concat().Subscribe(_ => { }, onError, onCompleted);
+
     private static void FastForEach<T>(IObserver<T> observer, IEnumerable<T> source)
     {
         if (source is List<T> fullList)