1

目的は、非同期的に呼び出されたいくつかの処理を同期させることです。
問題は、タイトルの方法が上手くいかず、原因がわからないことです。
また、もっと良いシンプルな方法があれば、教えてください。

まず、非同期的な処理をコルーチンで記述し、それをIObservableに変換したものをMergeしてConcatするということを考えました。

IEnumerator Cor1 () {
    yield return new WaitForSeconds (3);
    Debug.Log ("Cor1: waited 3 sec");
    yield return null;
}

IEnumerator Cor2 () {
    yield return new WaitForSeconds (1);
    Debug.Log ("Cor2: waited 1 sec");
    yield return new WaitForSeconds (9);
    Debug.Log ("Cor2: waited 9 sec");
    yield return null;
}

void DoCoroutineOnObservable () {
    var a = new Subject<Unit> ();
    var b = new Subject<Unit> ();
    var obs1 = a.Select (_ => Observable.FromCoroutine (Cor1));
    var obs2 = b.Select (_ => Observable.FromCoroutine (Cor2));
    Observable.Merge (obs1, obs2).Concat ().Subscribe ();

    // 発行
    a.OnNext (Unit.Default);
    b.OnNext (Unit.Default);
    a.OnNext (Unit.Default);
}

これは想定通り、以下のように出力されます。

cor1: waited 3 sec
cor2: waited 1 sec
cor2: waited 9 sec
cor1: waited 3 sec

しかし、同様の方法でコルーチンをUniTaskに変えてみたのですが、上手くいきません。

async UniTask<Unit> Task1 () {
    await UniTask.Delay (3000); // note! Delayはミリ秒指定
    Debug.Log ("Task1: waited 3 sec");
    return Unit.Default;
}

async UniTask<Unit> Task2 () {
    await UniTask.Delay (1000);
    Debug.Log ("Task2: waited 1 sec");
    await UniTask.Delay (9000);
    Debug.Log ("Task2: waited 9 sec");
    return Unit.Default;
}

void DoUniTaskOnObservable () {
    // ストリームの構築
    var a = new Subject<Unit> ();
    var b = new Subject<Unit> ();
    var obs1 = a.Select (_ => Task1 ().ToObservable ());
    var obs2 = b.Select (_ => Task2 ().ToObservable ());
    Observable.Merge (obs1, obs2).Concat ().Subscribe ();

    // 発行
    a.OnNext (Unit.Default);
    b.OnNext (Unit.Default);
    a.OnNext (Unit.Default);
}

これは上手くいかず、以下のように出力されます。

Task2: waited 1 sec
Task1: waited 3 sec
Task1: waited 3 sec
Task2: waited 9 sec

2番目と3番目のTask1の出力は同時にされます。
つまり、全ての呼び出しが並行(非同期的)に行われているのです。

どのようにすれば、同期的に処理できるでしょうか?

1 件の回答 1

2

Unityに詳しくないので普通のRxの話になるのですが、
Merge が Subscribe された時点で obs1, obs2 共に Subscribe されます。
そして a, b が発火すれば Task が開始されます。
obs の Select のラムダ式の中で Task を開始していることに注意してください。
ToObservable は開始済みのタスクを、タスク終了時に一つだけ値を送出して終了するストリームに変換するものです。
Concat によって直列化されるのはこの Task の結果のみです。

解決法ですが、そもそもの目的によって違ってくるのですが、
単にその順序でロギング部分の処理を行いたいのであれば、処理に必要なパラメータ ("Task1: waited 3 sec") もしくは処理自身 ( ()=>Debug.Log ("Task1: waited 3 sec") ) を Task の戻り値として Concat の Subscribe で処理すればよいと思います。

処理全体を直列化したいのであれば

var obs1 = a.Select(_ => Observable.FromAsync(() => Task1()));
var obs2 = b.Select(_ => Observable.FromAsync(() => Task2()));

とすればよいです。

1
  • FromAsync! ありがとうございます!
    – kazurego7
    Commented 2018年11月12日 10:20

この質問に回答するには、ログインする必要があります。

求めていた回答ではありませんか? のタグが付いた他の質問を参照する。