今回の内容

「R3」に登場するForEachAsyncが便利なので活用してみよう、という話です。

執筆時の環境

Unity - 2023.1.14f1
R3 - 1.0.4
UniTask - 2.5.3

ForEachAsyncが便利

ForEachAsyncを用いると「Observableを購読しつつ、Observableの完了をawaitできる」という挙動を実現できます。もうちょっとざっくりいえば「Subscribeしつつ、そのSubscribeが終わるのをawaitできるようにする機能」です。

ForEachAsyncでSubscribeとawaitを両立させる

using System.Collections.Generic;
using Cysharp.Threading.Tasks;
using R3;
using UnityEngine;
using UnityEngine.UI;

public class ForEachSample : MonoBehaviour
{
    [SerializeField] private InputField _inputField;
    [SerializeField] private Button _button;
    [SerializeField] private Text _text;

    private async UniTaskVoid Start()
    {
        var buttonText = _button.GetComponentInChildren<Text>();

        var results = new List<string>();

        buttonText.text = "1回目の入力";

        // ボタンが押されるたびにその時のInputFieldの値を保存する
        // 3回押されたら終了
        await _button.OnClickAsObservable()
            .Select(_inputField, (_, inputField) => inputField.text)
            .Index()
            .Take(3)
            .ForEachAsync(x =>
            {
                // 結果を格納してUIを更新
                results.Add(x.Item);
                buttonText.text = $"{x.Index + 2}回目の入力";
                _inputField.text = "";
            }, destroyCancellationToken);

        _button.gameObject.SetActive(false);
        _inputField.gameObject.SetActive(false);

        // 保存した値を表示する
        _text.text = string.Join(", ", results);
    }
}

応用:Repeat/Retryの代替手段

R3にはRepeat/Retryオペレータが存在しません。そのため一度完了したObservableをオペレータ経由で再生成/再購読することができません。

そのためもし「Observableの再購読」を行いたい場合、ForEachAsyncとループを利用して再購読の処理を実現するとよいでしょう。

Repeat（正常終了時に再購読する）の代替としてのForEachAsync

using System.Collections.Generic;
using Cysharp.Threading.Tasks;
using R3;
using UnityEngine;
using UnityEngine.UI;

public class ForEachSample : MonoBehaviour
{
    [SerializeField] private InputField _inputField;
    [SerializeField] private Button _button;
    [SerializeField] private Text _text;

    private async UniTaskVoid Start()
    {
        var buttonText = _button.GetComponentInChildren<Text>();

        // ループさせれば完了後にまた購読できる
        while (!destroyCancellationToken.IsCancellationRequested)
        {
            var results = new List<string>();

            buttonText.text = "1回目の入力";

            // ボタンが押されるたびにその時のInputFieldの値を保存する
            // 3回押されたら終了
            await _button.OnClickAsObservable()
                .Select(_inputField, (_, inputField) => inputField.text)
                .Index()
                .Take(3)
                .ForEachAsync(x =>
                {
                    // 結果を格納してUIを更新
                    results.Add(x.Item);
                    buttonText.text = $"{x.Index + 2}回目の入力";
                    _inputField.text = "";
                }, destroyCancellationToken);
            
            // 保存した値を表示する
            _text.text = string.Join(", ", results);
        }
    }
}

Retry（エラー発生時に再購読する）の代替

// 失敗時に指定回数までObservableをリトライする
private async UniTask RetryAsync<T>(
    Observable<T> mayBeErrorObservable,
    int maxRetryCount = 3,
    CancellationToken ct = default)
{
    var retryCount = maxRetryCount;

    while (!ct.IsCancellationRequested)
    {
        try
        {
            await mayBeErrorObservable
                // OnErrorResume を OnCompleted(Exception) に変換
                .OnErrorResumeAsFailure()
                .ForEachAsync(x =>
                {
                    // do something...
                }, cancellationToken: ct);
            break;
        }
        catch (Exception)
        {
            if (retryCount-- <= 0) throw;
        }
    }
}