Unityのコルーチンの分解、或いはUniRxのMainThreadDispatcherについて

void Start()
{
    Debug.Log("begin-start:" + Time.frameCount);
    StartCoroutine(MyCoroutine());
    Debug.Log("end-start" + Time.frameCount);
}
 
IEnumerator MyCoroutine()
{
    Debug.Log("start-coroutine:" + Time.frameCount);
 
    yield return null;
    Debug.Log("after-yield-null:" + Time.frameCount);
 
    yield return new WaitForSeconds(3);
    Debug.Log("end-coroutine:" + Time.frameCount);
}

begin-start:1
start-coroutine:1
end-start:1
after-yield-null:2
end-coroutine:168

// 偶数のシーケンスを生成
IEnumerable<int> EvenSequence(int from, int to)
{
    for (int i = from; i <= to; i++)
    {
        if (i % 2 == 0)
        {
            yield return i;
        }
    }
}
 
void Run()
{
    var seq = EvenSequence(1, 10);
 
    // シーケンスはforeachで消費可能
    foreach (var item in seq)
    {
        Debug.Log(item);
    }
 
    // あるいはEnumeratorを取得し回す(foreachは↓のコードを生成する)
    // Unityでのコルーチンでの利用され方はこっちのイメージのほうが近い
    using (var e = seq.GetEnumerator())
    {
        while (e.MoveNext())
        {
            Debug.Log(e.Current);
        }
    }
}

public class CoroutineConsumer : MonoBehaviour
{
    public IEnumerator TargetCoroutine; // 何か外からセットしといて
 
    void Update()
    {
        if (TargetCoroutine.MoveNext())
        {
            var current = TargetCoroutine.Current;
            // 基本的にCurrent自体はそんな意味を持たないで次フレームに回すだけ
            if (current == null)
            {
                // next frame
            }
            // ただしもし固有の何かが返された時はちょっとした別の挙動する
            if (current is WaitForSeconds)
            {
                // なんか適当に秒数待つ（ThreadをSleepするんじゃなく挙動的には次フレームへ）
            }
            else if (current is WWW)
            {
                // isDoneになってるまで適当に待つ（ThreadをSleepするんじゃなく挙動的には次フレームへ）
            }
            // 以下略
        }
    }
}

// こんなのがあるとして
IEnumerator CoroutineA()
{
    Debug.Log("a start");
    yield return new WaitForSeconds(1);
    Debug.Log("a end");
}
 
// こんなふうに使える
Observable.FromCoroutine(CoroutineA)
    .Subscribe(_ => Debug.Log("complete"));
 
// 戻り値のあるバージョンがあるとして
IEnumerator CoroutineB(IObserver<int> observer)
{
    observer.OnNext(100);
    yield return new WaitForSeconds(2);
    observer.OnNext(200);
    observer.OnCompleted();
}
 
// こんなふうに合成もできる
var coroutineA = Observable.FromCoroutine(CoroutineA);
var coroutineB = Observable.FromCoroutine<int>(observer => CoroutineB(observer));
 
// Aが終わった後にBの起動、Subscribeには100, 200が送られてくる
var subscription = coroutineA.SelectMany(coroutineB).Subscribe(x => Debug.Log(x));
 
// Subscribeの戻り値からDisposeを呼ぶとキャンセル可能
// subscription.Dispose();

// Func<IEnumerator>はメソッド宣言的には「IEnumerator Hoge()」になる
public static IObservable<Unit> FromCoroutine(Func<IEnumerator> coroutine, bool publishEveryYield = false)
{
    return FromCoroutine<Unit>((observer, cancellationToken) => WrapEnumerator(coroutine(), observer, cancellationToken, publishEveryYield));
}
 
// ↑のはWrapEnumeratorを介してこれになっている
public static IObservable<T> FromCoroutine<T>(Func<IObserver<T>, CancellationToken, IEnumerator> coroutine)
{
    return Observable.Create<T>(observer =>
    {
        var cancel = new BooleanDisposable();
 
        MainThreadDispatcher.SendStartCoroutine(coroutine(observer, new CancellationToken(cancel)));
 
        return cancel;
    });
}
 
