2.0.17

Fixed CancelationToken not being passed correctly

README.md

@@ -2,7 +2,7 @@ UniTask
 ===
 [![GitHub Actions](https://github.com/Cysharp/UniTask/workflows/Build-Debug/badge.svg)](https://github.com/Cysharp/UniTask/actions) [![Releases](https://img.shields.io/github/release/Cysharp/UniTask.svg)](https://github.com/Cysharp/UniTask/releases)
 
-Provides an efficient async/await integration to Unity. 
+Provides an efficient allocation free async/await integration to Unity.
 
 * Struct based `UniTask<T>` and custom AsyncMethodBuilder to achive zero allocation
 * All Unity AsyncOperations and Coroutine to awaitable
@@ -13,6 +13,9 @@ Provides an efficient async/await integration to Unity.
 * TaskTracker window to prevent memory leak
 * Highly compatible behaviour with Task/ValueTask/IValueTaskSource
 
+Techinical details, see blog post: [UniTask v2 — Zero Allocation async/await for Unity, with Asynchronous LINQ
+](https://medium.com/@neuecc/unitask-v2-zero-allocation-async-await-for-unity-with-asynchronous-linq-1aa9c96aa7dd)
+
 <!-- START doctoc generated TOC please keep comment here to allow auto update -->
 <!-- DON'T EDIT THIS SECTION, INSTEAD RE-RUN doctoc TO UPDATE -->
 ## Table of Contents

src/UniTask/Assets/Plugins/UniTask/Runtime/Linq/Subscribe.cs

@@ -0,0 +1,263 @@
+using Cysharp.Threading.Tasks.Internal;
+using System;
+using System.Threading;
+using Subscribes = Cysharp.Threading.Tasks.Linq.Subscribe;
+
+namespace Cysharp.Threading.Tasks.Linq
+{
+    public static partial class UniTaskAsyncEnumerable
+    {
+        // OnNext
+
+        public static IDisposable Subscribe<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Action<TSource> action)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+            Error.ThrowArgumentNullException(action, nameof(action));
+
+            var cts = new CancellationTokenDisposable();
+            Subscribes.SubscribeCore(source, action, Subscribes.NopError, Subscribes.NopCompleted, cts.Token).Forget();
+            return cts;
+        }
+
+        public static IDisposable Subscribe<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTaskVoid> action)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+            Error.ThrowArgumentNullException(action, nameof(action));
+
+            var cts = new CancellationTokenDisposable();
+            Subscribes.SubscribeCore(source, action, Subscribes.NopError, Subscribes.NopCompleted, cts.Token).Forget();
+            return cts;
+        }
+
+        public static void Subscribe<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Action<TSource> action, CancellationToken cancellationToken)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+            Error.ThrowArgumentNullException(action, nameof(action));
+
+            Subscribes.SubscribeCore(source, action, Subscribes.NopError, Subscribes.NopCompleted, cancellationToken).Forget();
+        }
+
+        public static void Subscribe<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTaskVoid> action, CancellationToken cancellationToken)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+            Error.ThrowArgumentNullException(action, nameof(action));
+
+            Subscribes.SubscribeCore(source, action, Subscribes.NopError, Subscribes.NopCompleted, cancellationToken).Forget();
+        }
+
+        // OnNext, OnError
+
+        public static IDisposable Subscribe<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Action<TSource> onNext, Action<Exception> onError)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+            Error.ThrowArgumentNullException(onNext, nameof(onNext));
+            Error.ThrowArgumentNullException(onError, nameof(onError));
+
+            var cts = new CancellationTokenDisposable();
+            Subscribes.SubscribeCore(source, onNext, onError, Subscribes.NopCompleted, cts.Token).Forget();
+            return cts;
+        }
+
+        public static IDisposable Subscribe<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTaskVoid> onNext, Action<Exception> onError)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+            Error.ThrowArgumentNullException(onNext, nameof(onNext));
+            Error.ThrowArgumentNullException(onError, nameof(onError));
+
+            var cts = new CancellationTokenDisposable();
+            Subscribes.SubscribeCore(source, onNext, onError, Subscribes.NopCompleted, cts.Token).Forget();
+            return cts;
+        }
+
+        public static void Subscribe<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Action<TSource> onNext, Action<Exception> onError, CancellationToken cancellationToken)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+            Error.ThrowArgumentNullException(onNext, nameof(onNext));
+            Error.ThrowArgumentNullException(onError, nameof(onError));
+
+            Subscribes.SubscribeCore(source, onNext, onError, Subscribes.NopCompleted, cancellationToken).Forget();
+        }
+
+        public static void Subscribe<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTaskVoid> onNext, Action<Exception> onError, CancellationToken cancellationToken)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+            Error.ThrowArgumentNullException(onNext, nameof(onNext));
+            Error.ThrowArgumentNullException(onError, nameof(onError));
+
+            Subscribes.SubscribeCore(source, onNext, onError, Subscribes.NopCompleted, cancellationToken).Forget();
+        }
+
+        // OnNext, OnCompleted
+
+        public static IDisposable Subscribe<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Action<TSource> onNext, Action onCompleted)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+            Error.ThrowArgumentNullException(onNext, nameof(onNext));
+            Error.ThrowArgumentNullException(onCompleted, nameof(onCompleted));
+
+            var cts = new CancellationTokenDisposable();
+            Subscribes.SubscribeCore(source, onNext, Subscribes.NopError, onCompleted, cts.Token).Forget();
+            return cts;
+        }
+
+        public static IDisposable Subscribe<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTaskVoid> onNext, Action onCompleted)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+            Error.ThrowArgumentNullException(onNext, nameof(onNext));
+            Error.ThrowArgumentNullException(onCompleted, nameof(onCompleted));
+
+            var cts = new CancellationTokenDisposable();
+            Subscribes.SubscribeCore(source, onNext, Subscribes.NopError, onCompleted, cts.Token).Forget();
+            return cts;
+        }
+
+        public static void Subscribe<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Action<TSource> onNext, Action onCompleted, CancellationToken cancellationToken)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+            Error.ThrowArgumentNullException(onNext, nameof(onNext));
+            Error.ThrowArgumentNullException(onCompleted, nameof(onCompleted));
+
+            Subscribes.SubscribeCore(source, onNext, Subscribes.NopError, onCompleted, cancellationToken).Forget();
+        }
+
+        public static void Subscribe<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTaskVoid> onNext, Action onCompleted, CancellationToken cancellationToken)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+            Error.ThrowArgumentNullException(onNext, nameof(onNext));
+            Error.ThrowArgumentNullException(onCompleted, nameof(onCompleted));
+
+            Subscribes.SubscribeCore(source, onNext, Subscribes.NopError, onCompleted, cancellationToken).Forget();
+        }
+
+        // IObserver
+
+        public static IDisposable Subscribe<TSource>(this IUniTaskAsyncEnumerable<TSource> source, IObserver<TSource> observer)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+            Error.ThrowArgumentNullException(observer, nameof(observer));
+
+            var cts = new CancellationTokenDisposable();
+            Subscribes.SubscribeCore(source, observer, cts.Token).Forget();
+            return cts;
+        }
+
+        public static void Subscribe<TSource>(this IUniTaskAsyncEnumerable<TSource> source, IObserver<TSource> observer, CancellationToken cancellationToken)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+            Error.ThrowArgumentNullException(observer, nameof(observer));
+
+            Subscribes.SubscribeCore(source, observer, cancellationToken).Forget();
+        }
+    }
+
+    internal sealed class CancellationTokenDisposable : IDisposable
+    {
+        readonly CancellationTokenSource cts = new CancellationTokenSource();
+
+        public CancellationToken Token => cts.Token;
+
+        public void Dispose()
+        {
+            if (!cts.IsCancellationRequested)
+            {
+                cts.Cancel();
+            }
+        }
+    }
+
+    internal static class Subscribe
+    {
+        public static readonly Action<Exception> NopError = _ => { };
+        public static readonly Action NopCompleted = () => { };
+
+        public static async UniTaskVoid SubscribeCore<TSource>(IUniTaskAsyncEnumerable<TSource> source, Action<TSource> onNext, Action<Exception> onError, Action onCompleted, CancellationToken cancellationToken)
+        {
+            var e = source.GetAsyncEnumerator(cancellationToken);
+            try
+            {
+                while (await e.MoveNextAsync())
+                {
+                    onNext(e.Current);
+                }
+                onCompleted();
+            }
+            catch (Exception ex)
+            {
+                if (onError == NopError)
+                {
+                    UniTaskScheduler.PublishUnobservedTaskException(ex);
+                    return;
+                }
+
+                if (ex is OperationCanceledException) return;
+
+                onError(ex);
+            }
+            finally
+            {
+                if (e != null)
+                {
+                    await e.DisposeAsync();
+                }
+            }
+        }
+
+        public static async UniTaskVoid SubscribeCore<TSource>(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTaskVoid> onNext, Action<Exception> onError, Action onCompleted, CancellationToken cancellationToken)
+        {
+            var e = source.GetAsyncEnumerator(cancellationToken);
+            try
+            {
+                while (await e.MoveNextAsync())
+                {
+                    onNext(e.Current).Forget();
+                }
+                onCompleted();
+            }
+            catch (Exception ex)
+            {
+                if (onError == NopError)
+                {
+                    UniTaskScheduler.PublishUnobservedTaskException(ex);
+                    return;
+                }
+
+                if (ex is OperationCanceledException) return;
+
+                onError(ex);
+            }
+            finally
+            {
+                if (e != null)
+                {
+                    await e.DisposeAsync();
+                }
+            }
+        }
+
+        public static async UniTaskVoid SubscribeCore<TSource>(IUniTaskAsyncEnumerable<TSource> source, IObserver<TSource> observer, CancellationToken cancellationToken)
+        {
+            var e = source.GetAsyncEnumerator(cancellationToken);
+            try
+            {
+                while (await e.MoveNextAsync())
+                {
+                    observer.OnNext(e.Current);
+                }
+                observer.OnCompleted();
+            }
+            catch (Exception ex)
+            {
+                if (ex is OperationCanceledException) return;
+
+                observer.OnError(ex);
+            }
+            finally
+            {
+                if (e != null)
+                {
+                    await e.DisposeAsync();
+                }
+            }
+        }
+    }
+}

