2.0.9-rc6

README.md

@@ -1,29 +1,30 @@
-<!-- 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
-
-- [UniTask](#unitask)
-  - [Getting started](#getting-started)
-  - [`UniTask<T>`](#unitaskt)
-  - [Cancellation and Exception handling](#cancellation-and-exception-handling)
-  - [Progress](#progress)
-  - [UniTaskTracker](#unitasktracker)
-  - [Reusable Promises](#reusable-promises)
-  - [awaitable Events](#awaitable-events)
-  - [async void vs async UniTask/UniTaskVoid](#async-void-vs-async-unitaskunitaskvoid)
-  - [For Unit Testing](#for-unit-testing)
-  - [Method List](#method-list)
-  - [UPM Package](#upm-package)
-  - [License](#license)
-
-<!-- END doctoc generated TOC please keep comment here to allow auto update -->
-
 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.
 
+<!-- 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
+
+- [Getting started](#getting-started)
+- [`UniTask<T>`](#unitaskt)
+- [Cancellation and Exception handling](#cancellation-and-exception-handling)
+- [Progress](#progress)
+- [UniTaskTracker](#unitasktracker)
+- [Reusable Promises](#reusable-promises)
+- [awaitable Events](#awaitable-events)
+- [async void vs async UniTask/UniTaskVoid](#async-void-vs-async-unitaskunitaskvoid)
+- [For Unit Testing](#for-unit-testing)
+- [Method List](#method-list)
+- [UPM Package](#upm-package)
+- [ECS, PlayerLoop](#ecs-playerloop)
+- [License](#license)
+
+<!-- END doctoc generated TOC please keep comment here to allow auto update -->
+
+
 Getting started
 ---
 Install package(`UniRx.Async.unitypackage`) is available in [UniTask/releases](https://github.com/Cysharp/UniTask/releases) page.
@@ -372,6 +373,15 @@ or add `"com.cysharp.unitask": "https://github.com/Cysharp/UniTask.git?path=Asse
 
 If you want to set a target version, UniTask is using `*.*.*` release tag so you can specify a version like `#1.3.0`. For example `https://github.com/Cysharp/UniTask.git?path=Assets/UniRx.Async#1.3.1`.
 
+ECS, PlayerLoop
+---
+TODO:
+
+```csharp
+var loop = PlayerLoop.GetCurrentPlayerLoop();
+PlayerLoopHelper.Initialize(ref loop);
+```
+
 License
 ---
 This library is under the MIT License.

src/UniTask.NetCoreSandbox/Program.cs

@@ -93,10 +93,16 @@ namespace NetCoreSandbox
 
             var channel = Channel.CreateSingleConsumerUnbounded<int>();
 
+            // Observable.Range(1,10).CombineLatest(
+
+            var cts = new CancellationTokenSource();
+
+            var token = cts.Token;
+
+            FooAsync(token).ForEachAsync(x => { }, token);
 
 
-
-
+            // Observable.Range(1,10).CombineLatest(
 
         }
 

src/UniTask.NetCoreTests/AsyncReactivePropertyTest.cs

@@ -51,6 +51,67 @@ namespace NetCoreTests
 
             ar.Should().BeEquivalentTo(new[] { 100, 100, 100, 131, 191 });
         }
+
+        //[Fact]
+        //public async Task StateIteration()
+        //{
+        //    var rp = new ReadOnlyAsyncReactiveProperty<int>(99);
+        //    var setter = rp.GetSetter();
+
+        //    var f = await rp.FirstAsync();
+        //    f.Should().Be(99);
+
+        //    var array = rp.Take(5).ToArrayAsync();
+
+        //    setter(100);
+        //    setter(100);
+        //    setter(100);
+        //    setter(131);
+
+        //    var ar = await array;
+
+        //    ar.Should().BeEquivalentTo(new[] { 99, 100, 100, 100, 131 });
+        //}
+
+        //[Fact]
+        //public async Task StateWithoutCurrent()
+        //{
+        //    var rp = new ReadOnlyAsyncReactiveProperty<int>(99);
+        //    var setter = rp.GetSetter();
+
+        //    var array = rp.WithoutCurrent().Take(5).ToArrayAsync();
+        //    setter(100);
+        //    setter(100);
+        //    setter(100);
+        //    setter(131);
+        //    setter(191);
+
+        //    var ar = await array;
+
+        //    ar.Should().BeEquivalentTo(new[] { 100, 100, 100, 131, 191 });
+        //}
+
+
+
+        [Fact]
+        public void StateFromEnumeration()
+        {
+            var rp = new AsyncReactiveProperty<int>(10);
+
+            var state = rp.ToReadOnlyAsyncReactiveProperty(CancellationToken.None);
+
+            rp.Value = 10;
+            state.Value.Should().Be(10);
+
+            rp.Value = 20;
+            state.Value.Should().Be(20);
+
+            state.Dispose();
+
+            rp.Value = 30;
+            state.Value.Should().Be(20);
+        }
+
     }
 
 

src/UniTask.NetCoreTests/CancellationTokenTest.cs

@@ -0,0 +1,51 @@
+using Cysharp.Threading.Tasks;
+using FluentAssertions;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Channels;
+using Cysharp.Threading.Tasks.Linq;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace NetCoreTests
+{
+    public class CancellationTokenTest
+    {
+        [Fact]
+        public async Task WaitUntilCanceled()
+        {
+            var cts = new CancellationTokenSource();
+
+            cts.CancelAfter(TimeSpan.FromSeconds(1.5));
+
+            var now = DateTime.UtcNow;
+
+            await cts.Token.WaitUntilCanceled();
+
+            var elapsed = DateTime.UtcNow - now;
+
+            elapsed.Should().BeGreaterThan(TimeSpan.FromSeconds(1));
+        }
+
+        [Fact]
+        public void AlreadyCanceled()
+        {
+            var cts = new CancellationTokenSource();
+
+            cts.Cancel();
+
+            cts.Token.WaitUntilCanceled().GetAwaiter().IsCompleted.Should().BeTrue();
+        }
+
+        [Fact]
+        public void None()
+        {
+            CancellationToken.None.WaitUntilCanceled().GetAwaiter().IsCompleted.Should().BeTrue();
+        }
+    }
+
+
+}

src/UniTask.NetCoreTests/DeferTest.cs

@@ -0,0 +1,47 @@
+using Cysharp.Threading.Tasks;
+using FluentAssertions;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Channels;
+using Cysharp.Threading.Tasks.Linq;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace NetCoreTests
+{
+    public class DeferTest
+    {
+        [Fact]
+        public async Task D()
+        {
+            var created = false;
+            var v = UniTask.Defer(() => { created = true; return UniTask.Run(() => 10); });
+
+            created.Should().BeFalse();
+
+            var t = await v;
+
+            created.Should().BeTrue();
+
+            t.Should().Be(10);
+        }
+
+        [Fact]
+        public async Task D2()
+        {
+            var created = false;
+            var v = UniTask.Defer(() => { created = true; return UniTask.Run(() => 10).AsUniTask(); });
+
+            created.Should().BeFalse();
+
+            await v;
+
+            created.Should().BeTrue();
+        }
+    }
+
+
+}

src/UniTask.NetCoreTests/Linq/Projection.cs

@@ -319,5 +319,119 @@ namespace NetCoreTests.Linq
                 await Assert.ThrowsAsync<UniTaskTestException>(async () => await ys);
             }
         }
+
+        [Fact]
+        public async Task CombineLatestOK()
+        {
+            var a = new AsyncReactiveProperty<int>(0);
+            var b = new AsyncReactiveProperty<int>(0);
+
+            var list = new List<(int, int)>();
+            var complete = a.WithoutCurrent().CombineLatest(b.WithoutCurrent(), (x, y) => (x, y)).ForEachAsync(x => list.Add(x));
+
+            list.Count.Should().Be(0);
+
+            a.Value = 10;
+            list.Count.Should().Be(0);
+
+            a.Value = 20;
+            list.Count.Should().Be(0);
+
+            b.Value = 1;
+            list.Count.Should().Be(1);
+
+            list[0].Should().Be((20, 1));
+
+            a.Value = 30;
+            list.Last().Should().Be((30, 1));
+
+            b.Value = 2;
+            list.Last().Should().Be((30, 2));
+
+            a.Dispose();
+            b.Value = 3;
+            list.Last().Should().Be((30, 3));
+
+            b.Dispose();
+
+            await complete;
+        }
+
+        [Fact]
+        public async Task CombineLatestLong()
+        {
+            var a = UniTaskAsyncEnumerable.Range(1, 100000);
+            var b = new AsyncReactiveProperty<int>(0);
+
+            var list = new List<(int, int)>();
+            var complete = a.CombineLatest(b.WithoutCurrent(), (x, y) => (x, y)).ForEachAsync(x => list.Add(x));
+
+            b.Value = 1;
+
+            list[0].Should().Be((100000, 1));
+
+            b.Dispose();
+
+            await complete;
+        }
+
+        [Fact]
+        public async Task CombineLatestError()
+        {
+            var a = new AsyncReactiveProperty<int>(0);
+            var b = new AsyncReactiveProperty<int>(0);
+
+            var list = new List<(int, int)>();
+            var complete = a.WithoutCurrent()
+                .Select(x => { if (x == 0) { throw new MyException(); } return x; })
+                .CombineLatest(b.WithoutCurrent(), (x, y) => (x, y)).ForEachAsync(x => list.Add(x));
+
+
+            a.Value = 10;
+            b.Value = 1;
+            list.Last().Should().Be((10, 1));
+
+            a.Value = 0;
+
+            await Assert.ThrowsAsync<MyException>(async () => await complete);
+        }
+
+
+
+        [Fact]
+        public async Task PariwiseImmediate()
+        {
+            var xs = await UniTaskAsyncEnumerable.Range(1, 5).Pairwise().ToArrayAsync();
+            xs.Should().BeEquivalentTo((1, 2), (2, 3), (3, 4), (4, 5));
+        }
+
+        [Fact]
+        public async Task Pariwise()
+        {
+            var a = new AsyncReactiveProperty<int>(0);
+
+            var list = new List<(int, int)>();
+            var complete = a.WithoutCurrent().Pairwise().ForEachAsync(x => list.Add(x));
+
+            list.Count.Should().Be(0);
+            a.Value = 10;
+            list.Count.Should().Be(0);
+            a.Value = 20;
+            list.Count.Should().Be(1);
+            a.Value = 30;
+            a.Value = 40;
+            a.Value = 50;
+
+            a.Dispose();
+
+            await complete;
+
+            list.Should().BeEquivalentTo((10, 20), (20, 30), (30, 40), (40, 50));
+        }
+
+        class MyException : Exception
+        {
+
+        }
     }
 }

src/UniTask.NetCoreTests/Linq/PulbishTest.cs

@@ -0,0 +1,78 @@
+using Cysharp.Threading.Tasks;
+using Cysharp.Threading.Tasks.Linq;
+using FluentAssertions;
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace NetCoreTests.Linq
+{
+    public class PublishTest
+    {
+        [Fact]
+        public async Task Normal()
+        {
+            var rp = new AsyncReactiveProperty<int>(1);
+
+            var multicast = rp.Publish();
+
+            var a = multicast.ToArrayAsync();
+            var b = multicast.Take(2).ToArrayAsync();
+
+            var disp = multicast.Connect();
+
+            rp.Value = 2;
+
+            (await b).Should().BeEquivalentTo(1, 2);
+
+            var c = multicast.ToArrayAsync();
+
+            rp.Value = 3;
+            rp.Value = 4;
+            rp.Value = 5;
+
+            rp.Dispose();
+
+            (await a).Should().BeEquivalentTo(1, 2, 3, 4, 5);
+            (await c).Should().BeEquivalentTo(3, 4, 5);
+
+            disp.Dispose();
+        }
+
+        [Fact]
+        public async Task Cancel()
+        {
+            var rp = new AsyncReactiveProperty<int>(1);
+
+            var multicast = rp.Publish();
+
+            var a = multicast.ToArrayAsync();
+            var b = multicast.Take(2).ToArrayAsync();
+
+            var disp = multicast.Connect();
+
+            rp.Value = 2;
+
+            (await b).Should().BeEquivalentTo(1, 2);
+
+            var c = multicast.ToArrayAsync();
+
+            rp.Value = 3;
+
+            disp.Dispose();
+
+            rp.Value = 4;
+            rp.Value = 5;
+
+            rp.Dispose();
+
+            await Assert.ThrowsAsync<OperationCanceledException>(async () => await a);
+            await Assert.ThrowsAsync<OperationCanceledException>(async () => await c);
+        }
+
+
+    }
+}

src/UniTask.NetCoreTests/Linq/TakeInfinityTest.cs

@@ -4,6 +4,7 @@ using FluentAssertions;
 using System;
 using System.Collections.Generic;
 using System.Text;
+using System.Threading;
 using System.Threading.Tasks;
 using Xunit;
 
@@ -40,5 +41,66 @@ namespace NetCoreTests.Linq
 
             (await xs).Should().BeEquivalentTo(1, 2, 3, 4);
         }
+
+        [Fact]
+        public async Task TakeUntil()
+        {
+            var cts = new CancellationTokenSource();
+
+            var rp = new AsyncReactiveProperty<int>(1);
+
+            var xs = rp.TakeUntilCanceled(cts.Token).ToArrayAsync();
+
+            var c = CancelAsync();
+
+            await c;
+            var foo = await xs;
+
+            foo.Should().BeEquivalentTo(new[] { 1, 10, 20 });
+
+            async Task CancelAsync()
+            {
+                rp.Value = 10;
+                await Task.Yield();
+                rp.Value = 20;
+                await Task.Yield();
+                cts.Cancel();
+                rp.Value = 30;
+                await Task.Yield();
+                rp.Value = 40;
+            }
+        }
+
+        [Fact]
+        public async Task SkipUntil()
+        {
+            var cts = new CancellationTokenSource();
+
+            var rp = new AsyncReactiveProperty<int>(1);
+
+            var xs = rp.SkipUntilCanceled(cts.Token).ToArrayAsync();
+
+            var c = CancelAsync();
+
+            await c;
+            var foo = await xs;
+
+            foo.Should().BeEquivalentTo(new[] { 30, 40 });
+
+            async Task CancelAsync()
+            {
+                rp.Value = 10;
+                await Task.Yield();
+                rp.Value = 20;
+                await Task.Yield();
+                cts.Cancel();
+                rp.Value = 30;
+                await Task.Yield();
+                rp.Value = 40;
+
+                rp.Dispose(); // complete.
+            }
+        }
+
     }
 }

src/UniTask/Assets/Plugins/UniTask/Runtime/AsyncReactiveProperty.cs

@@ -4,12 +4,13 @@ using System.Threading;
 
 namespace Cysharp.Threading.Tasks
 {
-    public interface IAsyncReadOnlyReactiveProperty<T> : IUniTaskAsyncEnumerable<T>
+    public interface IReadOnlyAsyncReactiveProperty<T> : IUniTaskAsyncEnumerable<T>
     {
         T Value { get; }
+        IUniTaskAsyncEnumerable<T> WithoutCurrent();
     }
 
-    public interface IAsyncReactiveProperty<T> : IAsyncReadOnlyReactiveProperty<T>
+    public interface IAsyncReactiveProperty<T> : IReadOnlyAsyncReactiveProperty<T>
     {
         new T Value { get; set; }
     }
@@ -55,7 +56,25 @@ namespace Cysharp.Threading.Tasks
 
         public void Dispose()
         {
-            triggerEvent.SetCanceled(CancellationToken.None);
+            triggerEvent.SetCompleted();
+        }
+
+        public static implicit operator T(AsyncReactiveProperty<T> value)
+        {
+            return value.Value;
+        }
+
+        public override string ToString()
+        {
+            if (isValueType) return latestValue.ToString();
+            return latestValue?.ToString();
+        }
+
+        static bool isValueType;
+
+        static AsyncReactiveProperty()
+        {
+            isValueType = typeof(T).IsValueType;
         }
 
         class WithoutCurrentEnumerable : IUniTaskAsyncEnumerable<T>
@@ -73,7 +92,7 @@ namespace Cysharp.Threading.Tasks
             }
         }
 
-        sealed class Enumerator : MoveNextSource, IUniTaskAsyncEnumerator<T>, IResolveCancelPromise<T>
+        sealed class Enumerator : MoveNextSource, IUniTaskAsyncEnumerator<T>, ITriggerHandler<T>
         {
             static Action<object> cancellationCallback = CancellationCallback;
 
@@ -127,17 +146,25 @@ namespace Cysharp.Threading.Tasks
                 return default;
             }
 
-            public bool TrySetResult(T value)
+            public void OnNext(T value)
             {
                 this.value = value;
                 completionSource.TrySetResult(true);
-                return true;
             }
 
-            public bool TrySetCanceled(CancellationToken cancellationToken = default)
+            public void OnCanceled(CancellationToken cancellationToken)
             {
                 DisposeAsync().Forget();
-                return true;
+            }
+
+            public void OnCompleted()
+            {
+                completionSource.TrySetResult(false);
+            }
+
+            public void OnError(Exception ex)
+            {
+                completionSource.TrySetException(ex);
             }
 
             static void CancellationCallback(object state)
@@ -147,4 +174,198 @@ namespace Cysharp.Threading.Tasks
             }
         }
     }
-}
+
+    public class ReadOnlyAsyncReactiveProperty<T> : IReadOnlyAsyncReactiveProperty<T>, IDisposable
+    {
+        TriggerEvent<T> triggerEvent;
+
+        T latestValue;
+        IUniTaskAsyncEnumerator<T> enumerator;
+
+        public T Value
+        {
+            get
+            {
+                return latestValue;
+            }
+        }
+
+        public ReadOnlyAsyncReactiveProperty(T initialValue, IUniTaskAsyncEnumerable<T> source, CancellationToken cancellationToken)
+        {
+            latestValue = initialValue;
+            ConsumeEnumerator(source, cancellationToken).Forget();
+        }
+
+        public ReadOnlyAsyncReactiveProperty(IUniTaskAsyncEnumerable<T> source, CancellationToken cancellationToken)
+        {
+            ConsumeEnumerator(source, cancellationToken).Forget();
+        }
+
+        async UniTaskVoid ConsumeEnumerator(IUniTaskAsyncEnumerable<T> source, CancellationToken cancellationToken)
+        {
+            enumerator = source.GetAsyncEnumerator(cancellationToken);
+            try
+            {
+                while (await enumerator.MoveNextAsync())
+                {
+                    var value = enumerator.Current;
+                    this.latestValue = value;
+                    triggerEvent.SetResult(value);
+                }
+            }
+            finally
+            {
+                await enumerator.DisposeAsync();
+                enumerator = null;
+            }
+        }
+
+        public IUniTaskAsyncEnumerable<T> WithoutCurrent()
+        {
+            return new WithoutCurrentEnumerable(this);
+        }
+
+        public IUniTaskAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken)
+        {
+            return new Enumerator(this, cancellationToken, true);
+        }
+
+        public void Dispose()
+        {
+            if (enumerator != null)
+            {
+                enumerator.DisposeAsync().Forget();
+            }
+
+            triggerEvent.SetCompleted();
+        }
+
+        public static implicit operator T(ReadOnlyAsyncReactiveProperty<T> value)
+        {
+            return value.Value;
+        }
+
+        public override string ToString()
+        {
+            if (isValueType) return latestValue.ToString();
+            return latestValue?.ToString();
+        }
+
+        static bool isValueType;
+
+        static ReadOnlyAsyncReactiveProperty()
+        {
+            isValueType = typeof(T).IsValueType;
+        }
+
+        class WithoutCurrentEnumerable : IUniTaskAsyncEnumerable<T>
+        {
+            readonly ReadOnlyAsyncReactiveProperty<T> parent;
+
+            public WithoutCurrentEnumerable(ReadOnlyAsyncReactiveProperty<T> parent)
+            {
+                this.parent = parent;
+            }
+
+            public IUniTaskAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default)
+            {
+                return new Enumerator(parent, cancellationToken, false);
+            }
+        }
+
+        sealed class Enumerator : MoveNextSource, IUniTaskAsyncEnumerator<T>, ITriggerHandler<T>
+        {
+            static Action<object> cancellationCallback = CancellationCallback;
+
+            readonly ReadOnlyAsyncReactiveProperty<T> parent;
+            readonly CancellationToken cancellationToken;
+            readonly CancellationTokenRegistration cancellationTokenRegistration;
+            T value;
+            bool isDisposed;
+            bool firstCall;
+
+            public Enumerator(ReadOnlyAsyncReactiveProperty<T> parent, CancellationToken cancellationToken, bool publishCurrentValue)
+            {
+                this.parent = parent;
+                this.cancellationToken = cancellationToken;
+                this.firstCall = publishCurrentValue;
+
+                parent.triggerEvent.Add(this);
+                TaskTracker.TrackActiveTask(this, 3);
+
+                if (cancellationToken.CanBeCanceled)
+                {
+                    cancellationTokenRegistration = cancellationToken.RegisterWithoutCaptureExecutionContext(cancellationCallback, this);
+                }
+            }
+
+            public T Current => value;
+
+            public UniTask<bool> MoveNextAsync()
+            {
+                // raise latest value on first call.
+                if (firstCall)
+                {
+                    firstCall = false;
+                    value = parent.Value;
+                    return CompletedTasks.True;
+                }
+
+                completionSource.Reset();
+                return new UniTask<bool>(this, completionSource.Version);
+            }
+
+            public UniTask DisposeAsync()
+            {
+                if (!isDisposed)
+                {
+                    isDisposed = true;
+                    TaskTracker.RemoveTracking(this);
+                    completionSource.TrySetCanceled(cancellationToken);
+                    parent.triggerEvent.Remove(this);
+                }
+                return default;
+            }
+
+            public void OnNext(T value)
+            {
+                this.value = value;
+                completionSource.TrySetResult(true);
+            }
+
+            public void OnCanceled(CancellationToken cancellationToken)
+            {
+                DisposeAsync().Forget();
+            }
+
+            public void OnCompleted()
+            {
+                completionSource.TrySetResult(false);
+            }
+
+            public void OnError(Exception ex)
+            {
+                completionSource.TrySetException(ex);
+            }
+
+            static void CancellationCallback(object state)
+            {
+                var self = (Enumerator)state;
+                self.DisposeAsync().Forget();
+            }
+        }
+    }
+
+    public static class StateExtensions
+    {
+        public static ReadOnlyAsyncReactiveProperty<T> ToReadOnlyAsyncReactiveProperty<T>(this IUniTaskAsyncEnumerable<T> source, CancellationToken cancellationToken)
+        {
+            return new ReadOnlyAsyncReactiveProperty<T>(source, cancellationToken);
+        }
+
+        public static ReadOnlyAsyncReactiveProperty<T> ToReadOnlyAsyncReactiveProperty<T>(this IUniTaskAsyncEnumerable<T> source, T initialValue, CancellationToken cancellationToken)
+        {
+            return new ReadOnlyAsyncReactiveProperty<T>(initialValue, source, cancellationToken);
+        }
+    }
+}

src/UniTask/Assets/Plugins/UniTask/Runtime/CancellationTokenExtensions.cs

@@ -1,6 +1,7 @@
 #pragma warning disable CS1591 // Missing XML comment for publicly visible type or member
 
 using System;
+using System.Runtime.CompilerServices;
 using System.Threading;
 
 namespace Cysharp.Threading.Tasks
@@ -9,15 +10,15 @@ namespace Cysharp.Threading.Tasks
     {
         static readonly Action<object> cancellationTokenCallback = Callback;
 
-        public static (UniTask, CancellationTokenRegistration) ToUniTask(this CancellationToken cts)
+        public static (UniTask, CancellationTokenRegistration) ToUniTask(this CancellationToken cancellationToken)
         {
-            if (cts.IsCancellationRequested)
+            if (cancellationToken.IsCancellationRequested)
             {
-                return (UniTask.FromCanceled(cts), default(CancellationTokenRegistration));
+                return (UniTask.FromCanceled(cancellationToken), default(CancellationTokenRegistration));
             }
 
             var promise = new UniTaskCompletionSource();
-            return (promise.Task, cts.RegisterWithoutCaptureExecutionContext(cancellationTokenCallback, promise));
+            return (promise.Task, cancellationToken.RegisterWithoutCaptureExecutionContext(cancellationTokenCallback, promise));
         }
 
         static void Callback(object state)
@@ -26,6 +27,11 @@ namespace Cysharp.Threading.Tasks
             promise.TrySetResult();
         }
 
+        public static CancellationTokenAwaitable WaitUntilCanceled(this CancellationToken cancellationToken)
+        {
+            return new CancellationTokenAwaitable(cancellationToken);
+        }
+
         public static CancellationTokenRegistration RegisterWithoutCaptureExecutionContext(this CancellationToken cancellationToken, Action callback)
         {
             var restoreFlow = false;
@@ -70,5 +76,46 @@ namespace Cysharp.Threading.Tasks
             }
         }
     }
+
+    public struct CancellationTokenAwaitable
+    {
+        CancellationToken cancellationToken;
+
+        public CancellationTokenAwaitable(CancellationToken cancellationToken)
+        {
+            this.cancellationToken = cancellationToken;
+        }
+
+        public Awaiter GetAwaiter()
+        {
+            return new Awaiter(cancellationToken);
+        }
+
+        public struct Awaiter : ICriticalNotifyCompletion
+        {
+            CancellationToken cancellationToken;
+
+            public Awaiter(CancellationToken cancellationToken)
+            {
+                this.cancellationToken = cancellationToken;
+            }
+
+            public bool IsCompleted => !cancellationToken.CanBeCanceled || cancellationToken.IsCancellationRequested;
+
+            public void GetResult()
+            {
+            }
+
+            public void OnCompleted(Action continuation)
+            {
+                UnsafeOnCompleted(continuation);
+            }
+
+            public void UnsafeOnCompleted(Action continuation)
+            {
+                cancellationToken.RegisterWithoutCaptureExecutionContext(continuation);
+            }
+        }
+    }
 }
 

src/UniTask/Assets/Plugins/UniTask/Runtime/Channel.cs

@@ -91,7 +91,8 @@ namespace Cysharp.Threading.Tasks
     {
         readonly Queue<T> items;
         readonly SingleConsumerUnboundedChannelReader readerSource;
-        readonly UniTaskCompletionSource completedTask;
+        UniTaskCompletionSource completedTaskSource;
+        UniTask completedTask;
 
         Exception completionError;
         bool closed;
@@ -99,7 +100,6 @@ namespace Cysharp.Threading.Tasks
         public SingleConsumerUnboundedChannel()
         {
             items = new Queue<T>();
-            completedTask = new UniTaskCompletionSource();
             Writer = new SingleConsumerUnboundedChannelWriter(this);
             readerSource = new SingleConsumerUnboundedChannelReader(this);
             Reader = readerSource;
@@ -146,11 +146,25 @@ namespace Cysharp.Threading.Tasks
                     {
                         if (error == null)
                         {
-                            parent.completedTask.TrySetResult();
+                            if (parent.completedTaskSource != null)
+                            {
+                                parent.completedTaskSource.TrySetResult();
+                            }
+                            else
+                            {
+                                parent.completedTask = UniTask.CompletedTask;
+                            }
                         }
                         else
                         {
-                            parent.completedTask.TrySetException(error);
+                            if (parent.completedTaskSource != null)
+                            {
+                                parent.completedTaskSource.TrySetException(error);
+                            }
+                            else
+                            {
+                                parent.completedTask = UniTask.FromException(error);
+                            }
                         }
 
                         if (waiting)
@@ -179,9 +193,25 @@ namespace Cysharp.Threading.Tasks
             public SingleConsumerUnboundedChannelReader(SingleConsumerUnboundedChannel<T> parent)
             {
                 this.parent = parent;
+
+                TaskTracker.TrackActiveTask(this, 4);
             }
 
-            public override UniTask Completion => parent.completedTask.Task;
+            public override UniTask Completion
+            {
+                get
+                {
+                    if (parent.completedTaskSource != null) return parent.completedTaskSource.Task;
+
+                    if (parent.closed)
+                    {
+                        return parent.completedTask;
+                    }
+
+                    parent.completedTaskSource = new UniTaskCompletionSource();
+                    return parent.completedTaskSource.Task;
+                }
+            }
 
             public override bool TryRead(out T item)
             {
@@ -196,11 +226,25 @@ namespace Cysharp.Threading.Tasks
                         {
                             if (parent.completionError != null)
                             {
-                                parent.completedTask.TrySetException(parent.completionError);
+                                if (parent.completedTaskSource != null)
+                                {
+                                    parent.completedTaskSource.TrySetException(parent.completionError);
+                                }
+                                else
+                                {
+                                    parent.completedTask = UniTask.FromException(parent.completionError);
+                                }
                             }
                             else
                             {
-                                parent.completedTask.TrySetResult();
+                                if (parent.completedTaskSource != null)
+                                {
+                                    parent.completedTaskSource.TrySetResult();
+                                }
+                                else
+                                {
+                                    parent.completedTask = UniTask.CompletedTask;
+                                }
                             }
                         }
                     }
@@ -262,6 +306,7 @@ namespace Cysharp.Threading.Tasks
 
             public void SingalCancellation(CancellationToken cancellationToken)
             {
+                TaskTracker.RemoveTracking(this);
                 core.TrySetCanceled(cancellationToken);
             }
 
@@ -269,10 +314,12 @@ namespace Cysharp.Threading.Tasks
             {
                 if (error != null)
                 {
+                    TaskTracker.RemoveTracking(this);
                     core.TrySetException(error);
                 }
                 else
                 {
+                    TaskTracker.RemoveTracking(this);
                     core.TrySetResult(false);
                 }
             }

src/UniTask/Assets/Plugins/UniTask/Runtime/IUniTaskAsyncEnumerable.cs

@@ -28,6 +28,12 @@ namespace Cysharp.Threading.Tasks
         IUniTaskOrderedAsyncEnumerable<TElement> CreateOrderedEnumerable<TKey>(Func<TElement, CancellationToken, UniTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending);
     }
 
+    public interface IConnectableUniTaskAsyncEnumerable<out T> : IUniTaskAsyncEnumerable<T>
+    {
+        IDisposable Connect();
+    }
+
+    // don't use AsyncGrouping.
     //public interface IUniTaskAsyncGrouping<out TKey, out TElement> : IUniTaskAsyncEnumerable<TElement>
     //{
     //    TKey Key { get; }

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

@@ -65,6 +65,8 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.element = element;
                 this.state = append ? State.RequireAppend : State.RequirePrepend;
                 this.cancellationToken = cancellationToken;
+
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TSource Current { get; private set; }
@@ -136,6 +138,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (enumerator != null)
                 {
                     return enumerator.DisposeAsync();

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

@@ -76,6 +76,7 @@ namespace Cysharp.Threading.Tasks.Linq
         {
             this.source = source;
             this.cancellationToken = cancellationToken;
+            TaskTracker.TrackActiveTask(this, 4);
         }
 
         // abstract
@@ -178,6 +179,7 @@ namespace Cysharp.Threading.Tasks.Linq
         // if require additional resource to dispose, override and call base.DisposeAsync.
         public virtual UniTask DisposeAsync()
         {
+            TaskTracker.RemoveTracking(this);
             if (enumerator != null)
             {
                 return enumerator.DisposeAsync();
@@ -204,6 +206,7 @@ namespace Cysharp.Threading.Tasks.Linq
         {
             this.source = source;
             this.cancellationToken = cancellationToken;
+            TaskTracker.TrackActiveTask(this, 4);
         }
 
         // abstract
@@ -399,6 +402,7 @@ namespace Cysharp.Threading.Tasks.Linq
         // if require additional resource to dispose, override and call base.DisposeAsync.
         public virtual UniTask DisposeAsync()
         {
+            TaskTracker.RemoveTracking(this);
             if (enumerator != null)
             {
                 return enumerator.DisposeAsync();

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

@@ -61,6 +61,8 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.source = source;
                 this.count = count;
                 this.cancellationToken = cancellationToken;
+
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public IList<TSource> Current { get; private set; }
@@ -167,6 +169,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (enumerator != null)
                 {
                     return enumerator.DisposeAsync();
@@ -217,6 +220,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.count = count;
                 this.skip = skip;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public IList<TSource> Current { get; private set; }
@@ -329,6 +333,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (enumerator != null)
                 {
                     return enumerator.DisposeAsync();

src/UniTask/Assets/Plugins/UniTask/Runtime/Linq/CombineLatest.cs.meta

@@ -0,0 +1,11 @@
+fileFormatVersion: 2
+guid: 6cb07f6e88287e34d9b9301a572284a5
+MonoImporter:
+  externalObjects: {}
+  serializedVersion: 2
+  defaultReferences: []
+  executionOrder: 0
+  icon: {instanceID: 0}
+  userData: 
+  assetBundleName: 
+  assetBundleVariant: 

src/UniTask/Assets/Plugins/UniTask/Runtime/Linq/CombineLatest.tt

@@ -0,0 +1,219 @@
+<#@ template debug="false" hostspecific="false" language="C#" #>
+<#@ assembly name="System.Core" #>
+<#@ import namespace="System.Linq" #>
+<#@ import namespace="System.Text" #>
+<#@ import namespace="System.Collections.Generic" #>
+<#@ output extension=".cs" #>
+<#
+    var tMax = 15;
+    Func<int, string> typeArgs = x => string.Join(", ", Enumerable.Range(1, x).Select(x => $"T{x}")) + ", TResult";
+    Func<int, string> paramArgs = x => string.Join(", ", Enumerable.Range(1, x).Select(x => $"IUniTaskAsyncEnumerable<T{x}> source{x}"));
+    Func<int, string> parameters = x => string.Join(", ", Enumerable.Range(1, x).Select(x => $"source{x}"));
+
+
+#>
+using Cysharp.Threading.Tasks.Internal;
+using System;
+using System.Threading;
+
+namespace Cysharp.Threading.Tasks.Linq
+{
+    public static partial class UniTaskAsyncEnumerable
+    {
+<# for(var i = 2; i <= tMax; i++) { #>
+        public static IUniTaskAsyncEnumerable<TResult> CombineLatest<<#= typeArgs(i) #>>(this <#= paramArgs(i) #>, Func<<#= typeArgs(i) #>> resultSelector)
+        {
+<# for(var j = 1; j <= i; j++) { #>
+            Error.ThrowArgumentNullException(source<#= j #>, nameof(source<#= j #>));
+<# } #>
+            Error.ThrowArgumentNullException(resultSelector, nameof(resultSelector));
+
+            return new CombineLatest<<#= typeArgs(i) #>>(<#= parameters(i) #>, resultSelector);
+        }
+
+<# } #>
+    }
+
+<# for(var i = 2; i <= tMax; i++) { #>
+    internal class CombineLatest<<#= typeArgs(i) #>> : IUniTaskAsyncEnumerable<TResult>
+    {
+<# for(var j = 1; j <= i; j++) { #>
+        readonly IUniTaskAsyncEnumerable<T<#= j #>> source<#= j #>;
+<# } #>        
+        readonly Func<<#= typeArgs(i) #>> resultSelector;
+
+        public CombineLatest(<#= paramArgs(i) #>, Func<<#= typeArgs(i) #>> resultSelector)
+        {
+<# for(var j = 1; j <= i; j++) { #>
+            this.source<#= j #> = source<#= j #>;
+<# } #>        
+            this.resultSelector = resultSelector;
+        }
+
+        public IUniTaskAsyncEnumerator<TResult> GetAsyncEnumerator(CancellationToken cancellationToken = default)
+        {
+            return new _CombineLatest(<#= parameters(i) #>, resultSelector, cancellationToken);
+        }
+
+        class _CombineLatest : MoveNextSource, IUniTaskAsyncEnumerator<TResult>
+        {
+<# for(var j = 1; j <= i; j++) { #>
+            static readonly Action<object> Completed<#= j #>Delegate = Completed<#= j #>;
+<# } #>
+            const int CompleteCount = <#= i #>;
+
+<# for(var j = 1; j <= i; j++) { #>
+            readonly IUniTaskAsyncEnumerable<T<#= j #>> source<#= j #>;
+<# } #>       
+            readonly Func<<#= typeArgs(i) #>> resultSelector;
+            CancellationToken cancellationToken;
+
+<# for(var j = 1; j <= i; j++) { #>
+            IUniTaskAsyncEnumerator<T<#= j #>> enumerator<#= j #>;
+            UniTask<bool>.Awaiter awaiter<#= j #>;
+            bool hasCurrent<#= j #>;
+            bool running<#= j #>;
+            T<#= j #> current<#= j #>;
+
+<# } #>
+            int completedCount;
+            bool syncRunning;
+            TResult result;
+
+            public _CombineLatest(<#= paramArgs(i) #>, Func<<#= typeArgs(i) #>> resultSelector, CancellationToken cancellationToken)
+            {
+<# for(var j = 1; j <= i; j++) { #>
+                this.source<#= j #> = source<#= j #>;
+<# } #>                
+                this.resultSelector = resultSelector;
+                this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
+            }
+
+            public TResult Current => result;
+
+            public UniTask<bool> MoveNextAsync()
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                if (completedCount == CompleteCount) return CompletedTasks.False;
+
+                if (enumerator1 == null)
+                {
+<# for(var j = 1; j <= i; j++) { #>
+                    enumerator<#= j #> = source<#= j #>.GetAsyncEnumerator(cancellationToken);
+<# } #>
+                }
+
+                completionSource.Reset();
+
+                AGAIN:
+                syncRunning = true;
+<# for(var j = 1; j <= i; j++) { #>
+                if (!running<#= j #>)
+                {
+                    running<#= j #> = true;
+                    awaiter<#= j #> = enumerator<#= j #>.MoveNextAsync().GetAwaiter();
+                    if (awaiter<#= j #>.IsCompleted)
+                    {
+                        Completed<#= j #>(this);
+                    }
+                    else
+                    {
+                        awaiter<#= j #>.SourceOnCompleted(Completed<#= j #>Delegate, this);
+                    }
+                }
+<# } #>
+
+                if (<#= string.Join(" || ", Enumerable.Range(1, i).Select(x => $"!running{x}")) #>)
+                {
+                    goto AGAIN;
+                }
+                syncRunning = false;
+
+                return new UniTask<bool>(this, completionSource.Version);
+            }
+
+<# for(var j = 1; j <= i; j++) { #>
+            static void Completed<#= j #>(object state)
+            {
+                var self = (_CombineLatest)state;
+                self.running<#= j #> = false;
+
+                try
+                {
+                    if (self.awaiter<#= j #>.GetResult())
+                    {
+                        self.hasCurrent<#= j #> = true;
+                        self.current<#= j #> = self.enumerator<#= j #>.Current;
+                        goto SUCCESS;
+                    }
+                    else
+                    {
+                        self.running<#= j #> = true; // as complete, no more call MoveNextAsync.
+                        if (Interlocked.Increment(ref self.completedCount) == CompleteCount)
+                        {
+                            goto COMPLETE;
+                        }
+                        return;
+                    }
+                }
+                catch (Exception ex)
+                {
+                    self.completedCount = CompleteCount;
+                    self.completionSource.TrySetException(ex);
+                    return;
+                }
+
+                SUCCESS:
+                if (!self.TrySetResult())
+                {
+                    if (self.syncRunning) return;
+                    try
+                    {
+                        self.awaiter<#= j #> = self.enumerator<#= j #>.MoveNextAsync().GetAwaiter();
+                    }
+                    catch (Exception ex)
+                    {
+                        self.completedCount = CompleteCount;
+                        self.completionSource.TrySetException(ex);
+                        return;
+                    }
+
+                    self.awaiter<#= j #>.SourceOnCompleted(Completed<#= j #>Delegate, self);
+                }
+                return;
+                COMPLETE:
+                self.completionSource.TrySetResult(false);
+                return;
+            }
+
+<# } #>
+            bool TrySetResult()
+            {
+                if (<#= string.Join(" && ", Enumerable.Range(1, i).Select(x => $"hasCurrent{x}")) #>)
+                {
+                    result = resultSelector(<#= string.Join(", ", Enumerable.Range(1, i).Select(x => $"current{x}")) #>);
+                    completionSource.TrySetResult(true);
+                    return true;
+                }
+                else
+                {
+                    return false;
+                }
+            }
+
+            public async UniTask DisposeAsync()
+            {
+                TaskTracker.RemoveTracking(this);
+<# for(var j = 1; j <= i; j++) { #>
+                if (enumerator<#= j #> != null)
+                {
+                    await enumerator<#= j #>.DisposeAsync();
+                }
+<# } #>
+            }
+        }
+    }
+
+<# } #>
+}

src/UniTask/Assets/Plugins/UniTask/Runtime/Linq/CombineLatest.tt.meta

@@ -0,0 +1,7 @@
+fileFormatVersion: 2
+guid: b1b8cfa9d17af814a971ee2224aaaaa2
+DefaultImporter:
+  externalObjects: {}
+  userData: 
+  assetBundleName: 
+  assetBundleVariant: 

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

@@ -57,6 +57,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.second = second;
                 this.cancellationToken = cancellationToken;
                 this.iteratingState = IteratingState.IteratingFirst;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TSource Current { get; private set; }
@@ -150,6 +151,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (enumerator != null)
                 {
                     return enumerator.DisposeAsync();

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

@@ -63,6 +63,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.cancellationToken = cancellationToken;
 
                 this.iteratingState = IteratingState.Empty;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TSource Current { get; private set; }
@@ -128,6 +129,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (enumerator != null)
                 {
                     return enumerator.DisposeAsync();

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

@@ -124,6 +124,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.onError = onError;
                 this.onCompleted = onCompleted;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TSource Current { get; private set; }
@@ -244,6 +245,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (enumerator != null)
                 {
                     return enumerator.DisposeAsync();

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

@@ -22,6 +22,22 @@ namespace Cysharp.Threading.Tasks.Linq
             return Cysharp.Threading.Tasks.Linq.ForEach.ForEachAsync(source, action, cancellationToken);
         }
 
+        /// <summary>Obsolete(Error), Use Use ForEachAwaitAsync instead.</summary>
+        [Obsolete("Use ForEachAwaitAsync instead.", true)]
+        [System.ComponentModel.EditorBrowsable(System.ComponentModel.EditorBrowsableState.Never)]
+        public static UniTask ForEachAsync<T>(this IUniTaskAsyncEnumerable<T> source, Func<T, UniTask> action, CancellationToken cancellationToken = default)
+        {
+            throw new NotSupportedException("Use ForEachAwaitAsync instead.");
+        }
+
+        /// <summary>Obsolete(Error), Use Use ForEachAwaitAsync instead.</summary>
+        [Obsolete("Use ForEachAwaitAsync instead.", true)]
+        [System.ComponentModel.EditorBrowsable(System.ComponentModel.EditorBrowsableState.Never)]
+        public static UniTask ForEachAsync<T>(this IUniTaskAsyncEnumerable<T> source, Func<T, int, UniTask> action, CancellationToken cancellationToken = default)
+        {
+            throw new NotSupportedException("Use ForEachAwaitAsync instead.");
+        }
+
         public static UniTask ForEachAwaitAsync<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTask> action, CancellationToken cancellationToken = default)
         {
             Error.ThrowArgumentNullException(source, nameof(source));

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

@@ -255,6 +255,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.elementSelector = elementSelector;
                 this.comparer = comparer;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public IGrouping<TKey, TElement> Current { get; private set; }
@@ -313,6 +314,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (groupEnumerator != null)
                 {
                     groupEnumerator.Dispose();
@@ -364,6 +366,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.resultSelector = resultSelector;
                 this.comparer = comparer;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TResult Current { get; private set; }
@@ -423,6 +426,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (groupEnumerator != null)
                 {
                     groupEnumerator.Dispose();
@@ -470,6 +474,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.elementSelector = elementSelector;
                 this.comparer = comparer;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public IGrouping<TKey, TElement> Current { get; private set; }
@@ -528,6 +533,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (groupEnumerator != null)
                 {
                     groupEnumerator.Dispose();
@@ -582,6 +588,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.resultSelector = resultSelector;
                 this.comparer = comparer;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TResult Current { get; private set; }
@@ -661,6 +668,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (groupEnumerator != null)
                 {
                     groupEnumerator.Dispose();
@@ -708,6 +716,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.elementSelector = elementSelector;
                 this.comparer = comparer;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public IGrouping<TKey, TElement> Current { get; private set; }
@@ -766,6 +775,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (groupEnumerator != null)
                 {
                     groupEnumerator.Dispose();
@@ -820,6 +830,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.resultSelector = resultSelector;
                 this.comparer = comparer;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TResult Current { get; private set; }
@@ -899,6 +910,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (groupEnumerator != null)
                 {
                     groupEnumerator.Dispose();

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

@@ -129,6 +129,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.resultSelector = resultSelector;
                 this.comparer = comparer;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TResult Current { get; private set; }
@@ -208,6 +209,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (enumerator != null)
                 {
                     return enumerator.DisposeAsync();
@@ -273,6 +275,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.resultSelector = resultSelector;
                 this.comparer = comparer;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TResult Current { get; private set; }
@@ -401,6 +404,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (enumerator != null)
                 {
                     return enumerator.DisposeAsync();
@@ -466,6 +470,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.resultSelector = resultSelector;
                 this.comparer = comparer;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TResult Current { get; private set; }
@@ -594,6 +599,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (enumerator != null)
                 {
                     return enumerator.DisposeAsync();

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

@@ -131,6 +131,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.resultSelector = resultSelector;
                 this.comparer = comparer;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TResult Current { get; private set; }
@@ -248,6 +249,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (valueEnumerator != null)
                 {
                     valueEnumerator.Dispose();
@@ -321,6 +323,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.resultSelector = resultSelector;
                 this.comparer = comparer;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TResult Current { get; private set; }
@@ -476,6 +479,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (valueEnumerator != null)
                 {
                     valueEnumerator.Dispose();
@@ -549,6 +553,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.resultSelector = resultSelector;
                 this.comparer = comparer;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TResult Current { get; private set; }
@@ -704,6 +709,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (valueEnumerator != null)
                 {
                     valueEnumerator.Dispose();

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

@@ -422,6 +422,7 @@ namespace Cysharp.Threading.Tasks.Linq
             {
                 this.parent = parent;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TElement Current { get; private set; }
@@ -477,6 +478,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 return default;
             }
         }

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

@@ -0,0 +1,128 @@
+using Cysharp.Threading.Tasks.Internal;
+using System;
+using System.Threading;
+
+namespace Cysharp.Threading.Tasks.Linq
+{
+    public static partial class UniTaskAsyncEnumerable
+    {
+        public static IUniTaskAsyncEnumerable<(TSource, TSource)> Pairwise<TSource>(this IUniTaskAsyncEnumerable<TSource> source)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+
+            return new Pairwise<TSource>(source);
+        }
+    }
+
+    internal sealed class Pairwise<TSource> : IUniTaskAsyncEnumerable<(TSource, TSource)>
+    {
+        readonly IUniTaskAsyncEnumerable<TSource> source;
+
+        public Pairwise(IUniTaskAsyncEnumerable<TSource> source)
+        {
+            this.source = source;
+        }
+
+        public IUniTaskAsyncEnumerator<(TSource, TSource)> GetAsyncEnumerator(CancellationToken cancellationToken = default)
+        {
+            return new _Pairwise(source, cancellationToken);
+        }
+
+        sealed class _Pairwise : MoveNextSource, IUniTaskAsyncEnumerator<(TSource, TSource)>
+        {
+            static readonly Action<object> MoveNextCoreDelegate = MoveNextCore;
+
+            readonly IUniTaskAsyncEnumerable<TSource> source;
+            CancellationToken cancellationToken;
+
+            IUniTaskAsyncEnumerator<TSource> enumerator;
+            UniTask<bool>.Awaiter awaiter;
+
+            TSource prev;
+            bool isFirst;
+
+            public _Pairwise(IUniTaskAsyncEnumerable<TSource> source, CancellationToken cancellationToken)
+            {
+                this.source = source;
+                this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
+            }
+
+            public (TSource, TSource) Current { get; private set; }
+
+            public UniTask<bool> MoveNextAsync()
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+
+                if (enumerator == null)
+                {
+                    isFirst = true;
+                    enumerator = source.GetAsyncEnumerator(cancellationToken);
+                }
+
+                completionSource.Reset();
+                SourceMoveNext();
+                return new UniTask<bool>(this, completionSource.Version);
+            }
+
+            void SourceMoveNext()
+            {
+                try
+                {
+                    awaiter = enumerator.MoveNextAsync().GetAwaiter();
+                    if (awaiter.IsCompleted)
+                    {
+                        MoveNextCore(this);
+                    }
+                    else
+                    {
+                        awaiter.SourceOnCompleted(MoveNextCoreDelegate, this);
+                    }
+                }
+                catch (Exception ex)
+                {
+                    completionSource.TrySetException(ex);
+                }
+            }
+
+            static void MoveNextCore(object state)
+            {
+                var self = (_Pairwise)state;
+
+                if (self.TryGetResult(self.awaiter, out var result))
+                {
+                    if (result)
+                    {
+                        if (self.isFirst)
+                        {
+                            self.isFirst = false;
+                            self.prev = self.enumerator.Current;
+                            self.SourceMoveNext(); // run again. okay to use recursive(only one more).
+                        }
+                        else
+                        {
+                            var p = self.prev;
+                            self.prev = self.enumerator.Current;
+                            self.Current = (p, self.prev);
+                            self.completionSource.TrySetResult(true);
+                        }
+                    }
+                    else
+                    {
+                        self.completionSource.TrySetResult(false);
+                    }
+                }
+            }
+
+            public UniTask DisposeAsync()
+            {
+                TaskTracker.RemoveTracking(this);
+                if (enumerator != null)
+                {
+                    return enumerator.DisposeAsync();
+                }
+                return default;
+            }
+        }
+    }
+}

src/UniTask/Assets/Plugins/UniTask/Runtime/Linq/Pairwise.cs.meta

@@ -0,0 +1,11 @@
+fileFormatVersion: 2
+guid: cddbf051d2a88f549986c468b23214af
+MonoImporter:
+  externalObjects: {}
+  serializedVersion: 2
+  defaultReferences: []
+  executionOrder: 0
+  icon: {instanceID: 0}
+  userData: 
+  assetBundleName: 
+  assetBundleVariant: 

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

@@ -0,0 +1,171 @@
+using Cysharp.Threading.Tasks.Internal;
+using System;
+using System.Threading;
+
+namespace Cysharp.Threading.Tasks.Linq
+{
+    public static partial class UniTaskAsyncEnumerable
+    {
+        public static IConnectableUniTaskAsyncEnumerable<TSource> Publish<TSource>(this IUniTaskAsyncEnumerable<TSource> source)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+
+            return new Publish<TSource>(source);
+        }
+    }
+
+    internal sealed class Publish<TSource> : IConnectableUniTaskAsyncEnumerable<TSource>
+    {
+        readonly IUniTaskAsyncEnumerable<TSource> source;
+        readonly CancellationTokenSource cancellationTokenSource;
+
+        TriggerEvent<TSource> trigger;
+        IUniTaskAsyncEnumerator<TSource> enumerator;
+        IDisposable connectedDisposable;
+        bool isCompleted;
+
+        public Publish(IUniTaskAsyncEnumerable<TSource> source)
+        {
+            this.source = source;
+            this.cancellationTokenSource = new CancellationTokenSource();
+        }
+
+        public IDisposable Connect()
+        {
+            if (connectedDisposable != null) return connectedDisposable;
+
+            if (enumerator == null)
+            {
+                enumerator = source.GetAsyncEnumerator(cancellationTokenSource.Token);
+            }
+
+            ConsumeEnumerator().Forget();
+
+            connectedDisposable = new ConnectDisposable(cancellationTokenSource);
+            return connectedDisposable;
+        }
+
+        async UniTaskVoid ConsumeEnumerator()
+        {
+            try
+            {
+                try
+                {
+                    while (await enumerator.MoveNextAsync())
+                    {
+                        trigger.SetResult(enumerator.Current);
+                    }
+                    trigger.SetCompleted();
+                }
+                catch (Exception ex)
+                {
+                    trigger.SetError(ex);
+                }
+            }
+            finally
+            {
+                isCompleted = true;
+                await enumerator.DisposeAsync();
+            }
+        }
+
+        public IUniTaskAsyncEnumerator<TSource> GetAsyncEnumerator(CancellationToken cancellationToken = default)
+        {
+            return new _Publish(this, cancellationToken);
+        }
+
+        sealed class ConnectDisposable : IDisposable
+        {
+            readonly CancellationTokenSource cancellationTokenSource;
+
+            public ConnectDisposable(CancellationTokenSource cancellationTokenSource)
+            {
+                this.cancellationTokenSource = cancellationTokenSource;
+            }
+
+            public void Dispose()
+            {
+                this.cancellationTokenSource.Cancel();
+            }
+        }
+
+        sealed class _Publish : MoveNextSource, IUniTaskAsyncEnumerator<TSource>, ITriggerHandler<TSource>
+        {
+            static readonly Action<object> CancelDelegate = OnCanceled;
+
+            readonly Publish<TSource> parent;
+            CancellationToken cancellationToken;
+            CancellationTokenRegistration cancellationTokenRegistration;
+            bool isDisposed;
+
+            public _Publish(Publish<TSource> parent, CancellationToken cancellationToken)
+            {
+                if (cancellationToken.IsCancellationRequested) return;
+
+                this.parent = parent;
+                this.cancellationToken = cancellationToken;
+
+                if (cancellationToken.CanBeCanceled)
+                {
+                    this.cancellationTokenRegistration = cancellationToken.RegisterWithoutCaptureExecutionContext(CancelDelegate, this);
+                }
+
+                parent.trigger.Add(this);
+                TaskTracker.TrackActiveTask(this, 3);
+            }
+
+            public TSource Current { get; private set; }
+
+            public UniTask<bool> MoveNextAsync()
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+
+                if (parent.isCompleted) return CompletedTasks.False;
+
+                completionSource.Reset();
+                return new UniTask<bool>(this, completionSource.Version);
+            }
+
+            static void OnCanceled(object state)
+            {
+                var self = (_Publish)state;
+                self.completionSource.TrySetCanceled(self.cancellationToken);
+                self.DisposeAsync().Forget();
+            }
+
+            public UniTask DisposeAsync()
+            {
+                if (!isDisposed)
+                {
+                    isDisposed = true;
+                    TaskTracker.RemoveTracking(this);
+                    cancellationTokenRegistration.Dispose();
+                    parent.trigger.Remove(this);
+                }
+
+                return default;
+            }
+
+            public void OnNext(TSource value)
+            {
+                Current = value;
+                completionSource.TrySetResult(true);
+            }
+
+            public void OnCanceled(CancellationToken cancellationToken)
+            {
+                completionSource.TrySetCanceled(cancellationToken);
+            }
+
+            public void OnCompleted()
+            {
+                completionSource.TrySetResult(false);
+            }
+
+            public void OnError(Exception ex)
+            {
+                completionSource.TrySetException(ex);
+            }
+        }
+    }
+}

src/UniTask/Assets/Plugins/UniTask/Runtime/Linq/Publish.cs.meta

@@ -0,0 +1,11 @@
+fileFormatVersion: 2
+guid: 93c684d1e88c09d4e89b79437d97b810
+MonoImporter:
+  externalObjects: {}
+  serializedVersion: 2
+  defaultReferences: []
+  executionOrder: 0
+  icon: {instanceID: 0}
+  userData: 
+  assetBundleName: 
+  assetBundleVariant: 

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

@@ -33,6 +33,7 @@ namespace Cysharp.Threading.Tasks.Linq
             Channel<TSource> channel;
             IUniTaskAsyncEnumerator<TSource> channelEnumerator;
             IUniTaskAsyncEnumerator<TSource> sourceEnumerator;
+            bool channelClosed;
 
             public _Queue(IUniTaskAsyncEnumerable<TSource> source, CancellationToken cancellationToken)
             {
@@ -53,13 +54,13 @@ namespace Cysharp.Threading.Tasks.Linq
 
                     channelEnumerator = channel.Reader.ReadAllAsync().GetAsyncEnumerator(cancellationToken);
 
-                    ConsumeAll(sourceEnumerator, channel).Forget();
+                    ConsumeAll(this, sourceEnumerator, channel).Forget();
                 }
 
                 return channelEnumerator.MoveNextAsync();
             }
 
-            static async UniTaskVoid ConsumeAll(IUniTaskAsyncEnumerator<TSource> enumerator, ChannelWriter<TSource> writer)
+            static async UniTaskVoid ConsumeAll(_Queue self, IUniTaskAsyncEnumerator<TSource> enumerator, ChannelWriter<TSource> writer)
             {
                 try
                 {
@@ -75,6 +76,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 }
                 finally
                 {
+                    self.channelClosed = true;
                     await enumerator.DisposeAsync();
                 }
             }
@@ -89,6 +91,12 @@ namespace Cysharp.Threading.Tasks.Linq
                 {
                     await channelEnumerator.DisposeAsync();
                 }
+
+                if (!channelClosed)
+                {
+                    channelClosed = true;
+                    channel.Writer.TryComplete(new OperationCanceledException());
+                }
             }
         }
     }

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

@@ -40,6 +40,7 @@ namespace Cysharp.Threading.Tasks.Linq
             {
                 this.source = source;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TSource Current { get; private set; }
@@ -69,6 +70,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 return default;
             }
         }

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

@@ -160,6 +160,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.selector2 = selector2;
                 this.resultSelector = resultSelector;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TResult Current { get; private set; }
@@ -324,6 +325,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public async UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (selectedEnumerator != null)
                 {
                     await selectedEnumerator.DisposeAsync();
@@ -398,6 +400,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.selector2 = selector2;
                 this.resultSelector = resultSelector;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TResult Current { get; private set; }
@@ -598,6 +601,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public async UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (selectedEnumerator != null)
                 {
                     await selectedEnumerator.DisposeAsync();
@@ -672,6 +676,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.selector2 = selector2;
                 this.resultSelector = resultSelector;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TResult Current { get; private set; }
@@ -872,6 +877,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public async UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (selectedEnumerator != null)
                 {
                     await selectedEnumerator.DisposeAsync();

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

@@ -56,6 +56,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.source = source;
                 this.count = count;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TSource Current { get; private set; }
@@ -146,6 +147,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (enumerator != null)
                 {
                     return enumerator.DisposeAsync();

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

@@ -0,0 +1,144 @@
+using Cysharp.Threading.Tasks.Internal;
+using System;
+using System.Threading;
+
+namespace Cysharp.Threading.Tasks.Linq
+{
+    public static partial class UniTaskAsyncEnumerable
+    {
+        public static IUniTaskAsyncEnumerable<TSource> SkipUntilCanceled<TSource>(this IUniTaskAsyncEnumerable<TSource> source, CancellationToken cancellationToken)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+
+            return new SkipUntilCanceled<TSource>(source, cancellationToken);
+        }
+    }
+
+    internal sealed class SkipUntilCanceled<TSource> : IUniTaskAsyncEnumerable<TSource>
+    {
+        readonly IUniTaskAsyncEnumerable<TSource> source;
+        readonly CancellationToken cancellationToken;
+
+        public SkipUntilCanceled(IUniTaskAsyncEnumerable<TSource> source, CancellationToken cancellationToken)
+        {
+            this.source = source;
+            this.cancellationToken = cancellationToken;
+        }
+
+        public IUniTaskAsyncEnumerator<TSource> GetAsyncEnumerator(CancellationToken cancellationToken = default)
+        {
+            return new _SkipUntilCanceled(source, this.cancellationToken, cancellationToken);
+        }
+
+        sealed class _SkipUntilCanceled : MoveNextSource, IUniTaskAsyncEnumerator<TSource>
+        {
+            static readonly Action<object> MoveNextCoreDelegate = MoveNextCore;
+
+            readonly IUniTaskAsyncEnumerable<TSource> source;
+            CancellationToken cancellationToken1;
+            CancellationToken cancellationToken2;
+
+            bool isCanceled;
+            IUniTaskAsyncEnumerator<TSource> enumerator;
+            UniTask<bool>.Awaiter awaiter;
+            bool continueNext;
+
+            public _SkipUntilCanceled(IUniTaskAsyncEnumerable<TSource> source, CancellationToken cancellationToken1, CancellationToken cancellationToken2)
+            {
+                this.source = source;
+                this.cancellationToken1 = cancellationToken1;
+                this.cancellationToken2 = cancellationToken2;
+                TaskTracker.TrackActiveTask(this, 3);
+            }
+
+            public TSource Current { get; private set; }
+
+            public UniTask<bool> MoveNextAsync()
+            {
+                if (cancellationToken1.IsCancellationRequested) isCanceled = true;
+                if (cancellationToken2.IsCancellationRequested) isCanceled = true;
+
+                if (enumerator == null)
+                {
+                    enumerator = source.GetAsyncEnumerator(cancellationToken2); // use only AsyncEnumerator provided token.
+                }
+                completionSource.Reset();
+                SourceMoveNext();
+                return new UniTask<bool>(this, completionSource.Version);
+            }
+
+            void SourceMoveNext()
+            {
+                try
+                {
+                    LOOP:
+                    awaiter = enumerator.MoveNextAsync().GetAwaiter();
+                    if (awaiter.IsCompleted)
+                    {
+                        continueNext = true;
+                        MoveNextCore(this);
+                        if (continueNext)
+                        {
+                            continueNext = false;
+                            goto LOOP;
+                        }
+                    }
+                    else
+                    {
+                        awaiter.SourceOnCompleted(MoveNextCoreDelegate, this);
+                    }
+                }
+                catch (Exception ex)
+                {
+                    completionSource.TrySetException(ex);
+                }
+            }
+
+            static void MoveNextCore(object state)
+            {
+                var self = (_SkipUntilCanceled)state;
+
+                if (self.TryGetResult(self.awaiter, out var result))
+                {
+                    if (result)
+                    {
+                        AGAIN:
+
+                        if (self.isCanceled)
+                        {
+                            self.continueNext = false;
+                            self.Current = self.enumerator.Current;
+                            self.completionSource.TrySetResult(true);
+                        }
+                        else
+                        {
+                            if (self.cancellationToken1.IsCancellationRequested) self.isCanceled = true;
+                            if (self.cancellationToken2.IsCancellationRequested) self.isCanceled = true;
+
+                            if (self.isCanceled) goto AGAIN;
+
+                            if (!self.continueNext)
+                            {
+                                self.SourceMoveNext();
+                            }
+                        }
+                    }
+                    else
+                    {
+                        self.completionSource.TrySetResult(false);
+                    }
+                }
+            }
+
+            public UniTask DisposeAsync()
+            {
+                TaskTracker.RemoveTracking(this);
+                if (enumerator != null)
+                {
+                    return enumerator.DisposeAsync();
+                }
+                return default;
+            }
+        }
+    }
+}

src/UniTask/Assets/Plugins/UniTask/Runtime/Linq/SkipUntilCanceled.cs.meta

@@ -0,0 +1,11 @@
+fileFormatVersion: 2
+guid: 4b1a778aef7150d47b93a49aa1bc34ae
+MonoImporter:
+  externalObjects: {}
+  serializedVersion: 2
+  defaultReferences: []
+  executionOrder: 0
+  icon: {instanceID: 0}
+  userData: 
+  assetBundleName: 
+  assetBundleVariant: 

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

@@ -47,6 +47,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.source = source;
                 this.count = count;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TSource Current { get; private set; }
@@ -111,6 +112,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (enumerator != null)
                 {
                     return enumerator.DisposeAsync();

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

@@ -57,6 +57,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.source = source;
                 this.count = count;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TSource Current { get; private set; }
@@ -162,6 +163,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (enumerator != null)
                 {
                     return enumerator.DisposeAsync();

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

@@ -0,0 +1,164 @@
+using Cysharp.Threading.Tasks.Internal;
+using System;
+using System.Threading;
+
+namespace Cysharp.Threading.Tasks.Linq
+{
+    public static partial class UniTaskAsyncEnumerable
+    {
+        public static IUniTaskAsyncEnumerable<TSource> TakeUntilCanceled<TSource>(this IUniTaskAsyncEnumerable<TSource> source, CancellationToken cancellationToken)
+        {
+            Error.ThrowArgumentNullException(source, nameof(source));
+
+            return new TakeUntilCanceled<TSource>(source, cancellationToken);
+        }
+    }
+
+    internal sealed class TakeUntilCanceled<TSource> : IUniTaskAsyncEnumerable<TSource>
+    {
+        readonly IUniTaskAsyncEnumerable<TSource> source;
+        readonly CancellationToken cancellationToken;
+
+        public TakeUntilCanceled(IUniTaskAsyncEnumerable<TSource> source, CancellationToken cancellationToken)
+        {
+            this.source = source;
+            this.cancellationToken = cancellationToken;
+        }
+
+        public IUniTaskAsyncEnumerator<TSource> GetAsyncEnumerator(CancellationToken cancellationToken = default)
+        {
+            return new _TakeUntilCanceled(source, this.cancellationToken, cancellationToken);
+        }
+
+        sealed class _TakeUntilCanceled : MoveNextSource, IUniTaskAsyncEnumerator<TSource>
+        {
+            static readonly Action<object> CancelDelegate1 = OnCanceled1;
+            static readonly Action<object> CancelDelegate2 = OnCanceled2;
+            static readonly Action<object> MoveNextCoreDelegate = MoveNextCore;
+
+            readonly IUniTaskAsyncEnumerable<TSource> source;
+            CancellationToken cancellationToken1;
+            CancellationToken cancellationToken2;
+            CancellationTokenRegistration cancellationTokenRegistration1;
+            CancellationTokenRegistration cancellationTokenRegistration2;
+
+            bool isCanceled;
+            IUniTaskAsyncEnumerator<TSource> enumerator;
+            UniTask<bool>.Awaiter awaiter;
+
+            public _TakeUntilCanceled(IUniTaskAsyncEnumerable<TSource> source, CancellationToken cancellationToken1, CancellationToken cancellationToken2)
+            {
+                this.source = source;
+                this.cancellationToken1 = cancellationToken1;
+                this.cancellationToken2 = cancellationToken2;
+
+                if (cancellationToken1.CanBeCanceled)
+                {
+                    this.cancellationTokenRegistration1 = cancellationToken1.RegisterWithoutCaptureExecutionContext(CancelDelegate1, this);
+                }
+
+                if (cancellationToken1 != cancellationToken2 && cancellationToken2.CanBeCanceled)
+                {
+                    this.cancellationTokenRegistration2 = cancellationToken2.RegisterWithoutCaptureExecutionContext(CancelDelegate2, this);
+                }
+                TaskTracker.TrackActiveTask(this, 3);
+            }
+
+            public TSource Current { get; private set; }
+
+            public UniTask<bool> MoveNextAsync()
+            {
+                if (cancellationToken1.IsCancellationRequested) isCanceled = true;
+                if (cancellationToken2.IsCancellationRequested) isCanceled = true;
+
+                if (enumerator == null)
+                {
+                    enumerator = source.GetAsyncEnumerator(cancellationToken2); // use only AsyncEnumerator provided token.
+                }
+
+                if (isCanceled) return CompletedTasks.False;
+
+                completionSource.Reset();
+                SourceMoveNext();
+                return new UniTask<bool>(this, completionSource.Version);
+            }
+
+            void SourceMoveNext()
+            {
+                try
+                {
+                    awaiter = enumerator.MoveNextAsync().GetAwaiter();
+                    if (awaiter.IsCompleted)
+                    {
+                        MoveNextCore(this);
+                    }
+                    else
+                    {
+                        awaiter.SourceOnCompleted(MoveNextCoreDelegate, this);
+                    }
+                }
+                catch (Exception ex)
+                {
+                    completionSource.TrySetException(ex);
+                }
+            }
+
+            static void MoveNextCore(object state)
+            {
+                var self = (_TakeUntilCanceled)state;
+
+                if (self.TryGetResult(self.awaiter, out var result))
+                {
+                    if (result)
+                    {
+                        if (self.isCanceled)
+                        {
+                            self.completionSource.TrySetResult(false);
+                        }
+                        else
+                        {
+                            self.Current = self.enumerator.Current;
+                            self.completionSource.TrySetResult(true);
+                        }
+                    }
+                    else
+                    {
+                        self.completionSource.TrySetResult(false);
+                    }
+                }
+            }
+
+            static void OnCanceled1(object state)
+            {
+                var self = (_TakeUntilCanceled)state;
+                if (!self.isCanceled)
+                {
+                    self.cancellationTokenRegistration2.Dispose();
+                    self.completionSource.TrySetResult(false);
+                }
+            }
+
+            static void OnCanceled2(object state)
+            {
+                var self = (_TakeUntilCanceled)state;
+                if (!self.isCanceled)
+                {
+                    self.cancellationTokenRegistration1.Dispose();
+                    self.completionSource.TrySetResult(false);
+                }
+            }
+
+            public UniTask DisposeAsync()
+            {
+                TaskTracker.RemoveTracking(this);
+                cancellationTokenRegistration1.Dispose();
+                cancellationTokenRegistration2.Dispose();
+                if (enumerator != null)
+                {
+                    return enumerator.DisposeAsync();
+                }
+                return default;
+            }
+        }
+    }
+}

src/UniTask/Assets/Plugins/UniTask/Runtime/Linq/TakeUntilCanceled.cs.meta

@@ -0,0 +1,11 @@
+fileFormatVersion: 2
+guid: e82f498cf3a1df04cbf646773fc11319
+MonoImporter:
+  externalObjects: {}
+  serializedVersion: 2
+  defaultReferences: []
+  executionOrder: 0
+  icon: {instanceID: 0}
+  userData: 
+  assetBundleName: 
+  assetBundleVariant: 

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

@@ -1,7 +1,4 @@
-using Cysharp.Threading.Tasks.Internal;
-using System;
-using System.Collections.Generic;
-using System.Threading;
+using System.Threading;
 
 namespace Cysharp.Threading.Tasks.Linq
 {

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

@@ -0,0 +1,241 @@
+using Cysharp.Threading.Tasks.Internal;
+using System;
+using System.Collections.Generic;
+using System.Threading;
+
+namespace Cysharp.Threading.Tasks.Linq
+{
+    public static partial class UniTaskAsyncEnumerable
+    {
+        public static IUniTaskAsyncEnumerable<TProperty> EveryValueChanged<TTarget, TProperty>(TTarget target, Func<TTarget, TProperty> propertySelector, PlayerLoopTiming monitorTiming = PlayerLoopTiming.Update, IEqualityComparer<TProperty> equalityComparer = null)
+            where TTarget : class
+        {
+            var unityObject = target as UnityEngine.Object;
+            var isUnityObject = target is UnityEngine.Object; // don't use (unityObject == null)
+
+            if (isUnityObject)
+            {
+                return new EveryValueChangedUnityObject<TTarget, TProperty>(target, propertySelector, equalityComparer ?? UnityEqualityComparer.GetDefault<TProperty>(), monitorTiming);
+            }
+            else
+            {
+                return new EveryValueChangedStandardObject<TTarget, TProperty>(target, propertySelector, equalityComparer ?? UnityEqualityComparer.GetDefault<TProperty>(), monitorTiming);
+            }
+        }
+    }
+
+    internal sealed class EveryValueChangedUnityObject<TTarget, TProperty> : IUniTaskAsyncEnumerable<TProperty>
+    {
+        readonly TTarget target;
+        readonly Func<TTarget, TProperty> propertySelector;
+        readonly IEqualityComparer<TProperty> equalityComparer;
+        readonly PlayerLoopTiming monitorTiming;
+
+        public EveryValueChangedUnityObject(TTarget target, Func<TTarget, TProperty> propertySelector, IEqualityComparer<TProperty> equalityComparer, PlayerLoopTiming monitorTiming)
+        {
+            this.target = target;
+            this.propertySelector = propertySelector;
+            this.equalityComparer = equalityComparer;
+            this.monitorTiming = monitorTiming;
+        }
+
+        public IUniTaskAsyncEnumerator<TProperty> GetAsyncEnumerator(CancellationToken cancellationToken = default)
+        {
+            return new _EveryValueChanged(target, propertySelector, equalityComparer, monitorTiming, cancellationToken);
+        }
+
+        sealed class _EveryValueChanged : MoveNextSource, IUniTaskAsyncEnumerator<TProperty>, IPlayerLoopItem
+        {
+            readonly TTarget target;
+            readonly UnityEngine.Object targetAsUnityObject;
+            readonly IEqualityComparer<TProperty> equalityComparer;
+            readonly Func<TTarget, TProperty> propertySelector;
+            CancellationToken cancellationToken;
+
+            bool first;
+            TProperty currentValue;
+            bool disposed;
+
+            public _EveryValueChanged(TTarget target, Func<TTarget, TProperty> propertySelector, IEqualityComparer<TProperty> equalityComparer, PlayerLoopTiming monitorTiming, CancellationToken cancellationToken)
+            {
+                this.target = target;
+                this.targetAsUnityObject = target as UnityEngine.Object;
+                this.propertySelector = propertySelector;
+                this.equalityComparer = equalityComparer;
+                this.cancellationToken = cancellationToken;
+                this.first = true;
+                TaskTracker.TrackActiveTask(this, 2);
+                PlayerLoopHelper.AddAction(monitorTiming, this);
+            }
+
+            public TProperty Current => currentValue;
+
+            public UniTask<bool> MoveNextAsync()
+            {
+                // return false instead of throw
+                if (disposed || cancellationToken.IsCancellationRequested) return CompletedTasks.False;
+
+                if (first)
+                {
+                    first = false;
+                    if (targetAsUnityObject == null)
+                    {
+                        return CompletedTasks.False;
+                    }
+                    this.currentValue = propertySelector(target);
+                    return CompletedTasks.True;
+                }
+
+                completionSource.Reset();
+                return new UniTask<bool>(this, completionSource.Version);
+            }
+
+            public UniTask DisposeAsync()
+            {
+                if (!disposed)
+                {
+                    disposed = true;
+                    TaskTracker.RemoveTracking(this);
+                }
+                return default;
+            }
+
+            public bool MoveNext()
+            {
+                if (disposed || cancellationToken.IsCancellationRequested || targetAsUnityObject == null) // destroyed = cancel.
+                {
+                    completionSource.TrySetResult(false);
+                    DisposeAsync().Forget();
+                    return false;
+                }
+
+                TProperty nextValue = default(TProperty);
+                try
+                {
+                    nextValue = propertySelector(target);
+                    if (equalityComparer.Equals(currentValue, nextValue))
+                    {
+                        return true;
+                    }
+                }
+                catch (Exception ex)
+                {
+                    completionSource.TrySetException(ex);
+                    DisposeAsync().Forget();
+                    return false;
+                }
+
+                currentValue = nextValue;
+                completionSource.TrySetResult(true);
+                return true;
+            }
+        }
+    }
+
+    internal sealed class EveryValueChangedStandardObject<TTarget, TProperty> : IUniTaskAsyncEnumerable<TProperty>
+        where TTarget : class
+    {
+        readonly WeakReference<TTarget> target;
+        readonly Func<TTarget, TProperty> propertySelector;
+        readonly IEqualityComparer<TProperty> equalityComparer;
+        readonly PlayerLoopTiming monitorTiming;
+
+        public EveryValueChangedStandardObject(TTarget target, Func<TTarget, TProperty> propertySelector, IEqualityComparer<TProperty> equalityComparer, PlayerLoopTiming monitorTiming)
+        {
+            this.target = new WeakReference<TTarget>(target, false);
+            this.propertySelector = propertySelector;
+            this.equalityComparer = equalityComparer;
+            this.monitorTiming = monitorTiming;
+        }
+
+        public IUniTaskAsyncEnumerator<TProperty> GetAsyncEnumerator(CancellationToken cancellationToken = default)
+        {
+            return new _EveryValueChanged(target, propertySelector, equalityComparer, monitorTiming, cancellationToken);
+        }
+
+        sealed class _EveryValueChanged : MoveNextSource, IUniTaskAsyncEnumerator<TProperty>, IPlayerLoopItem
+        {
+            readonly WeakReference<TTarget> target;
+            readonly IEqualityComparer<TProperty> equalityComparer;
+            readonly Func<TTarget, TProperty> propertySelector;
+            CancellationToken cancellationToken;
+
+            bool first;
+            TProperty currentValue;
+            bool disposed;
+
+            public _EveryValueChanged(WeakReference<TTarget> target, Func<TTarget, TProperty> propertySelector, IEqualityComparer<TProperty> equalityComparer, PlayerLoopTiming monitorTiming, CancellationToken cancellationToken)
+            {
+                this.target = target;
+                this.propertySelector = propertySelector;
+                this.equalityComparer = equalityComparer;
+                this.cancellationToken = cancellationToken;
+                this.first = true;
+                TaskTracker.TrackActiveTask(this, 2);
+                PlayerLoopHelper.AddAction(monitorTiming, this);
+            }
+
+            public TProperty Current => currentValue;
+
+            public UniTask<bool> MoveNextAsync()
+            {
+                if (disposed || cancellationToken.IsCancellationRequested) return CompletedTasks.False;
+
+                if (first)
+                {
+                    first = false;
+                    if (!target.TryGetTarget(out var t))
+                    {
+                        return CompletedTasks.False;
+                    }
+                    this.currentValue = propertySelector(t);
+                    return CompletedTasks.True;
+                }
+
+                completionSource.Reset();
+                return new UniTask<bool>(this, completionSource.Version);
+            }
+
+            public UniTask DisposeAsync()
+            {
+                if (!disposed)
+                {
+                    disposed = true;
+                    TaskTracker.RemoveTracking(this);
+                }
+                return default;
+            }
+
+            public bool MoveNext()
+            {
+                UnityEngine.Debug.Log("TRY_RESULT:" + target.TryGetTarget(out var _));
+                if (disposed || cancellationToken.IsCancellationRequested || !target.TryGetTarget(out var t))
+                {
+                    completionSource.TrySetResult(false);
+                    DisposeAsync().Forget();
+                    return false;
+                }
+
+                TProperty nextValue = default(TProperty);
+                try
+                {
+                    nextValue = propertySelector(t);
+                    if (equalityComparer.Equals(currentValue, nextValue))
+                    {
+                        return true;
+                    }
+                }
+                catch (Exception ex)
+                {
+                    completionSource.TrySetException(ex);
+                    DisposeAsync().Forget();
+                    return false;
+                }
+
+                currentValue = nextValue;
+                completionSource.TrySetResult(true);
+                return true;
+            }
+        }
+    }
+}

src/UniTask/Assets/Plugins/UniTask/Runtime/Linq/UnityExtensions/EveryValueChanged.cs.meta

@@ -0,0 +1,11 @@
+fileFormatVersion: 2
+guid: 1ec39f1c41c305344854782c935ad354
+MonoImporter:
+  externalObjects: {}
+  serializedVersion: 2
+  defaultReferences: []
+  executionOrder: 0
+  icon: {instanceID: 0}
+  userData: 
+  assetBundleName: 
+  assetBundleVariant: 

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

@@ -84,6 +84,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.second = second;
                 this.resultSelector = resultSelector;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TResult Current { get; private set; }
@@ -181,6 +182,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public async UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (firstEnumerator != null)
                 {
                     await firstEnumerator.DisposeAsync();
@@ -236,6 +238,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.second = second;
                 this.resultSelector = resultSelector;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TResult Current { get; private set; }
@@ -351,6 +354,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public async UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (firstEnumerator != null)
                 {
                     await firstEnumerator.DisposeAsync();
@@ -406,6 +410,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 this.second = second;
                 this.resultSelector = resultSelector;
                 this.cancellationToken = cancellationToken;
+                TaskTracker.TrackActiveTask(this, 3);
             }
 
             public TResult Current { get; private set; }
@@ -521,6 +526,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             public async UniTask DisposeAsync()
             {
+                TaskTracker.RemoveTracking(this);
                 if (firstEnumerator != null)
                 {
                     await firstEnumerator.DisposeAsync();

src/UniTask/Assets/Plugins/UniTask/Runtime/Progress.cs

@@ -1,7 +1,4 @@
-#if NET_4_6 || NET_STANDARD_2_0 || CSHARP_7_OR_LATER
-#pragma warning disable CS1591 // Missing XML comment for publicly visible type or member
-
-using System;
+using System;
 using System.Collections.Generic;
 using Cysharp.Threading.Tasks.Internal;
 
@@ -21,7 +18,11 @@ namespace Cysharp.Threading.Tasks
         public static IProgress<T> CreateOnlyValueChanged<T>(Action<T> handler, IEqualityComparer<T> comparer = null)
         {
             if (handler == null) return NullProgress<T>.Instance;
+#if UNITY_2018_3_OR_NEWER
             return new OnlyValueChangedProgress<T>(handler, comparer ?? UnityEqualityComparer.GetDefault<T>());
+#else
+            return new OnlyValueChangedProgress<T>(handler, comparer ?? EqualityComparer<T>.Default);
+#endif
         }
 
         sealed class NullProgress<T> : IProgress<T>
@@ -83,6 +84,4 @@ namespace Cysharp.Threading.Tasks
             }
         }
     }
-}
-
-#endif
+}

src/UniTask/Assets/Plugins/UniTask/Runtime/TriggerEvent.cs

@@ -4,26 +4,26 @@ using System.Threading;
 
 namespace Cysharp.Threading.Tasks
 {
-    public interface ITriggerEvent<T>
+    public interface ITriggerHandler<T>
     {
-        void SetResult(T value);
-        void SetCanceled(CancellationToken cancellationToken);
-        void Add(IResolveCancelPromise<T> handler);
-        void Remove(IResolveCancelPromise<T> handler);
+        void OnNext(T value);
+        void OnError(Exception ex);
+        void OnCompleted();
+        void OnCanceled(CancellationToken cancellationToken);
     }
 
     // be careful to use, itself is struct.
-    public struct TriggerEvent<T> : ITriggerEvent<T>
+    public struct TriggerEvent<T>
     {
         // optimize: many cases, handler is single.
-        IResolveCancelPromise<T> singleHandler;
+        ITriggerHandler<T> singleHandler;
 
-        IResolveCancelPromise<T>[] handlers;
+        ITriggerHandler<T>[] handlers;
 
-        // when running(in TrySetResult), does not add immediately.
+        // when running(in TrySetResult), does not add immediately(trampoline).
         bool isRunning;
-        IResolveCancelPromise<T> waitHandler;
-        MinimumQueue<IResolveCancelPromise<T>> waitQueue;
+        ITriggerHandler<T> waitHandler;
+        MinimumQueue<ITriggerHandler<T>> waitQueue;
 
         public void SetResult(T value)
         {
@@ -33,7 +33,7 @@ namespace Cysharp.Threading.Tasks
             {
                 try
                 {
-                    singleHandler.TrySetResult(value);
+                    singleHandler.OnNext(value);
                 }
                 catch (Exception ex)
                 {
@@ -53,7 +53,7 @@ namespace Cysharp.Threading.Tasks
                     {
                         try
                         {
-                            handlers[i].TrySetResult(value);
+                            handlers[i].OnNext(value);
                         }
                         catch (Exception ex)
                         {
@@ -94,7 +94,7 @@ namespace Cysharp.Threading.Tasks
             {
                 try
                 {
-                    ((ICancelPromise)singleHandler).TrySetCanceled(cancellationToken);
+                    (singleHandler).OnCanceled(cancellationToken);
                 }
                 catch (Exception ex)
                 {
@@ -114,7 +114,7 @@ namespace Cysharp.Threading.Tasks
                     {
                         try
                         {
-                            ((ICancelPromise)handlers[i]).TrySetCanceled(cancellationToken);
+                            (handlers[i]).OnCanceled(cancellationToken);
                         }
                         catch (Exception ex)
                         {
@@ -147,7 +147,129 @@ namespace Cysharp.Threading.Tasks
             }
         }
 
-        public void Add(IResolveCancelPromise<T> handler)
+        public void SetCompleted()
+        {
+            isRunning = true;
+
+            if (singleHandler != null)
+            {
+                try
+                {
+                    (singleHandler).OnCompleted();
+                }
+                catch (Exception ex)
+                {
+#if UNITY_2018_3_OR_NEWER
+                    UnityEngine.Debug.LogException(ex);
+#else
+                    Console.WriteLine(ex);
+#endif
+                }
+            }
+
+            if (handlers != null)
+            {
+                for (int i = 0; i < handlers.Length; i++)
+                {
+                    if (handlers[i] != null)
+                    {
+                        try
+                        {
+                            (handlers[i]).OnCompleted();
+                        }
+                        catch (Exception ex)
+                        {
+#if UNITY_2018_3_OR_NEWER
+                            UnityEngine.Debug.LogException(ex);
+#else
+                            Console.WriteLine(ex);
+#endif
+                            handlers[i] = null;
+                        }
+                    }
+                }
+            }
+
+            isRunning = false;
+
+            if (waitHandler != null)
+            {
+                var h = waitHandler;
+                waitHandler = null;
+                Add(h);
+            }
+
+            if (waitQueue != null)
+            {
+                while (waitQueue.Count != 0)
+                {
+                    Add(waitQueue.Dequeue());
+                }
+            }
+        }
+
+        public void SetError(Exception exception)
+        {
+            isRunning = true;
+
+            if (singleHandler != null)
+            {
+                try
+                {
+                    singleHandler.OnError(exception);
+                }
+                catch (Exception ex)
+                {
+#if UNITY_2018_3_OR_NEWER
+                    UnityEngine.Debug.LogException(ex);
+#else
+                    Console.WriteLine(ex);
+#endif
+                }
+            }
+
+            if (handlers != null)
+            {
+                for (int i = 0; i < handlers.Length; i++)
+                {
+                    if (handlers[i] != null)
+                    {
+                        try
+                        {
+                            handlers[i].OnError(exception);
+                        }
+                        catch (Exception ex)
+                        {
+                            handlers[i] = null;
+#if UNITY_2018_3_OR_NEWER
+                            UnityEngine.Debug.LogException(ex);
+#else
+                            Console.WriteLine(ex);
+#endif
+                        }
+                    }
+                }
+            }
+
+            isRunning = false;
+
+            if (waitHandler != null)
+            {
+                var h = waitHandler;
+                waitHandler = null;
+                Add(h);
+            }
+
+            if (waitQueue != null)
+            {
+                while (waitQueue.Count != 0)
+                {
+                    Add(waitQueue.Dequeue());
+                }
+            }
+        }
+
+        public void Add(ITriggerHandler<T> handler)
         {
             if (isRunning)
             {
@@ -159,7 +281,7 @@ namespace Cysharp.Threading.Tasks
 
                 if (waitQueue == null)
                 {
-                    waitQueue = new MinimumQueue<IResolveCancelPromise<T>>(4);
+                    waitQueue = new MinimumQueue<ITriggerHandler<T>>(4);
                 }
                 waitQueue.Enqueue(handler);
                 return;
@@ -173,7 +295,7 @@ namespace Cysharp.Threading.Tasks
             {
                 if (handlers == null)
                 {
-                    handlers = new IResolveCancelPromise<T>[4];
+                    handlers = new ITriggerHandler<T>[4];
                 }
 
                 // check empty
@@ -195,15 +317,15 @@ namespace Cysharp.Threading.Tasks
             }
         }
 
-        static void EnsureCapacity(ref IResolveCancelPromise<T>[] array)
+        static void EnsureCapacity(ref ITriggerHandler<T>[] array)
         {
             var newSize = array.Length * 2;
-            var newArray = new IResolveCancelPromise<T>[newSize];
+            var newArray = new ITriggerHandler<T>[newSize];
             Array.Copy(array, 0, newArray, 0, array.Length);
             array = newArray;
         }
 
-        public void Remove(IResolveCancelPromise<T> handler)
+        public void Remove(ITriggerHandler<T> handler)
         {
             if (singleHandler == handler)
             {

src/UniTask/Assets/Plugins/UniTask/Runtime/Triggers/AsyncTriggerBase.cs

@@ -24,10 +24,10 @@ namespace Cysharp.Threading.Tasks.Triggers
             if (calledDestroy) return;
             calledDestroy = true;
 
-            triggerEvent.SetCanceled(CancellationToken.None);
+            triggerEvent.SetCompleted();
         }
 
-        internal void AddHandler(IResolveCancelPromise<T> handler)
+        internal void AddHandler(ITriggerHandler<T> handler)
         {
             if (!calledAwake)
             {
@@ -37,7 +37,7 @@ namespace Cysharp.Threading.Tasks.Triggers
             triggerEvent.Add(handler);
         }
 
-        internal void RemoveHandler(IResolveCancelPromise<T> handler)
+        internal void RemoveHandler(ITriggerHandler<T> handler)
         {
             if (!calledAwake)
             {
@@ -57,7 +57,7 @@ namespace Cysharp.Threading.Tasks.Triggers
             return new AsyncTriggerEnumerator(this, cancellationToken);
         }
 
-        sealed class AsyncTriggerEnumerator : MoveNextSource, IUniTaskAsyncEnumerator<T>, IResolveCancelPromise<T>
+        sealed class AsyncTriggerEnumerator : MoveNextSource, IUniTaskAsyncEnumerator<T>, ITriggerHandler<T>
         {
             static Action<object> cancellationCallback = CancellationCallback;
 
@@ -73,15 +73,25 @@ namespace Cysharp.Threading.Tasks.Triggers
                 this.cancellationToken = cancellationToken;
             }
 
-            public bool TrySetCanceled(CancellationToken cancellationToken = default)
+            public void OnCanceled(CancellationToken cancellationToken = default)
             {
-                return completionSource.TrySetCanceled(cancellationToken);
+                completionSource.TrySetCanceled(cancellationToken);
             }
 
-            public bool TrySetResult(T value)
+            public void OnNext(T value)
             {
                 Current = value;
-                return completionSource.TrySetResult(true);
+                completionSource.TrySetResult(true);
+            }
+
+            public void OnCompleted()
+            {
+                completionSource.TrySetResult(false);
+            }
+
+            public void OnError(Exception ex)
+            {
+                completionSource.TrySetException(ex);
             }
 
             static void CancellationCallback(object state)
@@ -164,7 +174,7 @@ namespace Cysharp.Threading.Tasks.Triggers
         }
     }
 
-    public sealed partial class AsyncTriggerHandler<T> : IUniTaskSource<T>, IResolveCancelPromise<T>, IDisposable
+    public sealed partial class AsyncTriggerHandler<T> : IUniTaskSource<T>, ITriggerHandler<T>, IDisposable
     {
         static Action<object> cancellationCallback = CancellationCallback;
 
@@ -253,14 +263,24 @@ namespace Cysharp.Threading.Tasks.Triggers
             }
         }
 
-        bool IResolvePromise<T>.TrySetResult(T result)
+        void ITriggerHandler<T>.OnNext(T value)
         {
-            return core.TrySetResult(result);
+            core.TrySetResult(value);
         }
 
-        bool ICancelPromise.TrySetCanceled(CancellationToken cancellationToken)
+        void ITriggerHandler<T>.OnCanceled(CancellationToken cancellationToken)
         {
-            return core.TrySetCanceled(cancellationToken);
+            core.TrySetCanceled(cancellationToken);
+        }
+
+        void ITriggerHandler<T>.OnCompleted()
+        {
+            core.TrySetCanceled(CancellationToken.None);
+        }
+
+        void ITriggerHandler<T>.OnError(Exception ex)
+        {
+            core.TrySetException(ex);
         }
 
         void IUniTaskSource.GetResult(short token)

src/UniTask/Assets/Plugins/UniTask/Runtime/UniTask.Factory.cs

@@ -132,6 +132,101 @@ namespace Cysharp.Threading.Tasks
         {
             asyncAction(state).Forget();
         }
+
+        /// <summary>
+        /// Defer the task creation just before call await.
+        /// </summary>
+        public static UniTask Defer(Func<UniTask> factory)
+        {
+            return new UniTask(new DeferPromise(factory), 0);
+        }
+
+        /// <summary>
+        /// Defer the task creation just before call await.
+        /// </summary>
+        public static UniTask<T> Defer<T>(Func<UniTask<T>> factory)
+        {
+            return new UniTask<T>(new DeferPromise<T>(factory), 0);
+        }
+
+        sealed class DeferPromise : IUniTaskSource
+        {
+            Func<UniTask> factory;
+            UniTask task;
+            UniTask.Awaiter awaiter;
+
+            public DeferPromise(Func<UniTask> factory)
+            {
+                this.factory = factory;
+            }
+
+            public void GetResult(short token)
+            {
+                awaiter.GetResult();
+            }
+
+            public UniTaskStatus GetStatus(short token)
+            {
+                var f = Interlocked.Exchange(ref factory, null);
+                if (f == null) throw new InvalidOperationException("Can't call twice.");
+
+                task = f();
+                awaiter = f().GetAwaiter();
+                return task.Status;
+            }
+
+            public void OnCompleted(Action<object> continuation, object state, short token)
+            {
+                awaiter.SourceOnCompleted(continuation, state);
+            }
+
+            public UniTaskStatus UnsafeGetStatus()
+            {
+                return task.Status;
+            }
+        }
+
+        sealed class DeferPromise<T> : IUniTaskSource<T>
+        {
+            Func<UniTask<T>> factory;
+            UniTask<T> task;
+            UniTask<T>.Awaiter awaiter;
+
+            public DeferPromise(Func<UniTask<T>> factory)
+            {
+                this.factory = factory;
+            }
+
+            public T GetResult(short token)
+            {
+                return awaiter.GetResult();
+            }
+
+            void IUniTaskSource.GetResult(short token)
+            {
+                awaiter.GetResult();
+            }
+
+            public UniTaskStatus GetStatus(short token)
+            {
+                var f = Interlocked.Exchange(ref factory, null);
+                if (f == null) throw new InvalidOperationException("Can't call twice.");
+
+                task = f();
+                awaiter = f().GetAwaiter();
+                return task.Status;
+            }
+
+            public void OnCompleted(Action<object> continuation, object state, short token)
+            {
+                awaiter.SourceOnCompleted(continuation, state);
+            }
+
+            public UniTaskStatus UnsafeGetStatus()
+            {
+                return task.Status;
+            }
+        }
     }
 
     internal static class CompletedTasks

src/UniTask/Assets/Plugins/UniTask/Runtime/UniTask.WaitUntil.cs

@@ -19,11 +19,16 @@ namespace Cysharp.Threading.Tasks
             return new UniTask(WaitWhilePromise.Create(predicate, timing, cancellationToken, out var token), token);
         }
 
+        public static UniTask WaitUntilCanceled(CancellationToken cancellationToken, PlayerLoopTiming timing = PlayerLoopTiming.Update)
+        {
+            return new UniTask(WaitUntilCanceledPromise.Create(cancellationToken, timing, out var token), token);
+        }
+
         public static UniTask<U> WaitUntilValueChanged<T, U>(T target, Func<T, U> monitorFunction, PlayerLoopTiming monitorTiming = PlayerLoopTiming.Update, IEqualityComparer<U> equalityComparer = null, CancellationToken cancellationToken = default(CancellationToken))
           where T : class
         {
             var unityObject = target as UnityEngine.Object;
-            var isUnityObject = !object.ReferenceEquals(target, null); // don't use (unityObject == null)
+            var isUnityObject = target is UnityEngine.Object; // don't use (unityObject == null)
 
             return new UniTask<U>(isUnityObject
                 ? WaitUntilValueChangedUnityObjectPromise<T, U>.Create(target, monitorFunction, equalityComparer, monitorTiming, cancellationToken, out var token)
@@ -234,12 +239,98 @@ namespace Cysharp.Threading.Tasks
             }
         }
 
+        sealed class WaitUntilCanceledPromise : IUniTaskSource, IPlayerLoopItem, IPromisePoolItem
+        {
+            static readonly PromisePool<WaitUntilCanceledPromise> pool = new PromisePool<WaitUntilCanceledPromise>();
+
+            CancellationToken cancellationToken;
+
+            UniTaskCompletionSourceCore<object> core;
+
+            WaitUntilCanceledPromise()
+            {
+            }
+
+            public static IUniTaskSource Create(CancellationToken cancellationToken, PlayerLoopTiming timing, out short token)
+            {
+                if (cancellationToken.IsCancellationRequested)
+                {
+                    return AutoResetUniTaskCompletionSource.CreateFromCanceled(cancellationToken, out token);
+                }
+
+                var result = pool.TryRent() ?? new WaitUntilCanceledPromise();
+
+                result.cancellationToken = cancellationToken;
+
+                TaskTracker.TrackActiveTask(result, 3);
+
+                PlayerLoopHelper.AddAction(timing, result);
+
+                token = result.core.Version;
+                return result;
+            }
+
+            public void GetResult(short token)
+            {
+                try
+                {
+                    TaskTracker.RemoveTracking(this);
+                    core.GetResult(token);
+                }
+                finally
+                {
+                    pool.TryReturn(this);
+                }
+            }
+
+            public UniTaskStatus GetStatus(short token)
+            {
+                return core.GetStatus(token);
+            }
+
+            public UniTaskStatus UnsafeGetStatus()
+            {
+                return core.UnsafeGetStatus();
+            }
+
+            public void OnCompleted(Action<object> continuation, object state, short token)
+            {
+                core.OnCompleted(continuation, state, token);
+            }
+
+            public bool MoveNext()
+            {
+                if (cancellationToken.IsCancellationRequested)
+                {
+                    core.TrySetResult(null);
+                    return false;
+                }
+
+                return true;
+            }
+
+            public void Reset()
+            {
+                core.Reset();
+                cancellationToken = default;
+            }
+
+            ~WaitUntilCanceledPromise()
+            {
+                if (pool.TryReturn(this))
+                {
+                    GC.ReRegisterForFinalize(this);
+                }
+            }
+        }
+
         // where T : UnityEngine.Object, can not add constraint
         sealed class WaitUntilValueChangedUnityObjectPromise<T, U> : IUniTaskSource<U>, IPlayerLoopItem, IPromisePoolItem
         {
             static readonly PromisePool<WaitUntilValueChangedUnityObjectPromise<T, U>> pool = new PromisePool<WaitUntilValueChangedUnityObjectPromise<T, U>>();
 
             T target;
+            UnityEngine.Object targetAsUnityObject;
             U currentValue;
             Func<T, U> monitorFunction;
             IEqualityComparer<U> equalityComparer;
@@ -261,6 +352,7 @@ namespace Cysharp.Threading.Tasks
                 var result = pool.TryRent() ?? new WaitUntilValueChangedUnityObjectPromise<T, U>();
 
                 result.target = target;
+                result.targetAsUnityObject = target as UnityEngine.Object;
                 result.monitorFunction = monitorFunction;
                 result.currentValue = monitorFunction(target);
                 result.equalityComparer = equalityComparer ?? UnityEqualityComparer.GetDefault<U>();
@@ -309,7 +401,7 @@ namespace Cysharp.Threading.Tasks
 
             public bool MoveNext()
             {
-                if (cancellationToken.IsCancellationRequested || target == null) // destroyed = cancel.
+                if (cancellationToken.IsCancellationRequested || targetAsUnityObject == null) // destroyed = cancel.
                 {
                     core.TrySetCanceled(cancellationToken);
                     return false;
@@ -353,7 +445,6 @@ namespace Cysharp.Threading.Tasks
             }
         }
 
-
         sealed class WaitUntilValueChangedStandardObjectPromise<T, U> : IUniTaskSource<U>, IPlayerLoopItem, IPromisePoolItem
             where T : class
         {

src/UniTask/Assets/Plugins/UniTask/Runtime/UniTaskCompletionSource.cs

@@ -38,14 +38,6 @@ namespace Cysharp.Threading.Tasks
     {
     }
 
-    public interface IResolveCancelPromise : IResolvePromise, ICancelPromise
-    {
-    }
-
-    public interface IResolveCancelPromise<T> : IResolvePromise<T>, ICancelPromise
-    {
-    }
-
     [StructLayout(LayoutKind.Auto)]
     public struct UniTaskCompletionSourceCore<TResult>
     {
@@ -97,6 +89,11 @@ namespace Cysharp.Threading.Tasks
             }
         }
 
+        internal void MarkHandled()
+        {
+            hasUnhandledError = false;
+        }
+
         /// <summary>Completes with a successful result.</summary>
         /// <param name="result">The result.</param>
         [DebuggerHidden]
@@ -293,12 +290,12 @@ namespace Cysharp.Threading.Tasks
         }
 
         [DebuggerHidden]
-        [Conditional("UNITY_EDITOR")]
         internal void MarkHandled()
         {
             if (!handled)
             {
                 handled = true;
+                core.MarkHandled();
                 TaskTracker.RemoveTracking(this);
             }
         }
@@ -504,12 +501,12 @@ namespace Cysharp.Threading.Tasks
         }
 
         [DebuggerHidden]
-        [Conditional("UNITY_EDITOR")]
         internal void MarkHandled()
         {
             if (!handled)
             {
                 handled = true;
+                core.MarkHandled();
                 TaskTracker.RemoveTracking(this);
             }
         }

src/UniTask/Assets/Plugins/UniTask/Runtime/UnityAsyncExtensions.uGUI.cs

@@ -1,9 +1,8 @@
 #pragma warning disable CS1591 // Missing XML comment for publicly visible type or member
 
+using Cysharp.Threading.Tasks.Linq;
 using System;
 using System.Threading;
-using Cysharp.Threading.Tasks.Internal;
-using Cysharp.Threading.Tasks.Linq;
 using UnityEngine;
 using UnityEngine.Events;
 using UnityEngine.UI;
@@ -27,6 +26,21 @@ namespace Cysharp.Threading.Tasks
             return new UnityEventHandlerAsyncEnumerable(unityEvent, cancellationToken);
         }
 
+        public static AsyncUnityEventHandler<T> GetAsyncEventHandler<T>(this UnityEvent<T> unityEvent, CancellationToken cancellationToken)
+        {
+            return new AsyncUnityEventHandler<T>(unityEvent, cancellationToken, false);
+        }
+
+        public static UniTask<T> OnInvokeAsync<T>(this UnityEvent<T> unityEvent, CancellationToken cancellationToken)
+        {
+            return new AsyncUnityEventHandler<T>(unityEvent, cancellationToken, true).OnInvokeAsync();
+        }
+
+        public static IUniTaskAsyncEnumerable<T> OnInvokeAsAsyncEnumerable<T>(this UnityEvent<T> unityEvent, CancellationToken cancellationToken)
+        {
+            return new UnityEventHandlerAsyncEnumerable<T>(unityEvent, cancellationToken);
+        }
+
         public static IAsyncClickEventHandler GetAsyncClickEventHandler(this Button button)
         {
             return new AsyncUnityEventHandler(button.onClick, button.GetCancellationTokenOnDestroy(), false);
@@ -207,6 +221,36 @@ namespace Cysharp.Threading.Tasks
             return new UnityEventHandlerAsyncEnumerable<string>(inputField.onEndEdit, cancellationToken);
         }
 
+        public static IAsyncValueChangedEventHandler<string> GetAsyncValueChangedEventHandler(this InputField inputField)
+        {
+            return new AsyncUnityEventHandler<string>(inputField.onValueChanged, inputField.GetCancellationTokenOnDestroy(), false);
+        }
+
+        public static IAsyncValueChangedEventHandler<string> GetAsyncValueChangedEventHandler(this InputField inputField, CancellationToken cancellationToken)
+        {
+            return new AsyncUnityEventHandler<string>(inputField.onValueChanged, cancellationToken, false);
+        }
+
+        public static UniTask<string> OnValueChangedAsync(this InputField inputField)
+        {
+            return new AsyncUnityEventHandler<string>(inputField.onValueChanged, inputField.GetCancellationTokenOnDestroy(), true).OnInvokeAsync();
+        }
+
+        public static UniTask<string> OnValueChangedAsync(this InputField inputField, CancellationToken cancellationToken)
+        {
+            return new AsyncUnityEventHandler<string>(inputField.onValueChanged, cancellationToken, true).OnInvokeAsync();
+        }
+
+        public static IUniTaskAsyncEnumerable<string> OnValueChangedAsAsyncEnumerable(this InputField inputField)
+        {
+            return new UnityEventHandlerAsyncEnumerable<string>(inputField.onValueChanged, inputField.GetCancellationTokenOnDestroy());
+        }
+
+        public static IUniTaskAsyncEnumerable<string> OnValueChangedAsAsyncEnumerable(this InputField inputField, CancellationToken cancellationToken)
+        {
+            return new UnityEventHandlerAsyncEnumerable<string>(inputField.onValueChanged, cancellationToken);
+        }
+
         public static IAsyncValueChangedEventHandler<int> GetAsyncValueChangedEventHandler(this Dropdown dropdown)
         {
             return new AsyncUnityEventHandler<int>(dropdown.onValueChanged, dropdown.GetCancellationTokenOnDestroy(), false);

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

@@ -1,10 +1,10 @@
 {
     "name": "com.cysharp.unitask",
     "displayName": "UniTask",
-    "version": "2.0.5-rc2",
+    "version": "2.0.9-rc6",
     "unity": "2018.3",
     "description": "Provides an efficient async/await integration to Unity.",
-    "keywords": ["async/await", "async", "Task", "UniTask"],
+    "keywords": [ "async/await", "async", "Task", "UniTask" ],
     "license": "MIT",
     "category": "Task",
     "dependencies": {}

src/UniTask/Assets/Scenes/SandboxMain.cs

@@ -1,20 +1,16 @@
-using System;
-
+using Cysharp.Threading.Tasks;
 using Cysharp.Threading.Tasks.Linq;
 using Cysharp.Threading.Tasks.Triggers;
+using System;
 using System.Collections;
 using System.Collections.Generic;
-using System.IO;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
-using Cysharp.Threading.Tasks;
 using Unity.Collections;
 using Unity.Jobs;
 using UnityEngine;
-using UnityEngine.Networking;
 using UnityEngine.UI;
-using TMPro;
 
 public struct MyJob : IJob
 {
@@ -39,6 +35,36 @@ public enum MyEnum
 }
 
 
+public class SimplePresenter
+{
+    // View
+    public UnityEngine.UI.InputField Input;
+
+
+    // Presenter
+
+
+    public SimplePresenter()
+    {
+        //Input.OnValueChangedAsAsyncEnumerable()
+        //   .Queue()
+        //   .SelectAwait(async x =>
+        //   {
+        //       await UniTask.Delay(TimeSpan.FromSeconds(1));
+        //       return x;
+        //   })
+        //   .Select(x=> x.ToUpper())
+        //   .BindTo(
+
+
+
+    }
+
+}
+
+
+
+
 
 
 
@@ -50,14 +76,43 @@ public static partial class UnityUIComponentExtensions
 
 
 
+public class AsyncMessageBroker<T> : IDisposable
+{
+    Channel<T> channel;
 
+    IConnectableUniTaskAsyncEnumerable<T> multicastSource;
+    IDisposable connection;
+
+    public AsyncMessageBroker()
+    {
+        channel = Channel.CreateSingleConsumerUnbounded<T>();
+        multicastSource = channel.Reader.ReadAllAsync().Publish();
+        connection = multicastSource.Connect();
+    }
+
+    public void Publish(T value)
+    {
+        channel.Writer.TryWrite(value);
+    }
+
+    public IUniTaskAsyncEnumerable<T> Subscribe()
+    {
+        return multicastSource;
+    }
+
+    public void Dispose()
+    {
+        channel.Writer.TryComplete();
+        connection.Dispose();
+    }
+}
 
 
 public class SandboxMain : MonoBehaviour
 {
     public Button okButton;
     public Button cancelButton;
-    public Text text;
+
 
     CancellationTokenSource cts;
 
@@ -80,6 +135,48 @@ public class SandboxMain : MonoBehaviour
     }
 
 
+
+
+
+    public class Model
+    {
+        // State<int> Hp { get; }
+
+        
+
+
+        public Model()
+        {
+            // hp = new AsyncReactiveProperty<int>();
+
+
+
+
+
+
+            //setHp = Hp.GetSetter();
+        }
+
+        void Increment(int value)
+        {
+
+
+            // setHp(Hp.Value += value);
+        }
+    }
+
+
+
+    public Text text;
+    public Button button;
+
+
+    void Start2()
+    {
+        
+    }
+
+
     async UniTask RunStandardDelayAsync()
     {
         UnityEngine.Debug.Log("DEB");
@@ -96,9 +193,6 @@ public class SandboxMain : MonoBehaviour
 
         var scheduled = job.Schedule();
 
-
-
-
         UnityEngine.Debug.Log("OK");
         await scheduled; // .ConfigureAwait(PlayerLoopTiming.Update); // .WaitAsync(PlayerLoopTiming.Update);
         UnityEngine.Debug.Log("OK2");
@@ -139,152 +233,89 @@ public class SandboxMain : MonoBehaviour
         Debug.Log("Done");
     }
 
-    void Start()
+    public int MyProperty { get; set; }
+
+    public class MyClass
     {
-        Application.SetStackTraceLogType(LogType.Error, StackTraceLogType.Full);
-        Application.SetStackTraceLogType(LogType.Exception, StackTraceLogType.Full);
-
-        var playerLoop = UnityEngine.LowLevel.PlayerLoop.GetCurrentPlayerLoop();
-        //ShowPlayerLoop.DumpPlayerLoop("Current", playerLoop);
-
-
-        RP1 = new AsyncReactiveProperty<int>(999);
-
-        HogeAsync().Forget();
-
-        RP1.Select(x => x * x).BindTo(text);
-
-
-        //Update2().Forget();
-
-        //RunStandardDelayAsync().Forget();
-
-        //for (int i = 0; i < 14; i++)
-        //{
-        //    TimingDump((PlayerLoopTiming)i).Forget();
-        //}
-
-        //StartCoroutine(CoroutineDump("yield WaitForEndOfFrame", new WaitForEndOfFrame()));
-        //StartCoroutine(CoroutineDump("yield WaitForFixedUpdate", new WaitForFixedUpdate()));
-        //StartCoroutine(CoroutineDump("yield null", null));
-
-        // -----
-
-        // RunJobAsync().Forget();
-
-        //ClickOnce().Forget();
-        //ClickForever().Forget();
-
-        //var cor = UniTask.ToCoroutine(async () =>
-        // {
-        //     var job = new MyJob() { loopCount = 999, inOut = new NativeArray<int>(1, Allocator.TempJob) };
-        //     JobHandle.ScheduleBatchedJobs();
-        //     await job.Schedule().WaitAsync(PlayerLoopTiming.Update);
-        //     job.inOut.Dispose();
-        // });
-
-        //StartCoroutine(cor);
-
-        // UniTaskAsyncEnumerable.EveryUpdate(PlayerLoopTiming.FixedUpdate)
-
-
-        //await UniTask.Yield(PlayerLoopTiming.Update);
-        //Debug.Log("Start:" + Time.frameCount);
-
-        //await UniTaskAsyncEnumerable.TimerFrame(3, 5, PlayerLoopTiming.PostLateUpdate)
-        //    .Select(x => x)
-        //    .Do(x => Debug.Log("DODODO"))
-        //    .ForEachAsync(_ =>
-        //    {
-        //        Debug.Log("Call:" + Time.frameCount);
-        //    }, cancellationToken: this.GetCancellationTokenOnDestroy());
-
-        //try
-        //{
-        //    await this.GetAsyncUpdateTrigger().ForEachAsync(_ =>
-        //    {
-        //        UnityEngine.Debug.Log("EveryUpdate:" + Time.frameCount);
-        //    });
-        //}
-        //catch (OperationCanceledException ex)
-        //{
-        //    UnityEngine.Debug.Log("END");
-        //}
-
-
-
-        CancellationTokenSource cts = new CancellationTokenSource();
-
-        //var trigger = this.GetAsyncUpdateTrigger();
-        //Go(trigger, 1, cts.Token).Forget();
-        //Go(trigger, 2, cts.Token).Forget();
-        //Go(trigger, 3, cts.Token).Forget();
-        //Go(trigger, 4, cts.Token).Forget();
-        //Go(trigger, 5, cts.Token).Forget();
-
-
-        Application.logMessageReceived += Application_logMessageReceived;
-
-
-
-        // foo.Status.IsCanceled
-
-
-        // 5回クリックされるまで待つ、とか。
-        //Debug.Log("Await start.");
-
-
-
-        //await okButton.GetAsyncClickEventHandler().DisableAutoClose()
-        //  .Select((_, clickCount) => clickCount + 1)
-        // .FirstAsync(x => x == 5);
-
-        //Debug.Log("Click 5 times.");
-
-
-
-        // await this.GetAsyncUpdateTrigger().UpdateAsAsyncEnumerable()
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-        //ucs = new UniTaskCompletionSource();
-
-        //okButton.onClick.AddListener(async () =>
-        //{
-        //    await InnerAsync(false);
-        //});
-
-        okButton.onClick.AddListener(() =>
-        {
-            // FooAsync().Forget();
-
-            RP1.Value += 3;
-
-        });
-
-        cancelButton.onClick.AddListener(() =>
-        {
-            text.text = "";
-
-            // ucs.TrySetResult();
-
-            cts.Cancel();
-        });
+        public int MyProperty { get; set; }
     }
 
-    static void Foo(UniTask t)
+    MyClass mcc;
+
+    void Start()
     {
+        this.mcc = new MyClass();
+        this.MyProperty = 999;
+
+        CheckDest().Forget();
+
+
+        //UniTaskAsyncEnumerable.EveryValueChanged(mcc, x => x.MyProperty)
+        //    .Do(_ => { }, () => Debug.Log("COMPLETED"))
+        //    .ForEachAsync(x =>
+        //    {
+        //        Debug.Log("VALUE_CHANGED:" + x);
+        //    })
+        //    .Forget();
+
+
+
+        
+
+
+        okButton.OnClickAsAsyncEnumerable().ForEachAsync(_ =>
+        {
+
+            mcc.MyProperty += 10;
+
+
+
+        }).Forget();
+
+        cancelButton.OnClickAsAsyncEnumerable().ForEachAsync(_ =>
+        {
+            this.mcc = null;
+        });
+
+    }
+
+    async UniTaskVoid CheckDest()
+    {
+        try
+        {
+            Debug.Log("WAIT");
+            await UniTask.WaitUntilValueChanged(mcc, x => x.MyProperty);
+            Debug.Log("CHANGED?");
+        }
+        finally
+        {
+            Debug.Log("END");
+        }
+    }
+
+    async UniTaskVoid Running(CancellationToken ct)
+    {
+        Debug.Log("BEGIN");
+        await UniTask.WaitUntilCanceled(ct);
+        Debug.Log("DONE");
+    }
+
+    async UniTaskVoid WaitForChannelAsync(ChannelReader<int> reader, CancellationToken token)
+    {
+        try
+        {
+            //var result1 = await reader.ReadAsync(token);
+            //Debug.Log(result1);
+
+            await reader.ReadAllAsync().ForEachAsync(x => Debug.Log(x)/*, token*/);
+
+            Debug.Log("done");
+        }
+        catch (Exception ex)
+        {
+            Debug.Log("here");
+            Debug.LogException(ex);
+        }
     }
 
     async UniTaskVoid Go(AsyncUpdateTrigger trigger, int i, CancellationToken ct)