// WrapEnumeratorの中身は(ｵｪｪｪｪ
static IEnumerator WrapEnumerator(IEnumerator enumerator, IObserver<Unit> observer, CancellationToken cancellationToken, bool publishEveryYield)
{
    var hasNext = default(bool);
    var raisedError = false;
    do
    {
        try
        {
            hasNext = enumerator.MoveNext();
        }
        catch (Exception ex)
        {
            try
            {
                raisedError = true;
                observer.OnError(ex);
            }
            finally
            {
                var d = enumerator as IDisposable;
                if (d != null)
                {
                    d.Dispose();
                }
            }
            yield break;
        }
        if (hasNext && publishEveryYield)
        {
            try
            {
                observer.OnNext(Unit.Default);
            }
            catch
            {
                var d = enumerator as IDisposable;
                if (d != null)
                {
                    d.Dispose();
                }
                throw;
            }
        }
        if (hasNext)
        {
            yield return enumerator.Current; // yield inner YieldInstruction
        }
    } while (hasNext && !cancellationToken.IsCancellationRequested);
 
    try
    {
        if (!raisedError && !cancellationToken.IsCancellationRequested)
        {
            observer.OnNext(Unit.Default); // last one
            observer.OnCompleted();
        }
    }
    finally
    {
        var d = enumerator as IDisposable;
        if (d != null)
        {
            d.Dispose();
        }
    }
}

public sealed class MainThreadDispatcher : MonoBehaviour
{
    // 中略
 
    /// <summary>ThreadSafe StartCoroutine.</summary>
    public static void SendStartCoroutine(IEnumerator routine)
    {
#if UNITY_EDITOR
        if (!Application.isPlaying) { EditorThreadDispatcher.Instance.PseudoStartCoroutine(routine); return; }
#endif
 
        if (mainThreadToken != null)
        {
            StartCoroutine(routine);
        }
        else
        {
            Instance.queueWorker.Enqueue(() => Instance.StartCoroutine_Auto(routine));
        }
    }
 
    new public static Coroutine StartCoroutine(IEnumerator routine)
    {
#if UNITY_EDITOR
        if (!Application.isPlaying) { EditorThreadDispatcher.Instance.PseudoStartCoroutine(routine); return null; }
#endif
 
        return Instance.StartCoroutine_Auto(routine);
    }
}

class EditorThreadDispatcher
{
    // 中略
 
    ThreadSafeQueueWorker editorQueueWorker= new ThreadSafeQueueWorker();
 
    EditorThreadDispatcher()
    {
        UnityEditor.EditorApplication.update += Update;
    }
 
    // 中略
 
    void Update()
    {
        editorQueueWorker.ExecuteAll(x => Debug.LogException(x));
    }
 
    // 中略
 
    public void PseudoStartCoroutine(IEnumerator routine)
    {
        editorQueueWorker.Enqueue(() => ConsumeEnumerator(routine));
    }
 
    void ConsumeEnumerator(IEnumerator routine)
    {
        if (routine.MoveNext())
        {
            var current = routine.Current;
            if (current == null)
            {
                goto ENQUEUE;
            }
 
            var type = current.GetType();
            if (type == typeof(WWW))
            {
                var www = (WWW)current;
                editorQueueWorker.Enqueue(() => ConsumeEnumerator(UnwrapWaitWWW(www, routine)));
                return;
            }
            else if (type == typeof(WaitForSeconds))
            {
                var waitForSeconds = (WaitForSeconds)current;
                var accessor = typeof(WaitForSeconds).GetField("m_Seconds", BindingFlags.Instance | BindingFlags.GetField | BindingFlags.NonPublic);
                var second = (float)accessor.GetValue(waitForSeconds);
                editorQueueWorker.Enqueue(() => ConsumeEnumerator(UnwrapWaitForSeconds(second, routine)));
                return;
            }
            else if (type == typeof(Coroutine))
            {
                Debug.Log("Can't wait coroutine on UnityEditor");
                goto ENQUEUE;
            }
 
        ENQUEUE:
            editorQueueWorker.Enqueue(() => ConsumeEnumerator(routine)); // next update
        }
    }
 
    IEnumerator UnwrapWaitWWW(WWW www, IEnumerator continuation)
    {
        while (!www.isDone)
        {
            yield return null;
        }
        ConsumeEnumerator(continuation);
    }
 
    IEnumerator UnwrapWaitForSeconds(float second, IEnumerator continuation)
    {
        var startTime = DateTimeOffset.UtcNow;
        while (true)
        {
            yield return null;
 
            var elapsed = (DateTimeOffset.UtcNow - startTime).TotalSeconds;
            if (elapsed >= second)
            {
                break;
            }
        };
        ConsumeEnumerator(continuation);
    }
}