From 3e43324de528a7250d2ad4998a9492afa54b1083 Mon Sep 17 00:00:00 2001
From: Roland Pheasant <roland_pheasant@hotmail.com>
Date: Mon, 7 Mar 2022 18:29:14 +0000
Subject: [PATCH] Add appropriate locking syntax for TransformAsync

---
 .../Cache/Internal/TransformAsync.cs          | 36 ++++++++++++++++---
 version.json                                  |  2 +-
 2 files changed, 32 insertions(+), 6 deletions(-)

diff --git a/src/DynamicData/Cache/Internal/TransformAsync.cs b/src/DynamicData/Cache/Internal/TransformAsync.cs
index 6e863c622..123f404d9 100644
--- a/src/DynamicData/Cache/Internal/TransformAsync.cs
+++ b/src/DynamicData/Cache/Internal/TransformAsync.cs
@@ -5,6 +5,7 @@
 using System;
 using System.Linq;
 using System.Reactive.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 
 using DynamicData.Kernel;
@@ -32,16 +33,41 @@ public TransformAsync(IObservable<IChangeSet<TSource, TKey>> source, Func<TSourc
 
         public IObservable<IChangeSet<TDestination, TKey>> Run()
         {
-            return Observable.Create<IChangeSet<TDestination, TKey>>(
-                observer =>
+            return Observable.Create<IChangeSet<TDestination, TKey>>(observer =>
                     {
                         var cache = new ChangeAwareCache<TransformedItemContainer, TKey>();
-                        var transformer = _source.SelectMany(changes => DoTransform(cache, changes));
+                        var asyncLock = new SemaphoreSlim(1, 1);
+
+                        var transformer = _source.SelectMany(async changes =>
+                            {
+                                try
+                                {
+                                    await asyncLock.WaitAsync();
+                                    return await DoTransform(cache, changes).ConfigureAwait(false);
+
+                                }
+                                finally
+                                {
+                                    asyncLock.Release();
+                                }
+                            });
 
                         if (_forceTransform is not null)
                         {
                             var locker = new object();
-                            var forced = _forceTransform.Synchronize(locker).SelectMany(shouldTransform => DoTransform(cache, shouldTransform));
+                            var forced = _forceTransform.Synchronize(locker).SelectMany(async shouldTransform =>
+                            {
+                                try
+                                {
+                                    await asyncLock.WaitAsync();
+                                    return await DoTransform(cache, shouldTransform).ConfigureAwait(false);
+
+                                }
+                                finally
+                                {
+                                    asyncLock.Release();
+                                }
+                            });
 
                             transformer = transformer.Synchronize(locker).Merge(forced);
                         }
@@ -175,4 +201,4 @@ public TransformResult(Change<TSource, TKey> change, Exception error)
             public bool Success { get; }
         }
     }
-}
\ No newline at end of file
+}
diff --git a/version.json b/version.json
index 6f06f01c5..31bc57b7e 100644
--- a/version.json
+++ b/version.json
@@ -1,5 +1,5 @@
 {
-  "version": "7.5",
+    "version": "7.5.1",
   "publicReleaseRefSpec": [
     "^refs/heads/main$", // we release out of master
     "^refs/heads/preview/.*", // we release previews