2.0.5-rc2

.circleci/config.yml

@@ -1,114 +0,0 @@
-version: 2.1
-executors:
-  unity:
-    # https://hub.docker.com/r/gableroux/unity3d/tags
-    parameters:
-      version: {type: string}
-    docker:
-      - image: gableroux/unity3d:<< parameters.version >>
-  go:
-    docker:
-      - image: circleci/golang      
-commands:
-  unity_activate:
-    parameters:
-      unity_version: {type: string}
-      unity_license: {type: string}
-    steps:
-      # get activation file, if fail to activate unity, use this key and activate from https://license.unity3d.com/manual
-      - run: apt update && apt install libunwind8 -y
-      - run: /opt/Unity/Editor/Unity -quit -batchmode -nographics -logFile -createManualActivationFile || exit 0
-      - run: cat Unity_v<< parameters.unity_version >>.alf
-      # get from UNITY_LICENSE envvar(base64 encoded(cat foo.ulf | base64 )), this file is generated from above manual activation
-      - run: echo << parameters.unity_license >> | base64 -di  >> .circleci/Unity.ulf
-      - run: /opt/Unity/Editor/Unity -quit -batchmode -nographics -silent-crashes -logFile -manualLicenseFile .circleci/Unity.ulf || exit 0
-jobs:
-  build-and-test:
-    parameters:
-      unity_version: {type: string}
-      unity_license: {type: string}
-    executor:
-      name: unity
-      version: << parameters.unity_version >>
-    steps:
-      - checkout
-      - unity_activate:
-          unity_version: << parameters.unity_version >>
-          unity_license: << parameters.unity_license >>
-      - run:
-          name: Build Linux(Mono)
-          command: /opt/Unity/Editor/Unity -quit -batchmode -nographics -silent-crashes -logFile -projectPath . -executeMethod UnitTestBuilder.BuildUnitTest /headless /ScriptBackend Mono2x /BuildTarget StandaloneLinux64
-          working_directory: .
-      # TODO:check unity version and packages...
-      # - run: ./bin/UnitTest/StandaloneLinux64_Mono2x/test
-  build-and-create-package:
-    parameters:
-      unity_version: {type: string}
-      unity_license: {type: string}
-    executor:
-      name: unity
-      version: << parameters.unity_version >>
-    steps:
-      - checkout
-      - unity_activate:
-          unity_version: << parameters.unity_version >>
-          unity_license: << parameters.unity_license >>
-      - run:
-          name: Export unitypackage
-          command: /opt/Unity/Editor/Unity -quit -batchmode -nographics -silent-crashes -logFile -projectPath . -executeMethod PackageExporter.Export
-          working_directory: .
-      - store_artifacts:
-          path: ./UniRx.Async.unitypackage
-          destination: /UniRx.Async.unitypackage
-  # upload to github by ghr
-  upload-github:
-    executor: go
-    steps:
-      - attach_workspace:
-          at: .
-      - run: go get github.com/tcnksm/ghr
-      - run: ghr -t ${GITHUB_TOKEN} -u ${CIRCLE_PROJECT_USERNAME} -r ${CIRCLE_PROJECT_REPONAME} ${CIRCLE_TAG} .
-      - store_artifacts:
-          path: UniRx.Async.unitypackage
-          destination: UniRx.Async.unitypackage          
-workflows:
-  version: 2
-  build-unity:
-    jobs:
-      # does not exists yet.
-      # - build-and-test:
-      #     unity_version: 2019.3.0a2
-      #     unity_license: ${UNITY_LICENSE_2019_3}
-      # - build-and-test:
-      #     unity_version: 2019.2.0b2
-      #     unity_license: ${UNITY_LICENSE_2019_2}
-      - build-and-test:
-          unity_version: 2019.1.2f1
-          unity_license: ${UNITY_LICENSE_2019_1}
-          filters:
-            tags:
-              only: /.*/
-      # test asmdef will not found.
-      # - build-and-test:
-      #     unity_version: 2018.4.0f1
-      #     unity_license: ${UNITY_LICENSE_2018_4}
-      # # UniTask minimum support version is 2018.3(C# 7.x)
-      # - build-and-test:
-      #     unity_version: 2018.3.12f1
-      #     unity_license: ${UNITY_LICENSE_2018_3}
-      - build-and-create-package:
-          unity_version: 2019.1.2f1
-          unity_license: ${UNITY_LICENSE_2019_1}
-          filters:
-            tags:
-              only: /^\d\.\d\.\d.*/
-            branches:
-              ignore: /.*/
-      - upload-github:
-          requires:
-            - build-and-create-package
-          filters:
-            tags:
-              only: /^\d\.\d\.\d.*/
-            branches:
-              ignore: /.*/

.github/workflows/build-debug.yml

