2.0.8-rc5

README.md

@@ -378,15 +378,7 @@ ECS, PlayerLoop
 TODO:
 
 ```csharp
-// Setup Entities Loop.
+var loop = PlayerLoop.GetCurrentPlayerLoop();
-var loop = PlayerLoop.GetDefaultPlayerLoop();
-foreach (var world in Unity.Entities.World.All)
-{
-    ScriptBehaviourUpdateOrder.UpdatePlayerLoop(world, loop);
-    loop = PlayerLoop.GetCurrentPlayerLoop();
-}
-
-// UniTask PlayerLoop Initialize.
 PlayerLoopHelper.Initialize(ref loop);
 ```
 

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 State<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 State<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.ToState(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/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/Assets/Plugins/UniTask/Runtime/AsyncReactiveProperty.cs

@@ -59,6 +59,24 @@ namespace Cysharp.Threading.Tasks
             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>
         {
             readonly AsyncReactiveProperty<T> parent;
@@ -144,6 +162,11 @@ namespace Cysharp.Threading.Tasks
                 completionSource.TrySetResult(false);
             }
 
+            public void OnError(Exception ex)
+            {
+                completionSource.TrySetException(ex);
+            }
+
             static void CancellationCallback(object state)
             {
                 var self = (Enumerator)state;
@@ -151,4 +174,231 @@ namespace Cysharp.Threading.Tasks
             }
         }
     }
+
+    public class State<T> : IReadOnlyAsyncReactiveProperty<T>, IDisposable
+    {
+        TriggerEvent<T> triggerEvent;
+
+        T latestValue;
+
+        Action<T> setter;
+        IUniTaskAsyncEnumerator<T> enumerator;
+
+        public T Value
+        {
+            get
+            {
+                return latestValue;
+            }
+        }
+
+        public State(T value)
+        {
+            this.latestValue = value;
+            this.triggerEvent = default;
+        }
+
+        public State(T initialValue, IUniTaskAsyncEnumerable<T> source, CancellationToken cancellationToken)
+        {
+            latestValue = initialValue;
+            ConsumeEnumerator(source, cancellationToken).Forget();
+        }
+
+        public State(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())
+                {
+                    SetValue(enumerator.Current);
+                }
+            }
+            finally
+            {
+                await enumerator.DisposeAsync();
+                enumerator = null;
+            }
+        }
+
+        public Action<T> GetSetter()
+        {
+            if (enumerator != null)
+            {
+                throw new InvalidOperationException("Can not get setter when create from IUniTaskAsyncEnumerable source.");
+            }
+
+            if (setter != null)
+            {
+                throw new InvalidOperationException("GetSetter can only call once.");
+            }
+
+            setter = SetValue;
+            return setter;
+        }
+
+        void SetValue(T value)
+        {
+            this.latestValue = value;
+            triggerEvent.SetResult(value);
+        }
+
+        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 State<T>(T value)
+        {
+            return new State<T>(value);
+        }
+
+        public static implicit operator T(State<T> value)
+        {
+            return value.Value;
+        }
+
+        public override string ToString()
+        {
+            if (isValueType) return latestValue.ToString();
+            return latestValue?.ToString();
+        }
+
+        static bool isValueType;
+
+        static State()
+        {
+            isValueType = typeof(T).IsValueType;
+        }
+
+        class WithoutCurrentEnumerable : IUniTaskAsyncEnumerable<T>
+        {
+            readonly State<T> parent;
+
+            public WithoutCurrentEnumerable(State<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 State<T> parent;
+            readonly CancellationToken cancellationToken;
+            readonly CancellationTokenRegistration cancellationTokenRegistration;
+            T value;
+            bool isDisposed;
+            bool firstCall;
+
+            public Enumerator(State<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 State<T> ToState<T>(this IUniTaskAsyncEnumerable<T> source, CancellationToken cancellationToken)
+        {
+            return new State<T>(source, cancellationToken);
+        }
+
+        public static State<T> ToState<T>(this IUniTaskAsyncEnumerable<T> source, T initialValue, CancellationToken cancellationToken)
+        {
+            return new State<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/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/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/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/UnityExtensions/EveryUpdate.cs

@@ -1,7 +1,4 @@
-using Cysharp.Threading.Tasks.Internal;
+using System.Threading;
-using System;
-using System.Collections.Generic;
-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/TriggerEvent.cs

@@ -7,8 +7,9 @@ namespace Cysharp.Threading.Tasks
     public interface ITriggerHandler<T>
     {
         void OnNext(T value);
-        void OnCanceled(CancellationToken cancellationToken);
+        void OnError(Exception ex);
         void OnCompleted();
+        void OnCanceled(CancellationToken cancellationToken);
     }
 
     // be careful to use, itself is struct.
@@ -207,6 +208,67 @@ namespace Cysharp.Threading.Tasks
             }
         }
 
+        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)

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

@@ -89,6 +89,11 @@ namespace Cysharp.Threading.Tasks.Triggers
                 completionSource.TrySetResult(false);
             }
 
+            public void OnError(Exception ex)
+            {
+                completionSource.TrySetException(ex);
+            }
+
             static void CancellationCallback(object state)
             {
                 var self = (AsyncTriggerEnumerator)state;
@@ -273,6 +278,11 @@ namespace Cysharp.Threading.Tasks.Triggers
             core.TrySetCanceled(CancellationToken.None);
         }
 
