Added a new overload for .Synchronize() utilizing the new .NET 9 Lock class.

Author: JakenVeina

Rx.NET/Source/Directory.Build.targets

@@ -21,6 +21,9 @@
   <PropertyGroup Condition="$(TargetFramework.StartsWith('net6.0-windows')) or $(TargetFramework.StartsWith('net8.0-windows')) or $(TargetFramework.StartsWith('net9.0-windows'))">
     <DefineConstants>$(DefineConstants);HAS_WINRT;HAS_WINFORMS;HAS_WPF;HAS_DISPATCHER;DESKTOPCLR;WINDOWS;CSWINRT</DefineConstants>
   </PropertyGroup>
+  <PropertyGroup Condition="$(TargetFramework.StartsWith('net9.0'))">
+    <DefineConstants>$(DefineConstants);HAS_LOCK_CLASS</DefineConstants>
+  </PropertyGroup>
 
   <ItemGroup Condition="('$(TargetFramework)' == 'net472' or '$(TargetFramework)' == 'uap10.0.18362' or '$(TargetFramework)' == 'netstandard2.0') and $(IsPackable)">
     <PackageReference Include="System.Threading.Tasks.Extensions" Version="4.5.4" />

Rx.NET/Source/src/System.Reactive/Concurrency/Synchronization.Synchronize.cs

@@ -1,15 +1,16 @@
 // 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. 
+// See the LICENSE file in the project root for more information.
 
 namespace System.Reactive.Concurrency
 {
-    internal sealed class Synchronize<TSource> : Producer<TSource, Synchronize<TSource>._>
+    internal sealed class Synchronize<TSource, TGate> : Producer<TSource, Synchronize<TSource, TGate>._>
+        where TGate : notnull, new()
     {
         private readonly IObservable<TSource> _source;
-        private readonly object? _gate;
+        private readonly TGate? _gate;
 
-        public Synchronize(IObservable<TSource> source, object gate)
+        public Synchronize(IObservable<TSource> source, TGate gate)
         {
             _source = source;
             _gate = gate;
@@ -26,12 +27,12 @@ public Synchronize(IObservable<TSource> source)
 
         internal sealed class _ : IdentitySink<TSource>
         {
-            private readonly object _gate;
+            private readonly TGate _gate;
 
-            public _(Synchronize<TSource> parent, IObserver<TSource> observer)
+            public _(Synchronize<TSource, TGate> parent, IObserver<TSource> observer)
                 : base(observer)
             {
-                _gate = parent._gate ?? new object();
+                _gate = parent._gate ?? new TGate();
             }
 
             public override void OnNext(TSource value)

Rx.NET/Source/src/System.Reactive/Concurrency/Synchronization.cs

@@ -1,6 +1,6 @@
 // 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. 
+// See the LICENSE file in the project root for more information.
 
 using System.ComponentModel;
 using System.Reactive.Disposables;
@@ -229,7 +229,7 @@ public static IObservable<TSource> Synchronize<TSource>(IObservable<TSource> sou
                 throw new ArgumentNullException(nameof(source));
             }
 
-            return new Synchronize<TSource>(source);
+            return new Synchronize<TSource, object>(source);
         }
 
         /// <summary>
@@ -252,9 +252,34 @@ public static IObservable<TSource> Synchronize<TSource>(IObservable<TSource> sou
                 throw new ArgumentNullException(nameof(gate));
             }
 
-            return new Synchronize<TSource>(source, gate);
+            return new Synchronize<TSource, object>(source, gate);
         }
 
+        #if HAS_LOCK_CLASS
+        /// <summary>
+        /// Wraps the source sequence in order to ensure observer callbacks are synchronized using the specified gate object.
+        /// </summary>
+        /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
+        /// <param name="source">Source sequence.</param>
+        /// <param name="gate">Gate object to synchronize each observer call on.</param>
+        /// <returns>The source sequence whose outgoing calls to observers are synchronized on the given gate object.</returns>
+        /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="gate"/> is <c>null</c>.</exception>
+        public static IObservable<TSource> Synchronize<TSource>(IObservable<TSource> source, Lock gate)
+        {
+            if (source == null)
+            {
+                throw new ArgumentNullException(nameof(source));
+            }
+
+            if (gate == null)
+            {
+                throw new ArgumentNullException(nameof(gate));
+            }
+
+            return new Synchronize<TSource, Lock>(source, gate);
+        }
+        #endif
+
         #endregion
     }
 }

Rx.NET/Source/tests/Tests.System.Reactive/Tests/Concurrency/SynchronizationTest.cs

@@ -1,6 +1,6 @@
 // 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. 
+// See the LICENSE file in the project root for more information.
 
 using System;
 using System.Reactive.Concurrency;
@@ -37,7 +37,10 @@ public void Synchronization_Synchronize_ArgumentChecking()
         {
             ReactiveAssert.Throws<ArgumentNullException>(() => Synchronization.Synchronize(default(IObservable<int>)));
             ReactiveAssert.Throws<ArgumentNullException>(() => Synchronization.Synchronize(default(IObservable<int>), new object()));
-            ReactiveAssert.Throws<ArgumentNullException>(() => Synchronization.Synchronize(DummyObservable<int>.Instance, null));
+            ReactiveAssert.Throws<ArgumentNullException>(() => Synchronization.Synchronize(DummyObservable<int>.Instance, null as object));
+            #if HAS_LOCK_CLASS
+            ReactiveAssert.Throws<ArgumentNullException>(() => Synchronization.Synchronize(DummyObservable<int>.Instance, null as Lock));
+            #endif
         }
 
         private class MySyncCtx : SynchronizationContext