src/UniTask/Assets/Plugins/UniTask/Runtime/Linq/ToUniTaskAsyncEnumerable.cs

@@ -248,15 +248,18 @@ namespace Cysharp.Threading.Tasks.Linq
                         return current;
                     }
 
-                    if (queuedResult.Count != 0)
+                    lock (queuedResult)
                     {
-                        current = queuedResult.Dequeue();
-                        useCachedCurrent = true;
-                        return current;
-                    }
-                    else
-                    {
-                        return default; // undefined.
+                        if (queuedResult.Count != 0)
+                        {
+                            current = queuedResult.Dequeue();
+                            useCachedCurrent = true;
+                            return current;
+                        }
+                        else
+                        {
+                            return default; // undefined.
+                        }
                     }
                 }
             }

src/UniTask/Assets/Plugins/UniTask/Runtime/Linq/UnityExtensions/EveryUpdate.cs

@@ -34,6 +34,7 @@ namespace Cysharp.Threading.Tasks.Linq
             public _EveryUpdate(PlayerLoopTiming updateTiming, CancellationToken cancellationToken)
             {
                 this.updateTiming = updateTiming;
+                this.cancellationToken = cancellationToken;
 
                 TaskTracker.TrackActiveTask(this, 2);
                 PlayerLoopHelper.AddAction(updateTiming, this);

src/UniTask/Assets/Plugins/UniTask/Runtime/Linq/UnityExtensions/Timer.cs

@@ -104,6 +104,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.dueTimePhase = true;
                 this.updateTiming = updateTiming;
                 this.ignoreTimeScale = ignoreTimeScale;
+                this.cancellationToken = cancellationToken;
                 TaskTracker.TrackActiveTask(this, 2);
                 PlayerLoopHelper.AddAction(updateTiming, this);
             }
@@ -223,6 +224,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.dueTimePhase = true;
                 this.dueTimeFrameCount = dueTimeFrameCount;
                 this.periodFrameCount = periodFrameCount;
+                this.cancellationToken = cancellationToken;
 
                 TaskTracker.TrackActiveTask(this, 2);
                 PlayerLoopHelper.AddAction(updateTiming, this);

src/UniTask/Assets/Plugins/UniTask/package.json

@@ -1,7 +1,7 @@
 {
     "name": "com.cysharp.unitask",
     "displayName": "UniTask",
-    "version": "2.0.16",
+    "version": "2.0.17",
     "unity": "2018.4",
     "description": "Provides an efficient async/await integration to Unity.",
     "keywords": [ "async/await", "async", "Task", "UniTask" ],