+        void ITriggerHandler<T>.OnError(Exception ex)
+        {
+            core.TrySetException(ex);
+        }
+
         void IUniTaskSource.GetResult(short token)
         {
             ((IUniTaskSource<T>)this).GetResult(token);

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

@@ -28,7 +28,7 @@ namespace Cysharp.Threading.Tasks
           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)
@@ -330,6 +330,7 @@ namespace Cysharp.Threading.Tasks
             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;
@@ -351,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>();
@@ -399,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;

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.6-rc3",
+    "version": "2.0.8-rc5",
     "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

@@ -35,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(
+
+
+
+    }
+
+}
+
+
+
+
 
 
 
@@ -49,23 +79,15 @@ public static partial class UnityUIComponentExtensions
 public class AsyncMessageBroker<T> : IDisposable
 {
     Channel<T> channel;
-    List<Func<T, UniTask>> asyncEvents;
+
+    IConnectableUniTaskAsyncEnumerable<T> multicastSource;
+    IDisposable connection;
 
     public AsyncMessageBroker()
     {
         channel = Channel.CreateSingleConsumerUnbounded<T>();
-        asyncEvents = new List<Func<T, UniTask>>();
+        multicastSource = channel.Reader.ReadAllAsync().Publish();
-    }
+        connection = multicastSource.Connect();
-
-    async UniTaskVoid PublishAll()
-    {
-        await channel.Reader.ReadAllAsync().ForEachAwaitAsync(async x =>
-        {
-            foreach (var item in asyncEvents)
-            {
-                await item.Invoke(x);
-            }
-        });
     }
 
     public void Publish(T value)
@@ -73,33 +95,15 @@ public class AsyncMessageBroker<T> : IDisposable
         channel.Writer.TryWrite(value);
     }
 
-    public Subscription Subscribe(Func<T, UniTask> func)
+    public IUniTaskAsyncEnumerable<T> Subscribe()
     {
-        asyncEvents.Add(func);
+        return multicastSource;
-        return new Subscription(this, func);
     }
 
     public void Dispose()
     {
         channel.Writer.TryComplete();
-        asyncEvents.Clear();
+        connection.Dispose();
-    }
-
-    public readonly struct Subscription : IDisposable
-    {
-        readonly AsyncMessageBroker<T> broker;
-        readonly Func<T, UniTask> func;
-
-        public Subscription(AsyncMessageBroker<T> broker, Func<T, UniTask> func)
-        {
-            this.broker = broker;
-            this.func = func;
-        }
-
-        public void Dispose()
-        {
-            broker.asyncEvents.Remove(func);
-        }
     }
 }
 
@@ -108,7 +112,7 @@ public class SandboxMain : MonoBehaviour
 {
     public Button okButton;
     public Button cancelButton;
-    public Text text;
+
 
     CancellationTokenSource cts;
 
@@ -131,6 +135,63 @@ public class SandboxMain : MonoBehaviour
     }
 
 
+
+
+
+    public class Model
+    {
+        // State<int> Hp { get; }
+
+        AsyncReactiveProperty<int> hp;
+        IReadOnlyAsyncReactiveProperty<int> Hp => hp;
+
+
+
+        public Model()
+        {
+            // hp = new AsyncReactiveProperty<int>();
+
+
+
+
+
+
+            //setHp = Hp.GetSetter();
+        }
+
+        void Increment(int value)
+        {
+
+
+            // setHp(Hp.Value += value);
+        }
+    }
+
+
+
+    public Text text;
+    public Button button;
+
+    [SerializeField]
+    State<int> count;
+
+    void Start2()
+    {
+        count = 10;
+
+        var countS = count.GetSetter();
+
+        count.BindTo(text);
+        button.OnClickAsAsyncEnumerable().ForEachAsync(_ =>
+        {
+            // int foo = countS;
+            //countS.Set(countS += 10);
+
+            // setter.SetValue(count.Value + 10);
+        });
+    }
+
+
     async UniTask RunStandardDelayAsync()
     {
         UnityEngine.Debug.Log("DEB");
@@ -147,9 +208,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");
@@ -190,25 +248,62 @@ public class SandboxMain : MonoBehaviour
         Debug.Log("Done");
     }
 
+    public int MyProperty { get; set; }
+
+    public class MyClass
+    {
+        public int MyProperty { get; set; }
+    }
+
+    MyClass mcc;
+
     void Start()
     {
-        //var rp = new AsyncReactiveProperty<int>(10);
+        this.mcc = new MyClass();
+        this.MyProperty = 999;
 
-        //Running(rp).Forget();
+        CheckDest().Forget();
-
-        //await UniTaskAsyncEnumerable.EveryUpdate().Take(10).ForEachAsync((x, i) => rp.Value = i);
-
-        //rp.Dispose();
-
-        //var channel = Channel.CreateSingleConsumerUnbounded<int>();
-        //Debug.Log("wait channel");
-        //await channel.Reader.ReadAllAsync(this.GetCancellationTokenOnDestroy()).ForEachAsync(_ => { });
 
 
-        var rp = new AsyncReactiveProperty<int>(10);
+        //UniTaskAsyncEnumerable.EveryValueChanged(mcc, x => x.MyProperty)
+        //    .Do(_ => { }, () => Debug.Log("COMPLETED"))
+        //    .ForEachAsync(x =>
+        //    {
+        //        Debug.Log("VALUE_CHANGED:" + x);
+        //    })
+        //    .Forget();
 
-        rp.Append(10).Select(x => x * 100).Take(30).Prepend(99).SkipLast(9).Where(x => x % 2 == 0).ForEachAsync(_ => { }).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)