1 files changed, 40 insertions, 0 deletions
diff --git a/Rx.NET/System.Reactive.Core/Reactive/Internal/AsyncLockObserver.cs b/Rx.NET/System.Reactive.Core/Reactive/Internal/AsyncLockObserver.cs
new file mode 100644
index 0000000..7db2116
--- /dev/null
+++ b/Rx.NET/System.Reactive.Core/Reactive/Internal/AsyncLockObserver.cs
@@ -0,0 +1,40 @@
+using System.Reactive.Concurrency;
+
+namespace System.Reactive
+{
+    internal class AsyncLockObserver<T> : ObserverBase<T>
+    {
+        private readonly AsyncLock _gate;
+        private readonly IObserver<T> _observer;
+
+        public AsyncLockObserver(IObserver<T> observer, AsyncLock gate)
+        {
+            _gate = gate;
+            _observer = observer;
+        }
+
+        protected override void OnNextCore(T value)
+        {
+            _gate.Wait(() =>
+            {
+                _observer.OnNext(value);
+            });
+        }
+
+        protected override void OnErrorCore(Exception exception)
+        {
+            _gate.Wait(() =>
+            {
+                _observer.OnError(exception);
+            });
+        }
+
+        protected override void OnCompletedCore()
+        {
+            _gate.Wait(() =>
+            {
+                _observer.OnCompleted();
+            });
+        }
+    }
+}