@@ -0,0 +1,73 @@
+name: Build-Debug
+
+on:
+  push:
+    branches:
+      - "**"
+    tags:
+      - "!*" # not a tag push
+  pull_request:
+    types:
+      - opened
+      - synchronize
+
+jobs:
+  build-dotnet:
+    runs-on: ubuntu-latest
+    env:
+      DOTNET_CLI_TELEMETRY_OPTOUT: 1
+      DOTNET_SKIP_FIRST_TIME_EXPERIENCE: 1
+      NUGET_XMLDOC_MODE: skip
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions/setup-dotnet@v1
+        with:
+          dotnet-version: 3.1.101
+      - run: dotnet test -c Debug ./src/UniTask.NetCoreTests/UniTask.NetCoreTests.csproj
+
+  build-unity:
+    strategy:
+      matrix:
+        unity: ['2019.3.9f1', '2020.1.0b5']
+        include:
+          - unity: 2019.3.9f1
+            license: UNITY_2019_3
+          - unity: 2020.1.0b5
+            license: UNITY_2020_1
+    runs-on: ubuntu-latest
+    container:
+      # with linux-il2cpp. image from https://hub.docker.com/r/gableroux/unity3d/tags
+      image: gableroux/unity3d:${{ matrix.unity }}-linux-il2cpp
+    steps:
+      - run: apt update && apt install git -y
+      - uses: actions/checkout@v2
+      # create unity activation file and store to artifacts.
+      - run: /opt/Unity/Editor/Unity -quit -batchmode -nographics -logFile -createManualActivationFile || exit 0
+      - uses: actions/upload-artifact@v1
+        with:
+          name: Unity_v${{ matrix.unity }}.alf
+          path: ./Unity_v${{ matrix.unity }}.alf
+      # activate Unity from manual license file(ulf)
+      - run: echo -n "$UNITY_LICENSE" >> .Unity.ulf
+        env:
+          UNITY_LICENSE: ${{ secrets[matrix.license] }}
+      - name: Activate Unity, always returns a success. But if a subsequent run fails, the activation may have failed(if succeeded, shows `Next license update check is after` and not shows other message(like GUID != GUID). If fails not). In that case, upload the artifact's .alf file to https://license.unity3d.com/manual to get the .ulf file and set it to secrets.
+        run: /opt/Unity/Editor/Unity -quit -batchmode -nographics -silent-crashes -logFile -manualLicenseFile .Unity.ulf || exit 0
+
+      # Execute scripts: RuntimeUnitTestToolkit
+      - name: Build UnitTest(Linux64, mono)
+        run: /opt/Unity/Editor/Unity -quit -batchmode -nographics -silent-crashes -logFile -projectPath . -executeMethod UnitTestBuilder.BuildUnitTest /headless /ScriptBackend mono /BuildTarget StandaloneLinux64
+        working-directory: src/UniTask
+      - name: Execute UnitTest
+        run: ./src/UniTask/bin/UnitTest/StandaloneLinux64_Mono2x/test
+
+      # Execute scripts: Export Package
+      - name: Export unitypackage
+        run: /opt/Unity/Editor/Unity -quit -batchmode -nographics -silent-crashes -logFile -projectPath . -executeMethod PackageExporter.Export
+        working-directory: src/UniTask
+
+      # Store artifacts.      
+      - uses: actions/upload-artifact@v2
+        with:
+          name: UniTask.unitypackage.zip
+          path: ./src/UniTask/*.unitypackage

.github/workflows/build-release.yml

@@ -0,0 +1,108 @@
+name: Build-Release
+
+on:
+  push:
+    tags:
+      - "[0-9]+.[0-9]+.[0-9]+*"
+
+jobs:
+  build-dotnet:
+    runs-on: ubuntu-latest
+    env:
+      DOTNET_CLI_TELEMETRY_OPTOUT: 1
+      DOTNET_SKIP_FIRST_TIME_EXPERIENCE: 1
+      NUGET_XMLDOC_MODE: skip
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions/setup-dotnet@v1
+        with:
+          dotnet-version: 3.1.101
+      # set release tag(*.*.*) to env.GIT_TAG
+      - run: echo ::set-env name=GIT_TAG::${GITHUB_REF#refs/tags/}
+
+      # build CommandTools first (use dotnet run command in ZLogger.csproj)
+      - run: dotnet build -c Release ./tools/CommandTools/CommandTools.csproj
+      - run: dotnet build -c Release -p:Version=${{ env.GIT_TAG }}
+      - run: dotnet test -c Release --no-build
+      - run: dotnet pack ./src/ZLogger/ZLogger.csproj -c Release --no-build -p:Version=${{ env.GIT_TAG }}
+
+      # Store artifacts.
+      - uses: actions/upload-artifact@v1
+        with:
+          name: nuget
+          path: ./src/ZLogger/bin/Release/ZLogger.${{ env.GIT_TAG }}.nupkg
+      
+  build-unity:
+    strategy:
+      matrix:
+        unity: ['2019.3.9f1']
+        include:
+          - unity: 2019.3.9f1
+            license: UNITY_2019_3
+    runs-on: ubuntu-latest
+    container:
+      # with linux-il2cpp. image from https://hub.docker.com/r/gableroux/unity3d/tags
+      image: gableroux/unity3d:${{ matrix.unity }}-linux-il2cpp
+    steps:
+      - run: apt update && apt install git -y
+      - uses: actions/checkout@v2
+      - run: echo -n "$UNITY_LICENSE" >> .Unity.ulf
+        env:
+          UNITY_LICENSE: ${{ secrets[matrix.license] }}
+      - run: /opt/Unity/Editor/Unity -quit -batchmode -nographics -silent-crashes -logFile -manualLicenseFile .Unity.ulf || exit 0
+
+      # set release tag(*.*.*) to env.GIT_TAG
+      - run: echo ::set-env name=GIT_TAG::${GITHUB_REF#refs/tags/}
+
+      # Execute scripts: Export Package
+      - name: Export unitypackage
+        run: /opt/Unity/Editor/Unity -quit -batchmode -nographics -silent-crashes -logFile -projectPath . -executeMethod PackageExporter.Export
+        working-directory: src/ZLogger.Unity
+        env:
+          UNITY_PACKAGE_VERSION: ${{ env.GIT_TAG }}
+
+      # Store artifacts.
+      - uses: actions/upload-artifact@v1
+        with:
+          name: ZLogger.Unity.${{ env.GIT_TAG }}.unitypackage
+          path: ./src/ZLogger.Unity/ZLogger.Unity.${{ env.GIT_TAG }}.unitypackage
+
+  create-release:
+    needs: [build-dotnet, build-unity]
+    runs-on: ubuntu-latest
+    env:
+      DOTNET_CLI_TELEMETRY_OPTOUT: 1
+      DOTNET_SKIP_FIRST_TIME_EXPERIENCE: 1
+      NUGET_XMLDOC_MODE: skip
+    steps:
+    # setup dotnet for nuget push
+    - uses: actions/setup-dotnet@v1
+      with:
+        dotnet-version: 3.1.101
+    # set release tag(*.*.*) to env.GIT_TAG
+    - run: echo ::set-env name=GIT_TAG::${GITHUB_REF#refs/tags/}
+
+    # Create Releases
+    - uses: actions/create-release@v1
+      id: create_release
+      env:
+        GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+      with:
+        tag_name: ${{ github.ref }}
+        release_name: Ver.${{ github.ref }}
+
+    # Download (All) Artifacts to current directory
+    - uses: actions/download-artifact@v2-preview
+
+    # Upload to NuGet
+    - run: dotnet nuget push "./nuget/*.nupkg" -s https://www.nuget.org/api/v2/package -k ${{ secrets.NUGET_KEY }}
+
+    # Upload to Releases(unitypackage)
+    - uses: actions/upload-release-asset@v1
+      env:
+        GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+      with:
+        upload_url: ${{ steps.create_release.outputs.upload_url }}
+        asset_path: ./ZLogger.Unity.${{ env.GIT_TAG }}.unitypackage/ZLogger.Unity.${{ env.GIT_TAG }}.unitypackage
+        asset_name: ZLogger.Unity.${{ env.GIT_TAG }}.unitypackage
+        asset_content_type: application/octet-stream

.github/workflows/toc.yml

@@ -0,0 +1,15 @@
+name: TOC Generator
+
+on:
+  push:
+    paths:
+      - 'README.md'
+
+jobs:
+  generateTOC:
+    name: TOC Generator
+    runs-on: ubuntu-latest
+    steps:
+      - uses: technote-space/toc-generator@v2.4.0
+        with:
+          TOC_TITLE: "## Table of Contents"

README.md

@@ -1,11 +1,29 @@
-# UniTask
+<!-- 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
 
-[![CircleCI](https://circleci.com/gh/Cysharp/UniTask.svg?style=svg)](https://circleci.com/gh/Cysharp/UniTask)
+- [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.
 
-> UniTask was included in UniRx before v7 but now completely separated, it no dependent each other.
-
 Getting started
 ---
 Install package(`UniRx.Async.unitypackage`) is available in [UniTask/releases](https://github.com/Cysharp/UniTask/releases) page.

src/UniTask.NetCoreSandbox/Program.cs

@@ -86,30 +86,16 @@ namespace NetCoreSandbox
             await Task.Delay(10, cancellationToken);
         }
 
-        static async Task Main(string[] args)
+
+
+        static void Main(string[] args)
         {
-            await foreach (var item in UniTaskAsyncEnumerable.Range(1, 10)
-                .SelectAwait(x => UniTask.Run(() => x))
-                .TakeLast(6)
 
-                
-
-                )
-            {
-
-                
-
-                Console.WriteLine(item);
-            }
-
-// AsyncEnumerable.Range(1,10).FirstAsync(
-            // AsyncEnumerable.Range(1, 10).GroupBy(x=>x).Select(x=>x.first
+            var channel = Channel.CreateSingleConsumerUnbounded<int>();
 
 
-            // AsyncEnumerable.Range(1,10).WithCancellation(CancellationToken.None).WithCancellation
 
 
-            //Enumerable.Range(1,10).ToHashSet(
 
 
         }

src/UniTask.NetCoreTests/AsyncReactivePropertyTest.cs

@@ -0,0 +1,57 @@
+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 AsyncReactivePropertyTest
+    {
+        [Fact]
+        public async Task Iteration()
+        {
+            var rp = new AsyncReactiveProperty<int>(99);
+
+            var f = await rp.FirstAsync();
+            f.Should().Be(99);
+
+            var array = rp.Take(5).ToArrayAsync();
+
+            rp.Value = 100;
+            rp.Value = 100;
+            rp.Value = 100;
+            rp.Value = 131;
+
+            var ar = await array;
+
+            ar.Should().BeEquivalentTo(new[] { 99, 100, 100, 100, 131 });
+        }
+
+        [Fact]
+        public async Task WithoutCurrent()
+        {
+            var rp = new AsyncReactiveProperty<int>(99);
+
+            var array = rp.WithoutCurrent().Take(5).ToArrayAsync();
+
+            rp.Value = 100;
+            rp.Value = 100;
+            rp.Value = 100;
+            rp.Value = 131;
+            rp.Value = 191;
+
+            var ar = await array;
+
+            ar.Should().BeEquivalentTo(new[] { 100, 100, 100, 131, 191 });
+        }
+    }
+
+
+}

src/UniTask.NetCoreTests/ChannelTest.cs

@@ -0,0 +1,370 @@
+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 ChannelTest
+    {
+        (System.Threading.Channels.Channel<int>, Cysharp.Threading.Tasks.Channel<int>) CreateChannel()
+        {
+            var reference = System.Threading.Channels.Channel.CreateUnbounded<int>(new UnboundedChannelOptions
+            {
+                AllowSynchronousContinuations = true,
+                SingleReader = true,
+                SingleWriter = false
+            });
+
+            var channel = Cysharp.Threading.Tasks.Channel.CreateSingleConsumerUnbounded<int>();
+
+            return (reference, channel);
+        }
+
+        [Fact]
+        public async Task SingleWriteSingleRead()
+        {
+            var (reference, channel) = CreateChannel();
+
+            foreach (var item in new[] { 10, 20, 30 })
+            {
+                var t1 = reference.Reader.WaitToReadAsync();
+                var t2 = channel.Reader.WaitToReadAsync();
+
+                t1.IsCompleted.Should().BeFalse();
+                t2.Status.IsCompleted().Should().BeFalse();
+
+                reference.Writer.TryWrite(item);
+                channel.Writer.TryWrite(item);
+
+                (await t1).Should().BeTrue();
+                (await t2).Should().BeTrue();
+
+                reference.Reader.TryRead(out var refitem).Should().BeTrue();
+                channel.Reader.TryRead(out var chanitem).Should().BeTrue();
+                refitem.Should().Be(item);
+                chanitem.Should().Be(item);
+            }
+        }
+
+        [Fact]
+        public async Task MultiWrite()
+        {
+            var (reference, channel) = CreateChannel();
+
+            foreach (var item in new[] { 10, 20, 30 })
+            {
+                var t1 = reference.Reader.WaitToReadAsync();
+                var t2 = channel.Reader.WaitToReadAsync();
+
+                t1.IsCompleted.Should().BeFalse();
+                t2.Status.IsCompleted().Should().BeFalse();
+
+                foreach (var i in Enumerable.Range(1, 3))
+                {
+                    reference.Writer.TryWrite(item * i);
+                    channel.Writer.TryWrite(item * i);
+                }
+
+                (await t1).Should().BeTrue();
+                (await t2).Should().BeTrue();
+
+                foreach (var i in Enumerable.Range(1, 3))
+                {
+                    (await reference.Reader.WaitToReadAsync()).Should().BeTrue();
+                    (await channel.Reader.WaitToReadAsync()).Should().BeTrue();
+
+                    reference.Reader.TryRead(out var refitem).Should().BeTrue();
+                    channel.Reader.TryRead(out var chanitem).Should().BeTrue();
+                    refitem.Should().Be(item * i);
+                    chanitem.Should().Be(item * i);
+                }
+            }
+        }
+
+        [Fact]
+        public async Task CompleteOnEmpty()
+        {
+            var (reference, channel) = CreateChannel();
+
+            foreach (var item in new[] { 10, 20, 30 })
+            {
+                reference.Writer.TryWrite(item);
+                channel.Writer.TryWrite(item);
+                reference.Reader.TryRead(out var refitem);
+                channel.Reader.TryRead(out var chanitem);
+            }
+
+            // Empty.
+
+            var completion1 = reference.Reader.Completion;
+            var wait1 = reference.Reader.WaitToReadAsync();
+
+            var completion2 = channel.Reader.Completion;
+            var wait2 = channel.Reader.WaitToReadAsync();
+
+            reference.Writer.TryComplete();
+            channel.Writer.TryComplete();
+
+            completion1.Status.Should().Be(TaskStatus.RanToCompletion);
+            completion2.Status.Should().Be(UniTaskStatus.Succeeded);
+
+            (await wait1).Should().BeFalse();
+            (await wait2).Should().BeFalse();
+        }
+
+        [Fact]
+        public async Task CompleteErrorOnEmpty()
+        {
+            var (reference, channel) = CreateChannel();
+
+            foreach (var item in new[] { 10, 20, 30 })
+            {
+                reference.Writer.TryWrite(item);
+                channel.Writer.TryWrite(item);
+                reference.Reader.TryRead(out var refitem);
+                channel.Reader.TryRead(out var chanitem);
+            }
+
+            // Empty.
+
+            var completion1 = reference.Reader.Completion;
+            var wait1 = reference.Reader.WaitToReadAsync();
+
+            var completion2 = channel.Reader.Completion;
+            var wait2 = channel.Reader.WaitToReadAsync();
+
+            var ex = new Exception();
+            reference.Writer.TryComplete(ex);
+            channel.Writer.TryComplete(ex);
+
+            completion1.Status.Should().Be(TaskStatus.Faulted);
+            completion2.Status.Should().Be(UniTaskStatus.Faulted);
+
+            (await Assert.ThrowsAsync<Exception>(async () => await wait1)).Should().Be(ex);
+            (await Assert.ThrowsAsync<Exception>(async () => await wait2)).Should().Be(ex);
+        }
+
+        [Fact]
+        public async Task CompleteWithRest()
+        {
+            var (reference, channel) = CreateChannel();
+
+            foreach (var item in new[] { 10, 20, 30 })
+            {
+                reference.Writer.TryWrite(item);
+                channel.Writer.TryWrite(item);
+            }
+
+            // Three Item2.
+
+            var completion1 = reference.Reader.Completion;
+            var wait1 = reference.Reader.WaitToReadAsync();
+
+            var completion2 = channel.Reader.Completion;
+            var wait2 = channel.Reader.WaitToReadAsync();
+
+            reference.Writer.TryComplete();
+            channel.Writer.TryComplete();
+
+            // completion1.Status.Should().Be(TaskStatus.WaitingForActivation);
+            completion2.Status.Should().Be(UniTaskStatus.Pending);
+
+            (await wait1).Should().BeTrue();
+            (await wait2).Should().BeTrue();
+
+            foreach (var item in new[] { 10, 20, 30 })
+            {
+                reference.Reader.TryRead(out var i1).Should().BeTrue();
+                channel.Reader.TryRead(out var i2).Should().BeTrue();
+                i1.Should().Be(item);
+                i2.Should().Be(item);
+            }
+
+            (await reference.Reader.WaitToReadAsync()).Should().BeFalse();
+            (await channel.Reader.WaitToReadAsync()).Should().BeFalse();
+
+            completion1.Status.Should().Be(TaskStatus.RanToCompletion);
+            completion2.Status.Should().Be(UniTaskStatus.Succeeded);
+        }
+
+
+        [Fact]
+        public async Task CompleteErrorWithRest()
+        {
+            var (reference, channel) = CreateChannel();
+
+            foreach (var item in new[] { 10, 20, 30 })
+            {
+                reference.Writer.TryWrite(item);
+                channel.Writer.TryWrite(item);
+            }
+
+            // Three Item2.
+
+            var completion1 = reference.Reader.Completion;
+            var wait1 = reference.Reader.WaitToReadAsync();
+
+            var completion2 = channel.Reader.Completion;
+            var wait2 = channel.Reader.WaitToReadAsync();
+
+            var ex = new Exception();
+            reference.Writer.TryComplete(ex);
+            channel.Writer.TryComplete(ex);
+
+            // completion1.Status.Should().Be(TaskStatus.WaitingForActivation);
+            completion2.Status.Should().Be(UniTaskStatus.Pending);
+
+            (await wait1).Should().BeTrue();
+            (await wait2).Should().BeTrue();
+
+            foreach (var item in new[] { 10, 20, 30 })
+            {
+                reference.Reader.TryRead(out var i1).Should().BeTrue();
+                channel.Reader.TryRead(out var i2).Should().BeTrue();
+                i1.Should().Be(item);
+                i2.Should().Be(item);
+            }
+
+            wait1 = reference.Reader.WaitToReadAsync();
+            wait2 = channel.Reader.WaitToReadAsync();
+
+            (await Assert.ThrowsAsync<Exception>(async () => await wait1)).Should().Be(ex);
+            (await Assert.ThrowsAsync<Exception>(async () => await wait2)).Should().Be(ex);
+
+            completion1.Status.Should().Be(TaskStatus.Faulted);
+            completion2.Status.Should().Be(UniTaskStatus.Faulted);
+        }
+
+        [Fact]
+        public async Task Cancellation()
+        {
+            var (reference, channel) = CreateChannel();
+
+            var cts = new CancellationTokenSource();
+
+            var wait1 = reference.Reader.WaitToReadAsync(cts.Token);
+            var wait2 = channel.Reader.WaitToReadAsync(cts.Token);
+
+            cts.Cancel();
+
+            (await Assert.ThrowsAsync<OperationCanceledException>(async () => await wait1)).CancellationToken.Should().Be(cts.Token);
+            (await Assert.ThrowsAsync<OperationCanceledException>(async () => await wait2)).CancellationToken.Should().Be(cts.Token);
+        }
+
+        [Fact]
+        public async Task AsyncEnumerator()
+        {
+            var (reference, channel) = CreateChannel();
+
+            var ta1 = reference.Reader.ReadAllAsync().ToArrayAsync();
+            var ta2 = channel.Reader.ReadAllAsync().ToArrayAsync();
+
+            foreach (var item in new[] { 10, 20, 30 })
+            {
+                reference.Writer.TryWrite(item);
+                channel.Writer.TryWrite(item);
+            }
+
+            reference.Writer.TryComplete();
+            channel.Writer.TryComplete();
+
+            (await ta1).Should().BeEquivalentTo(new[] { 10, 20, 30 });
+            (await ta2).Should().BeEquivalentTo(new[] { 10, 20, 30 });
+        }
+
+        [Fact]
+        public async Task AsyncEnumeratorCancellation()
+        {
+            // Token1, Token2 and Cancel1
+            {
+                var cts1 = new CancellationTokenSource();
+                var cts2 = new CancellationTokenSource();
+
+                var (reference, channel) = CreateChannel();
+
+                var ta1 = reference.Reader.ReadAllAsync(cts1.Token).ToArrayAsync(cts2.Token);
+                var ta2 = channel.Reader.ReadAllAsync(cts1.Token).ToArrayAsync(cts2.Token);
+
+                foreach (var item in new[] { 10, 20, 30 })
+                {
+                    reference.Writer.TryWrite(item);
+                    channel.Writer.TryWrite(item);
+                }
+
+                cts1.Cancel();
+
+                await Assert.ThrowsAsync<OperationCanceledException>(async () => await ta1);
+                (await Assert.ThrowsAsync<OperationCanceledException>(async () => await ta2)).CancellationToken.Should().Be(cts1.Token);
+            }
+            // Token1, Token2 and Cancel2
+            {
+                var cts1 = new CancellationTokenSource();
+                var cts2 = new CancellationTokenSource();
+
+                var (reference, channel) = CreateChannel();
+
+                var ta1 = reference.Reader.ReadAllAsync(cts1.Token).ToArrayAsync(cts2.Token);
+                var ta2 = channel.Reader.ReadAllAsync(cts1.Token).ToArrayAsync(cts2.Token);
+
+                foreach (var item in new[] { 10, 20, 30 })
+                {
+                    reference.Writer.TryWrite(item);
+                    channel.Writer.TryWrite(item);
+                }
+
+                cts2.Cancel();
+
+                await Assert.ThrowsAsync<OperationCanceledException>(async () => await ta1);
+                (await Assert.ThrowsAsync<OperationCanceledException>(async () => await ta2)).CancellationToken.Should().Be(cts2.Token);
+            }
+            // Token1 and Cancel1
+            {
+                var cts1 = new CancellationTokenSource();
+
+                var (reference, channel) = CreateChannel();
+
+                var ta1 = reference.Reader.ReadAllAsync(cts1.Token).ToArrayAsync();
+                var ta2 = channel.Reader.ReadAllAsync(cts1.Token).ToArrayAsync();
+
+                foreach (var item in new[] { 10, 20, 30 })
+                {
+                    reference.Writer.TryWrite(item);
+                    channel.Writer.TryWrite(item);
+                }
+
+                cts1.Cancel();
+
+                await Assert.ThrowsAsync<OperationCanceledException>(async () => await ta1);
+                (await Assert.ThrowsAsync<OperationCanceledException>(async () => await ta2)).CancellationToken.Should().Be(cts1.Token);
+            }
+            // Token2 and Cancel2
+            {
+                var cts2 = new CancellationTokenSource();
+
+                var (reference, channel) = CreateChannel();
+
+                var ta1 = reference.Reader.ReadAllAsync().ToArrayAsync(cts2.Token);
+                var ta2 = channel.Reader.ReadAllAsync().ToArrayAsync(cts2.Token);
+
+                foreach (var item in new[] { 10, 20, 30 })
+                {
+                    reference.Writer.TryWrite(item);
+                    channel.Writer.TryWrite(item);
+                }
+
+                cts2.Cancel();
+
+                await Assert.ThrowsAsync<OperationCanceledException>(async () => await ta1);
+                (await Assert.ThrowsAsync<OperationCanceledException>(async () => await ta2)).CancellationToken.Should().Be(cts2.Token);
+            }
+        }
+    }
+}

src/UniTask.NetCoreTests/Linq/QueueTest.cs

@@ -0,0 +1,29 @@
+using Cysharp.Threading.Tasks;
+using Cysharp.Threading.Tasks.Linq;
+using FluentAssertions;
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace NetCoreTests.Linq
+{
+    public class QueueTest
+    {
+        [Fact]
+        public async Task Q()
+        {
+            var rp = new AsyncReactiveProperty<int>(100);
+
+            var l = new List<int>();
+            await rp.Take(10).Queue().ForEachAsync(x =>
+            {
+                rp.Value += 10;
+                l.Add(x);
+            });
+
+            l.Should().BeEquivalentTo(100, 110, 120, 130, 140, 150, 160, 170, 180, 190);
+        }
+    }
+}

src/UniTask.NetCoreTests/Linq/TakeInfinityTest.cs

@@ -0,0 +1,44 @@
+using Cysharp.Threading.Tasks;
+using Cysharp.Threading.Tasks.Linq;
+using FluentAssertions;
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace NetCoreTests.Linq
+{
+    public class TakeInfinityTest
+    {
+        [Fact]
+        public async Task Take()
+        {
+            var rp = new AsyncReactiveProperty<int>(1);
+
+            var xs = rp.Take(5).ToArrayAsync();
+
+            rp.Value = 2;
+            rp.Value = 3;
+            rp.Value = 4;
+            rp.Value = 5;
+
+            (await xs).Should().BeEquivalentTo(1, 2, 3, 4, 5);
+        }
+
+        [Fact]
+        public async Task TakeWhile()
+        {
+            var rp = new AsyncReactiveProperty<int>(1);
+
+            var xs = rp.TakeWhile(x => x != 5).ToArrayAsync();
+
+            rp.Value = 2;
+            rp.Value = 3;
+            rp.Value = 4;
+            rp.Value = 5;
+
+            (await xs).Should().BeEquivalentTo(1, 2, 3, 4);
+        }
+    }
+}

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

@@ -15,7 +15,7 @@ namespace Cysharp.Threading.Tasks
     }
 
     [Serializable]
-    public class AsyncReactiveProperty<T> : IUniTaskAsyncEnumerable<T>, IDisposable
+    public class AsyncReactiveProperty<T> : IAsyncReactiveProperty<T>, IDisposable
     {
         TriggerEvent<T> triggerEvent;
 
@@ -43,9 +43,14 @@ namespace Cysharp.Threading.Tasks
             this.triggerEvent = default;
         }
 
+        public IUniTaskAsyncEnumerable<T> WithoutCurrent()
+        {
+            return new WithoutCurrentEnumerable(this);
+        }
+
         public IUniTaskAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken)
         {
-            return new Enumerator(this, cancellationToken);
+            return new Enumerator(this, cancellationToken, true);
         }
 
         public void Dispose()
@@ -53,6 +58,21 @@ namespace Cysharp.Threading.Tasks
             triggerEvent.SetCanceled(CancellationToken.None);
         }
 
+        class WithoutCurrentEnumerable : IUniTaskAsyncEnumerable<T>
+        {
+            readonly AsyncReactiveProperty<T> parent;
+
+            public WithoutCurrentEnumerable(AsyncReactiveProperty<T> parent)
+            {
+                this.parent = parent;
+            }
+
+            public IUniTaskAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default)
+            {
+                return new Enumerator(parent, cancellationToken, false);
+            }
+        }
+
         sealed class Enumerator : MoveNextSource, IUniTaskAsyncEnumerator<T>, IResolveCancelPromise<T>
         {
             static Action<object> cancellationCallback = CancellationCallback;
@@ -62,11 +82,13 @@ namespace Cysharp.Threading.Tasks
             readonly CancellationTokenRegistration cancellationTokenRegistration;
             T value;
             bool isDisposed;
+            bool firstCall;
 
-            public Enumerator(AsyncReactiveProperty<T> parent, CancellationToken cancellationToken)
+            public Enumerator(AsyncReactiveProperty<T> parent, CancellationToken cancellationToken, bool publishCurrentValue)
             {
                 this.parent = parent;
                 this.cancellationToken = cancellationToken;
+                this.firstCall = publishCurrentValue;
 
                 parent.triggerEvent.Add(this);
                 TaskTracker.TrackActiveTask(this, 3);
@@ -81,6 +103,14 @@ namespace Cysharp.Threading.Tasks
 
             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);
             }

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

@@ -0,0 +1,403 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+
+namespace Cysharp.Threading.Tasks
+{
+    public static class Channel
+    {
+        public static Channel<T> CreateSingleConsumerUnbounded<T>()
+        {
+            return new SingleConsumerUnboundedChannel<T>();
+        }
+    }
+
+    public abstract class Channel<TWrite, TRead>
+    {
+        public ChannelReader<TRead> Reader { get; protected set; }
+        public ChannelWriter<TWrite> Writer { get; protected set; }
+
+        public static implicit operator ChannelReader<TRead>(Channel<TWrite, TRead> channel) => channel.Reader;
+        public static implicit operator ChannelWriter<TWrite>(Channel<TWrite, TRead> channel) => channel.Writer;
+    }
+
+    public abstract class Channel<T> : Channel<T, T>
+    {
+    }
+
+    public abstract class ChannelReader<T>
+    {
+        public abstract bool TryRead(out T item);
+        public abstract UniTask<bool> WaitToReadAsync(CancellationToken cancellationToken = default(CancellationToken));
+
+        public abstract UniTask Completion { get; }
+
+        public virtual UniTask<T> ReadAsync(CancellationToken cancellationToken = default(CancellationToken))
+        {
+            if (this.TryRead(out var item))
+            {
+                return UniTask.FromResult(item);
+            }
+
+            return ReadAsyncCore(cancellationToken);
+        }
+
+        async UniTask<T> ReadAsyncCore(CancellationToken cancellationToken = default(CancellationToken))
+        {
+            if (await WaitToReadAsync(cancellationToken))
+            {
+                if (TryRead(out var item))
+                {
+                    return item;
+                }
+            }
+
+            throw new ChannelClosedException();
+        }
+
+        public abstract IUniTaskAsyncEnumerable<T> ReadAllAsync(CancellationToken cancellationToken = default(CancellationToken));
+    }
+
+    public abstract class ChannelWriter<T>
+    {
+        public abstract bool TryWrite(T item);
+        public abstract bool TryComplete(Exception error = null);
+
+        public void Complete(Exception error = null)
+        {
+            if (!TryComplete(error))
+            {
+                throw new ChannelClosedException();
+            }
+        }
+    }
+
+    public partial class ChannelClosedException : InvalidOperationException
+    {
+        public ChannelClosedException() :
+            base("Channel is already closed.")
+        { }
+
+        public ChannelClosedException(string message) : base(message) { }
+
+        public ChannelClosedException(Exception innerException) :
+            base("Channel is already closed", innerException)
+        { }
+
+        public ChannelClosedException(string message, Exception innerException) : base(message, innerException) { }
+    }
+
+    internal class SingleConsumerUnboundedChannel<T> : Channel<T>
+    {
+        readonly Queue<T> items;
+        readonly SingleConsumerUnboundedChannelReader readerSource;
+        readonly UniTaskCompletionSource completedTask;
+
+        Exception completionError;
+        bool closed;
+
+        public SingleConsumerUnboundedChannel()
+        {
+            items = new Queue<T>();
+            completedTask = new UniTaskCompletionSource();
+            Writer = new SingleConsumerUnboundedChannelWriter(this);
+            readerSource = new SingleConsumerUnboundedChannelReader(this);
+            Reader = readerSource;
+        }
+
+        sealed class SingleConsumerUnboundedChannelWriter : ChannelWriter<T>
+        {
+            readonly SingleConsumerUnboundedChannel<T> parent;
+
+            public SingleConsumerUnboundedChannelWriter(SingleConsumerUnboundedChannel<T> parent)
+            {
+                this.parent = parent;
+            }
+
+            public override bool TryWrite(T item)
+            {
+                bool waiting;
+                lock (parent.items)
+                {
+                    if (parent.closed) return false;
+
+                    parent.items.Enqueue(item);
+                    waiting = parent.readerSource.isWaiting;
+                }
+
+                if (waiting)
+                {
+                    parent.readerSource.SingalContinuation();
+                }
+
+                return true;
+            }
+
+            public override bool TryComplete(Exception error = null)
+            {
+                bool waiting;
+                lock (parent.items)
+                {
+                    if (parent.closed) return false;
+                    parent.closed = true;
+                    waiting = parent.readerSource.isWaiting;
+
+                    if (parent.items.Count == 0)
+                    {
+                        if (error == null)
+                        {
+                            parent.completedTask.TrySetResult();
+                        }
+                        else
+                        {
+                            parent.completedTask.TrySetException(error);
+                        }
+
+                        if (waiting)
+                        {
+                            parent.readerSource.SingalCompleted(error);
+                        }
+                    }
+
+                    parent.completionError = error;
+                }
+
+                return true;
+            }
+        }
+
+        sealed class SingleConsumerUnboundedChannelReader : ChannelReader<T>, IUniTaskSource<bool>
+        {
+            readonly Action<object> CancellationCallbackDelegate = CancellationCallback;
+            readonly SingleConsumerUnboundedChannel<T> parent;
+
+            CancellationToken cancellationToken;
+            CancellationTokenRegistration cancellationTokenRegistration;
+            UniTaskCompletionSourceCore<bool> core;
+            internal bool isWaiting;
+
+            public SingleConsumerUnboundedChannelReader(SingleConsumerUnboundedChannel<T> parent)
+            {
+                this.parent = parent;
+            }
+
+            public override UniTask Completion => parent.completedTask.Task;
+
+            public override bool TryRead(out T item)
+            {
+                lock (parent.items)
+                {
+                    if (parent.items.Count != 0)
+                    {
+                        item = parent.items.Dequeue();
+
+                        // complete when all value was consumed.
+                        if (parent.closed && parent.items.Count == 0)
+                        {
+                            if (parent.completionError != null)
+                            {
+                                parent.completedTask.TrySetException(parent.completionError);
+                            }
+                            else
+                            {
+                                parent.completedTask.TrySetResult();
+                            }
+                        }
+                    }
+                    else
+                    {
+                        item = default;
+                        return false;
+                    }
+                }
+
+                return true;
+            }
+
+            public override UniTask<bool> WaitToReadAsync(CancellationToken cancellationToken)
+            {
+                if (cancellationToken.IsCancellationRequested)
+                {
+                    return UniTask.FromCanceled<bool>(cancellationToken);
+                }
+
+                lock (parent.items)
+                {
+                    if (parent.items.Count != 0)
+                    {
+                        return CompletedTasks.True;
+                    }
+
+                    if (parent.closed)
+                    {
+                        if (parent.completionError == null)
+                        {
+                            return CompletedTasks.False;
+                        }
+                        else
+                        {
+                            return UniTask.FromException<bool>(parent.completionError);
+                        }
+                    }
+
+                    cancellationTokenRegistration.Dispose();
+
+                    core.Reset();
+                    isWaiting = true;
+
+                    this.cancellationToken = cancellationToken;
+                    if (this.cancellationToken.CanBeCanceled)
+                    {
+                        cancellationTokenRegistration = this.cancellationToken.RegisterWithoutCaptureExecutionContext(CancellationCallbackDelegate, this);
+                    }
+
+                    return new UniTask<bool>(this, core.Version);
+                }
+            }
+
+            public void SingalContinuation()
+            {
+                core.TrySetResult(true);
+            }
+
+            public void SingalCancellation(CancellationToken cancellationToken)
+            {
+                core.TrySetCanceled(cancellationToken);
+            }
+
+            public void SingalCompleted(Exception error)
+            {
+                if (error != null)
+                {
+                    core.TrySetException(error);
+                }
+                else
+                {
+                    core.TrySetResult(false);
+                }
+            }
+
+            public override IUniTaskAsyncEnumerable<T> ReadAllAsync(CancellationToken cancellationToken = default)
+            {
+                return new ReadAllAsyncEnumerable(this, cancellationToken);
+            }
+
+            bool IUniTaskSource<bool>.GetResult(short token)
+            {
+                return core.GetResult(token);
+            }
+
+            void IUniTaskSource.GetResult(short token)
+            {
+                core.GetResult(token);
+            }
+
+            UniTaskStatus IUniTaskSource.GetStatus(short token)
+            {
+                return core.GetStatus(token);
+            }
+
+            void IUniTaskSource.OnCompleted(Action<object> continuation, object state, short token)
+            {
+                core.OnCompleted(continuation, state, token);
+            }
+
+            UniTaskStatus IUniTaskSource.UnsafeGetStatus()
+            {
+                return core.UnsafeGetStatus();
+            }
+
+            static void CancellationCallback(object state)
+            {
+                var self = (SingleConsumerUnboundedChannelReader)state;
+                self.SingalCancellation(self.cancellationToken);
+            }
+
+            sealed class ReadAllAsyncEnumerable : IUniTaskAsyncEnumerable<T>, IUniTaskAsyncEnumerator<T>
+            {
+                readonly Action<object> CancellationCallback1Delegate = CancellationCallback1;
+                readonly Action<object> CancellationCallback2Delegate = CancellationCallback2;
+
+                readonly SingleConsumerUnboundedChannelReader parent;
+                CancellationToken cancellationToken1;
+                CancellationToken cancellationToken2;
+                CancellationTokenRegistration CancellationTokenRegistration1;
+                CancellationTokenRegistration CancellationTokenRegistration2;
+
+                T current;
+                bool cacheValue;
+                bool running;
+
+                public ReadAllAsyncEnumerable(SingleConsumerUnboundedChannelReader parent, CancellationToken cancellationToken)
+                {
+                    this.parent = parent;
+                    this.cancellationToken1 = cancellationToken;
+                }
+
+                public IUniTaskAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default)
+                {
+                    if (running)
+                    {
+                        throw new InvalidOperationException("Enumerator is already running, does not allow call GetAsyncEnumerator twice.");
+                    }
+
+                    if (this.cancellationToken1 != cancellationToken)
+                    {
+                        this.cancellationToken2 = cancellationToken;
+                    }
+
+                    if (this.cancellationToken1.CanBeCanceled)
+                    {
+                        this.cancellationToken1.RegisterWithoutCaptureExecutionContext(CancellationCallback1Delegate, this);
+                    }
+
+                    if (this.cancellationToken2.CanBeCanceled)
+                    {
+                        this.cancellationToken2.RegisterWithoutCaptureExecutionContext(CancellationCallback2Delegate, this);
+                    }
+
+                    running = true;
+                    return this;
+                }
+
+                public T Current
+                {
+                    get
+                    {
+                        if (cacheValue)
+                        {
+                            return current;
+                        }
+                        parent.TryRead(out current);
+                        return current;
+                    }
+                }
+
+                public UniTask<bool> MoveNextAsync()
+                {
+                    cacheValue = false;
+                    return parent.WaitToReadAsync(CancellationToken.None); // ok to use None, registered in ctor.
+                }
+
+                public UniTask DisposeAsync()
+                {
+                    CancellationTokenRegistration1.Dispose();
+                    CancellationTokenRegistration2.Dispose();
+                    return default;
+                }
+
+                static void CancellationCallback1(object state)
+                {
+                    var self = (ReadAllAsyncEnumerable)state;
+                    self.parent.SingalCancellation(self.cancellationToken1);
+                }
+
+                static void CancellationCallback2(object state)
+                {
+                    var self = (ReadAllAsyncEnumerable)state;
+                    self.parent.SingalCancellation(self.cancellationToken2);
+                }
+            }
+        }
+    }
+}

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

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

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

@@ -0,0 +1,95 @@
+using System;
+using System.Threading;
+
+namespace Cysharp.Threading.Tasks.Linq
+{
+    public static partial class UniTaskAsyncEnumerable
+    {
+        public static IUniTaskAsyncEnumerable<TSource> Queue<TSource>(this IUniTaskAsyncEnumerable<TSource> source)
+        {
+            return new QueueOperator<TSource>(source);
+        }
+    }
+
+    internal sealed class QueueOperator<TSource> : IUniTaskAsyncEnumerable<TSource>
+    {
+        readonly IUniTaskAsyncEnumerable<TSource> source;
+
+        public QueueOperator(IUniTaskAsyncEnumerable<TSource> source)
+        {
+            this.source = source;
+        }
+
+        public IUniTaskAsyncEnumerator<TSource> GetAsyncEnumerator(CancellationToken cancellationToken = default)
+        {
+            return new _Queue(source, cancellationToken);
+        }
+
+        sealed class _Queue : IUniTaskAsyncEnumerator<TSource>
+        {
+            readonly IUniTaskAsyncEnumerable<TSource> source;
+            CancellationToken cancellationToken;
+
+            Channel<TSource> channel;
+            IUniTaskAsyncEnumerator<TSource> channelEnumerator;
+            IUniTaskAsyncEnumerator<TSource> sourceEnumerator;
+
+            public _Queue(IUniTaskAsyncEnumerable<TSource> source, CancellationToken cancellationToken)
+            {
+                this.source = source;
+                this.cancellationToken = cancellationToken;
+            }
+
+            public TSource Current => channelEnumerator.Current;
+
+            public UniTask<bool> MoveNextAsync()
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+
+                if (sourceEnumerator == null)
+                {
+                    sourceEnumerator = source.GetAsyncEnumerator(cancellationToken);
+                    channel = Channel.CreateSingleConsumerUnbounded<TSource>();
+
+                    channelEnumerator = channel.Reader.ReadAllAsync().GetAsyncEnumerator(cancellationToken);
+
+                    ConsumeAll(sourceEnumerator, channel).Forget();
+                }
+
+                return channelEnumerator.MoveNextAsync();
+            }
+
+            static async UniTaskVoid ConsumeAll(IUniTaskAsyncEnumerator<TSource> enumerator, ChannelWriter<TSource> writer)
+            {
+                try
+                {
+                    while (await enumerator.MoveNextAsync())
+                    {
+                        writer.TryWrite(enumerator.Current);
+                    }
+                    writer.TryComplete();
+                }
+                catch (Exception ex)
+                {
+                    writer.TryComplete(ex);
+                }
+                finally
+                {
+                    await enumerator.DisposeAsync();
+                }
+            }
+
+            public async UniTask DisposeAsync()
+            {
+                if (sourceEnumerator != null)
+                {
+                    await sourceEnumerator.DisposeAsync();
+                }
+                if (channelEnumerator != null)
+                {
+                    await channelEnumerator.DisposeAsync();
+                }
+            }
+        }
+    }
+}

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

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

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

@@ -30,40 +30,93 @@ namespace Cysharp.Threading.Tasks.Linq
             return new _Take(source, count, cancellationToken);
         }
 
-        sealed class _Take : AsyncEnumeratorBase<TSource, TSource>
+        sealed class _Take : MoveNextSource, IUniTaskAsyncEnumerator<TSource>
         {
-            readonly int count;
+            static readonly Action<object> MoveNextCoreDelegate = MoveNextCore;
 
+            readonly IUniTaskAsyncEnumerable<TSource> source;
+            readonly int count;
+            CancellationToken cancellationToken;
+
+            IUniTaskAsyncEnumerator<TSource> enumerator;
+            UniTask<bool>.Awaiter awaiter;
             int index;
 
             public _Take(IUniTaskAsyncEnumerable<TSource> source, int count, CancellationToken cancellationToken)
-                : base(source, cancellationToken)
             {
+                this.source = source;
                 this.count = count;
+                this.cancellationToken = cancellationToken;
             }
 
-            protected override bool TryMoveNextCore(bool sourceHasCurrent, out bool result)
+            public TSource Current { get; private set; }
+
+            public UniTask<bool> MoveNextAsync()
             {
-                if (sourceHasCurrent)
+                cancellationToken.ThrowIfCancellationRequested();
+
+                if (enumerator == null)
                 {
-                    if (checked(index++) < count)
+                    enumerator = source.GetAsyncEnumerator(cancellationToken);
+                }
+
+                if (checked(index) >= count)
+                {
+                    return CompletedTasks.False;
+                }
+
+                completionSource.Reset();
+                SourceMoveNext();
+                return new UniTask<bool>(this, completionSource.Version);
+            }
+
+            void SourceMoveNext()
+            {
+                try
+                {
+                    awaiter = enumerator.MoveNextAsync().GetAwaiter();
+                    if (awaiter.IsCompleted)
                     {
-                        Current = SourceCurrent;
-                        result = true;
-                        return true;
+                        MoveNextCore(this);
                     }
                     else
                     {
-                        result = false;
-                        return true;
+                        awaiter.SourceOnCompleted(MoveNextCoreDelegate, this);
                     }
                 }
-                else
+                catch (Exception ex)
                 {
-                    result = false;
-                    return true;
+                    completionSource.TrySetException(ex);
                 }
             }
+
+            static void MoveNextCore(object state)
+            {
+                var self = (_Take)state;
+
+                if (self.TryGetResult(self.awaiter, out var result))
+                {
+                    if (result)
+                    {
+                        self.index++;
+                        self.Current = self.enumerator.Current;
+                        self.completionSource.TrySetResult(true);
+                    }
+                    else
+                    {
+                        self.completionSource.TrySetResult(false);
+                    }
+                }
+            }
+
+            public UniTask DisposeAsync()
+            {
+                if (enumerator != null)
+                {
+                    return enumerator.DisposeAsync();
+                }
+                return default;
+            }
         }
     }
 }

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

@@ -66,6 +66,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             float elapsed;
             bool dueTimePhase;
+            bool completed;
             bool disposed;
 
             public _Timer(TimeSpan dueTime, TimeSpan? period, PlayerLoopTiming updateTiming, bool ignoreTimeScale, CancellationToken cancellationToken)
@@ -91,7 +92,7 @@ namespace Cysharp.Threading.Tasks.Linq
             public UniTask<bool> MoveNextAsync()
             {
                 // return false instead of throw
-                if (disposed || cancellationToken.IsCancellationRequested) return CompletedTasks.False;
+                if (disposed || cancellationToken.IsCancellationRequested || completed) return CompletedTasks.False;
 
                 // reset value here.
                 this.elapsed = 0;
@@ -131,6 +132,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 {
                     if (period == null)
                     {
+                        completed = true;
                         completionSource.TrySetResult(false);
                         return false;
                     }
@@ -172,6 +174,7 @@ namespace Cysharp.Threading.Tasks.Linq
 
             int currentFrame;
             bool dueTimePhase;
+            bool completed;
             bool disposed;
 
             public _TimerFrame(int dueTimeFrameCount, int? periodFrameCount, PlayerLoopTiming updateTiming, CancellationToken cancellationToken)
@@ -195,7 +198,7 @@ namespace Cysharp.Threading.Tasks.Linq
             public UniTask<bool> MoveNextAsync()
             {
                 // return false instead of throw
-                if (disposed || cancellationToken.IsCancellationRequested) return CompletedTasks.False;
+                if (disposed || cancellationToken.IsCancellationRequested || completed) return CompletedTasks.False;
 
 
                 // reset value here.
@@ -235,6 +238,7 @@ namespace Cysharp.Threading.Tasks.Linq
                 {
                     if (periodFrameCount == null)
                     {
+                        completed = true;
                         completionSource.TrySetResult(false);
                         return false;
                     }

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

@@ -1,5 +1,6 @@
 using System;
 using System.Threading;
+using UnityEngine;
 using UnityEngine.UI;
 
 namespace Cysharp.Threading.Tasks
@@ -40,11 +41,6 @@ namespace Cysharp.Threading.Tasks
                         if (rebindOnError && !repeat)
                         {
                             repeat = true;
-                            if (e != null)
-                            {
-                                await e.DisposeAsync();
-                            }
-
                             goto BIND_AGAIN;
                         }
                         else
@@ -106,11 +102,6 @@ namespace Cysharp.Threading.Tasks
                         if (rebindOnError && !repeat)
                         {
                             repeat = true;
-                            if (e != null)
-                            {
-                                await e.DisposeAsync();
-                            }
-
                             goto BIND_AGAIN;
                         }
                         else
@@ -167,11 +158,6 @@ namespace Cysharp.Threading.Tasks
                         if (rebindOnError && !repeat)
                         {
                             repeat = true;
-                            if (e != null)
-                            {
-                                await e.DisposeAsync();
-                            }
-
                             goto BIND_AGAIN;
                         }
                         else
@@ -195,6 +181,63 @@ namespace Cysharp.Threading.Tasks
             }
         }
 
+        // <T> -> Action
+
+        public static void BindTo<TSource, TObject>(this IUniTaskAsyncEnumerable<TSource> source, TObject monoBehaviour, Action<TObject, TSource> bindAction, bool rebindOnError = true)
+            where TObject : MonoBehaviour
+        {
+            BindToCore(source, monoBehaviour, bindAction, monoBehaviour.GetCancellationTokenOnDestroy(), rebindOnError).Forget();
+        }
+
+        public static void BindTo<TSource, TObject>(this IUniTaskAsyncEnumerable<TSource> source, TObject bindTarget, Action<TObject, TSource> bindAction, CancellationToken cancellationToken, bool rebindOnError = true)
+        {
+            BindToCore(source, bindTarget, bindAction, cancellationToken, rebindOnError).Forget();
+        }
+
+        static async UniTaskVoid BindToCore<TSource, TObject>(IUniTaskAsyncEnumerable<TSource> source, TObject bindTarget, Action<TObject, TSource> bindAction, CancellationToken cancellationToken, bool rebindOnError)
+        {
+            var repeat = false;
+            BIND_AGAIN:
+            var e = source.GetAsyncEnumerator(cancellationToken);
+            try
+            {
+                while (true)
+                {
+                    bool moveNext;
+                    try
+                    {
+                        moveNext = await e.MoveNextAsync();
+                        repeat = false;
+                    }
+                    catch (Exception ex)
+                    {
+                        if (ex is OperationCanceledException) return;
+
+                        if (rebindOnError && !repeat)
+                        {
+                            repeat = true;
+                            goto BIND_AGAIN;
+                        }
+                        else
+                        {
+                            throw;
+                        }
+                    }
+
+                    if (!moveNext) return;
+
+                    bindAction(bindTarget, e.Current);
+                }
+            }
+            finally
+            {
+                if (e != null)
+                {
+                    await e.DisposeAsync();
+                }
+            }
+        }
+
         // TMP
 
 #if UNITASK_TEXTMESHPRO_SUPPORT
@@ -231,11 +274,6 @@ namespace Cysharp.Threading.Tasks
                         if (rebindOnError && !repeat)
                         {
                             repeat = true;
-                            if (e != null)
-                            {
-                                await e.DisposeAsync();
-                            }
-
                             goto BIND_AGAIN;
                         }
                         else

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

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

src/UniTask/Assets/Scenes/SandboxMain.cs

@@ -124,6 +124,20 @@ public class SandboxMain : MonoBehaviour
 
 
     }
+    private async UniTaskVoid HogeAsync()
+    {
+        // await is not over
+        await UniTaskAsyncEnumerable
+            .TimerFrame(10)
+            .ForEachAwaitAsync(async _ =>
+            // .ForEachAsync(_ =>
+            {
+                await UniTask.Delay(1000);
+                Debug.Log(Time.time);
+            });
+
+        Debug.Log("Done");
+    }
 
     void Start()
     {
@@ -136,7 +150,7 @@ public class SandboxMain : MonoBehaviour
 
         RP1 = new AsyncReactiveProperty<int>(999);
 
-
+        HogeAsync().Forget();
 
         RP1.Select(x => x * x).BindTo(text);
 

src/UniTask/Packages/manifest.json

@@ -1,47 +0,0 @@
-{
-  "dependencies": {
-    "com.unity.addressables": "1.8.3",
-    "com.unity.ads": "3.4.4",
-    "com.unity.analytics": "3.3.5",
-    "com.unity.collab-proxy": "1.2.16",
-    "com.unity.ext.nunit": "1.0.0",
-    "com.unity.ide.rider": "1.1.4",
-    "com.unity.ide.visualstudio": "1.0.5",
-    "com.unity.ide.vscode": "1.1.4",
-    "com.unity.purchasing": "2.0.6",
-    "com.unity.test-framework": "1.1.13",
-    "com.unity.textmeshpro": "2.0.1",
-    "com.unity.timeline": "1.2.6",
-    "com.unity.modules.ai": "1.0.0",
-    "com.unity.modules.androidjni": "1.0.0",
-    "com.unity.modules.animation": "1.0.0",
-    "com.unity.modules.assetbundle": "1.0.0",
-    "com.unity.modules.audio": "1.0.0",
-    "com.unity.modules.cloth": "1.0.0",
-    "com.unity.modules.director": "1.0.0",
-    "com.unity.modules.imageconversion": "1.0.0",
-    "com.unity.modules.imgui": "1.0.0",
-    "com.unity.modules.jsonserialize": "1.0.0",
-    "com.unity.modules.particlesystem": "1.0.0",
-    "com.unity.modules.physics": "1.0.0",
-    "com.unity.modules.physics2d": "1.0.0",
-    "com.unity.modules.screencapture": "1.0.0",
-    "com.unity.modules.terrain": "1.0.0",
-    "com.unity.modules.terrainphysics": "1.0.0",
-    "com.unity.modules.tilemap": "1.0.0",
-    "com.unity.modules.ui": "1.0.0",
-    "com.unity.modules.uielements": "1.0.0",
-    "com.unity.modules.umbra": "1.0.0",
-    "com.unity.modules.unityanalytics": "1.0.0",
-    "com.unity.modules.unitywebrequest": "1.0.0",
-    "com.unity.modules.unitywebrequestassetbundle": "1.0.0",
-    "com.unity.modules.unitywebrequestaudio": "1.0.0",
-    "com.unity.modules.unitywebrequesttexture": "1.0.0",
-    "com.unity.modules.unitywebrequestwww": "1.0.0",
-    "com.unity.modules.vehicles": "1.0.0",
-    "com.unity.modules.video": "1.0.0",
-    "com.unity.modules.vr": "1.0.0",
-    "com.unity.modules.wind": "1.0.0",
-    "com.unity.modules.xr": "1.0.0"
-  }
-}