Enable application-defined Synchronize gate in AsyncRx.NET

AsyncRx.NET/System.Reactive.Async/Linq/Operators/Synchronize.cs

@@ -2,6 +2,7 @@
 // The .NET Foundation licenses this file to you under the MIT License.
 // See the LICENSE file in the project root for more information. 
 
+using System.Reactive.Threading;
 using System.Threading;
 
 namespace System.Reactive.Linq
@@ -16,7 +17,7 @@ public static IAsyncObservable<TSource> Synchronize<TSource>(this IAsyncObservab
             return Create(source, static (source, observer) => source.SubscribeSafeAsync(AsyncObserver.Synchronize(observer)));
         }
 
-        public static IAsyncObservable<TSource> Synchronize<TSource>(this IAsyncObservable<TSource> source, AsyncGate gate)
+        public static IAsyncObservable<TSource> Synchronize<TSource>(this IAsyncObservable<TSource> source, IAsyncGate gate)
         {
             if (source == null)
                 throw new ArgumentNullException(nameof(source));
@@ -40,7 +41,7 @@ public static IAsyncObserver<TSource> Synchronize<TSource>(IAsyncObserver<TSourc
             return Synchronize(observer, new AsyncGate());
         }
 
-        public static IAsyncObserver<TSource> Synchronize<TSource>(IAsyncObserver<TSource> observer, AsyncGate gate)
+        public static IAsyncObserver<TSource> Synchronize<TSource>(IAsyncObserver<TSource> observer, IAsyncGate gate)
         {
             if (observer == null)
                 throw new ArgumentNullException(nameof(observer));

AsyncRx.NET/System.Reactive.Async/Threading/AsyncGate.cs

@@ -3,17 +3,18 @@
 // See the LICENSE file in the project root for more information. 
 
 using System.Diagnostics;
+using System.Reactive.Threading;
 using System.Threading.Tasks;
 
 namespace System.Threading
 {
-    public sealed class AsyncGate
+    public sealed class AsyncGate : IAsyncGate
     {
         private readonly object _gate = new();
         private readonly SemaphoreSlim _semaphore = new(1, 1);
         private readonly AsyncLocal<int> _recursionCount = new();
 
-        public ValueTask<Releaser> LockAsync()
+        public ValueTask<AsyncGateReleaser> LockAsync()
         {
             var shouldAcquire = false;
 
@@ -32,13 +33,13 @@ public ValueTask<Releaser> LockAsync()
 
             if (shouldAcquire)
             {
-                return new ValueTask<Releaser>(_semaphore.WaitAsync().ContinueWith(_ => new Releaser(this)));
+                return new ValueTask<AsyncGateReleaser>(_semaphore.WaitAsync().ContinueWith(_ => new AsyncGateReleaser(this)));
             }
 
-            return new ValueTask<Releaser>(new Releaser(this));
+            return new ValueTask<AsyncGateReleaser>(new AsyncGateReleaser(this));
         }
 
-        private void Release()
+        void IAsyncGate.Release()
         {
             lock (_gate)
             {
@@ -50,14 +51,5 @@ private void Release()
                 }
             }
         }
-
-        public readonly struct Releaser : IDisposable
-        {
-            private readonly AsyncGate _parent;
-
-            public Releaser(AsyncGate parent) => _parent = parent;
-
-            public void Dispose() => _parent.Release();
-        }
     }
 }

AsyncRx.NET/System.Reactive.Async/Threading/AsyncGateReleaser.cs

@@ -0,0 +1,15 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT License.
+// See the LICENSE file in the project root for more information. 
+
+namespace System.Reactive.Threading
+{
+    public readonly struct AsyncGateReleaser : IDisposable
+    {
+        private readonly IAsyncGate _parent;
+
+        public AsyncGateReleaser(IAsyncGate parent) => _parent = parent;
+
+        public void Dispose() => _parent.Release();
+    }
+}

AsyncRx.NET/System.Reactive.Async/Threading/IAsyncGate.cs

@@ -0,0 +1,73 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT License.
+// See the LICENSE file in the project root for more information. 
+
+using System.Reactive.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace System.Reactive.Threading
+{
+    /// <summary>
+    /// Synchronization primitive that provides <see cref="System.Threading.Monitor"/>-style
+    /// exclusive access semantics, but with an asynchronous API.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// This enables <see cref="AsyncObservable.Synchronize{TSource}(IAsyncObservable{TSource}, IAsyncGate)"/>
+    /// and <see cref="AsyncObserver.Synchronize{TSource}(IAsyncObserver{TSource}, IAsyncGate)"/>
+    /// to be used to synchronize access to an observer with a custom synchronization primitive.
+    /// </para>
+    /// <para>
+    /// These methods model the equivalents for <see cref="IObservable{T}"/> and <see cref="IObserver{T}"/>
+    /// in <c>System.Reactive</c>. Those offer overloads accepting a 'gate' parameter, and if you pass
+    /// the same object to multiple calls to these methods, they will all synchronize their operation
+    /// through that same gate object. The <c>gate</c> parameter in those methods is of type
+    /// <see cref="System.Object"/>, which works because all .NET objects have an associated monitor.
+    /// (It's created on demand when you first use <c>lock</c> or something equivalent.)
+    /// </para>
+    /// <para>
+    /// That approach is problematic in an async world, because this built-in monitor blocks the
+    /// calling thread when contention occurs. The basic idea of AsyncRx.NET is to avoid such
+    /// blocking. It can't always be avoided, and in cases where we can be certain that lock
+    /// acquisition times will be short, the conventional .NET monitor is still a good choice.
+    /// But since these <c>Synchronize</c> operators allow the caller to pass a gate which the
+    /// application code itself might lock, we have no control over how long the lock might be
+    /// held. So it would be inappropriate to use a monitor here.
+    /// </para>
+    /// <para>
+    /// Since the .NET runtime does not currently offer any asynchronous direct equivalent to
+    /// monitor, this interface defines the required API. The <see cref="AsyncGate"/> class
+    /// provide a basic implementation. If applications require additional features, (e.g.
+    /// if they want cancellation support when the application tries to acquire the lock)
+    /// they can provide their own implementation.
+    /// </para>
+    /// </remarks>
+    public interface IAsyncGate
+    {
+        /// <summary>
+        /// Acquires the lock.
+        /// </summary>
+        /// <returns>
+        /// A task that completes when the lock has been acquired, returning an <see cref="AsyncGateReleaser"/>
+        /// which can be disposed to release the lock.
+        /// </returns>
+        /// <remarks>
+        /// <para>
+        /// Applications release the lock by disposing the <see cref="AsyncGateReleaser"/> returned by this
+        /// method. Typically this is done with a <c>using</c> statement or declaration.
+        /// </para>
+        /// </remarks>
+        public ValueTask<AsyncGateReleaser> LockAsync();
+
+        /// <summary>
+        /// Releases the lock. Applications typically won't call this directly, and will use
+        /// the <see cref="AsyncGateReleaser"/> returned by <see cref="LockAsync"/> instead.
+        /// </summary>
+        /// <remarks>
+        /// This method needs to be publicly accessible so that a single <see cref="AsyncGateReleaser"/>
+        /// can be shared by all implementations of this interface.
+        /// </remarks>
+        public void Release();
+    }
+}