ChannelExample.cs

using System;
using System.IO;
using System.Linq;
using System.Net.Http;
using System.Threading;
using Cysharp.Threading.Tasks;
using UnityEngine;
using UnityEngine.UI;

public class ChannelExample : MonoBehaviour
{
    [SerializeField]
    private RectTransform _content = default;

    [SerializeField]
    private RawImage _rawImagePrefab = default;

    private async void Start()
    {
        BigCat();
    }

    private System.Threading.Channels.Channel<string> _channelDownload = default;
    private System.Threading.Channels.Channel<byte[]> _channelParse = default;

    private async UniTask BigCat()
    {
        var baseUrl = "http://localhost:8888";
        var urlArray = new string[]
            {
                "20141103-PB030062.jpg",
                "kazu79_toratora.jpg",
                "newyork4M3A7141.jpg",
                "takeru522002.jpg",
                "tora4M3A7204.jpg",
            }
            .Select(s => Path.Combine(baseUrl, s)).ToArray();

        var destroyToken = this.GetCancellationTokenOnDestroy();
        var cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(destroyToken);
        var token = cancellationTokenSource.Token;
        _channelDownload = System.Threading.Channels.Channel.CreateUnbounded<string>();
        _channelParse = System.Threading.Channels.Channel.CreateUnbounded<byte[]>();

        const int parallelDownload = 2;
        const int parallelParse = 1;

        for (var count = 0; count < parallelDownload; count++)
        {
            Download(_channelDownload.Reader, _channelParse.Writer, token).Forget();
        }

        for (var count = 0; count < parallelParse; count++)
        {
            Parse(_channelParse.Reader, token).Forget();
        }

        var writer = _channelDownload.Writer;
        foreach (var url in urlArray)
        {
            await writer.WriteAsync(url, token);
        }

        _channelDownload.Writer.TryComplete();
    }

    private static async UniTask Download(
        System.Threading.Channels.ChannelReader<string> reader,
        System.Threading.Channels.ChannelWriter<byte[]> writer,
        CancellationToken token)
    {
        while (await reader.WaitToReadAsync(token))
        {
            if (reader.TryRead(out var uri))
            {
                await UniTask.SwitchToThreadPool();
                var buffer = await DownloadTextureBuffer(uri, token);
                await writer.WriteAsync(buffer, token);
            }
        }
    }

    private async UniTask Parse(System.Threading.Channels.ChannelReader<byte[]> reader, CancellationToken token)
    {
        while (await reader.WaitToReadAsync(token))
        {
            if (reader.TryRead(out var item))
            {
                await UniTask.SwitchToMainThread();
                var texture = ParseTexture(item);
                InstantiateTexture(texture);
            }
        }
    }

    private static HttpClient _httpClient = null;
    private static HttpClient HttpClient => _httpClient ?? (_httpClient = new HttpClient());

    private static async UniTask<byte[]> DownloadTextureBuffer(string uri, CancellationToken token)
    {
        using (var response = await HttpClient.GetAsync(new Uri(uri), HttpCompletionOption.ResponseHeadersRead, token))
        {
            response.EnsureSuccessStatusCode();
            return await response.Content.ReadAsByteArrayAsync();
        }
    }

    private Texture2D ParseTexture(in byte[] buffer)
    {
        var texture = new Texture2D(0, 0, TextureFormat.RGBA32, false);
        texture.LoadImage(buffer);
        return texture;
    }

    private void InstantiateTexture(Texture texture)
    {
        var raw = Instantiate(_rawImagePrefab, _content);
        raw.texture = texture;
        raw.gameObject.SetActive(true);
    }

    // private async UniTask TryChannel()
    // {
    //     var token = this.GetCancellationTokenOnDestroy();
    //     var cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(token);
    //     token = cancellationTokenSource.Token;
    //     _channelLoad = System.Threading.Channels.Channel.CreateUnbounded<string>();
    //     _channelParse = System.Threading.Channels.Channel.CreateUnbounded<string>();
    //
    //     const int count = 10;
    //     const int parallelLoad = 2;
    //     const int parallelParse = 1;
    //
    //     for (var i = 0; i < parallelLoad; i++)
    //     {
    //         Loading(_channelLoad.Reader, _channelParse.Writer, token).Forget();
    //     }
    //
    //     for (var i = 0; i < parallelParse; i++)
    //     {
    //         Parsing(_channelParse.Reader, token).Forget();
    //     }
    //
    //     for (var i = 0; i < count; i++)
    //     {
    //         await _channelLoad.Writer.WriteAsync(i.ToString(), token);
    //     }
    // }

    //
    // private static async UniTask Loading(
    //     System.Threading.Channels.ChannelReader<string> reader,
    //     System.Threading.Channels.ChannelWriter<string> writer,
    //     CancellationToken token)
    // {
    //     while (await reader.WaitToReadAsync(token))
    //     {
    //         if (reader.TryRead(out var item))
    //         {
    //             await UniTask.SwitchToThreadPool();
    //             await Load(item, token);
    //             await writer.WriteAsync(item, token);
    //         }
    //     }
    // }
    //
    // private static async UniTask Parsing(System.Threading.Channels.ChannelReader<string> reader, CancellationToken token)
    // {
    //     while (await reader.WaitToReadAsync(token))
    //     {
    //         if (reader.TryRead(out var item))
    //         {
    //             await UniTask.SwitchToMainThread();
    //             Parse(item);
    //         }
    //     }
    // }
    //
    //
    // private static async UniTask<string> Load(string url, CancellationToken token)
    // {
    //     // Debug.Log($"L-START {url}");
    //     var random = new System.Random();
    //     await UniTask.Delay(TimeSpan.FromSeconds(random.Next(5)), cancellationToken: token);
    //     Debug.Log($"L-END {url}");
    //     return url;
    // }
    //
    // private static async UniTask Parse(string url)
    // {
    //     Debug.Log($"P -{url}");
    // }
}