如何使用 UniRx Observable.Timeout 设置超时？

Question

我正在尝试使用 observable.Timeout 设置超时，但结果超出了我的预期。
像这样的代码：

public class ObservableTest : MonoBehaviour {

    // Use this for initialization
    void Start () {
        var subj = new Subject<int> ();
        new System.Threading.Thread (() => {
            Debug.Log("thread id - " + System.Threading.Thread.CurrentThread.ManagedThreadId);
            var i = 0;
            while(i < 10) {
                i += 1;
                System.Threading.Thread.Sleep(i*1000 );
                subj.OnNext(i);
            }
        }).Start();


        var timeout = subj.Timeout (System.TimeSpan.FromSeconds (4));
        timeout.Subscribe (x => Debug.Log("x - " + x), ex => Debug.LogError(ex), () => Debug.Log("completed"));
    }

}

控制台输出：

x - 1
x - 2
...
x - 10

为什么不在x - 5之前中断？如何设置一个observable的超时时间？

更新

x - 8 有时输出中断，我不知道它是如何工作的。

Answer 1

我已经检查了问题。从这篇文章中使用了 Spy() 扩展 > How can I see what my reactive extensions query is doing?

问题可能是调用者的线程没有正确处理。这个对我有用。在 x - 2 后断开连接。

var subj = new Subject<int>();
new Thread(() => {
    Console.WriteLine("thread id - " + Thread.CurrentThread.ManagedThreadId);
    var i = 0;
    while (i < 4)
    {
        i += 1;
        Thread.Sleep(i * 1000);
        subj.OnNext(i);
    }
    subj.OnCompleted();
}).Start();

//var timeout = subj.Timeout(TimeSpan.FromSeconds(0.5));
var timeout = subj.Timeout(TimeSpan.FromSeconds(2));
IDisposable disp = timeout.Spy().Subscribe(x => Debug.WriteLine("x - " + x), ex => Debug.WriteLine(ex.Message), () => Debug.WriteLine("completed"));
// That might be missing.
Thread.CurrentThread.Join(10500);
disp.Dispose();

输出

Debug Trace:
IObservable: Observable obtained on Thread: 14, 03:05:24.28
IObservable: Subscribed to on Thread: 14, 03:05:24.289
IObservable: Subscription completed. 03:05:24.296
IObservable: OnNext(1) on Thread: 17, 03:05:25.258 //<- i = 1. One sec wait.
x - 1
IObservable: OnNext(2) on Thread: 17, 03:05:27.261 //<- i = 2. Two sec wait. It's correct.
x - 2
IObservable: OnError(System.TimeoutException: The operation has timed out.) on Thread: 13, 03:05:29.264 //<- i = 3. Two sec timeout comes into play. Correct. Next step would need 3 sec to finish.
The operation has timed out.
IObservable: Cleaned up on Thread: 13, 03:05:29.266

我有一个不同的输出，这也是正确的

Debug Trace:
IObservable: Observable obtained on Thread: 14, 03:20:17.714
IObservable: Subscribed to on Thread: 14, 03:20:17.723
IObservable: Subscription completed. 03:20:17.73
IObservable: OnNext(1) on Thread: 17, 03:20:18.693 //<- 1 sec passed since subscription. correct
x - 1
IObservable: OnError(System.TimeoutException: The operation has timed out.) on Thread: 12, 03:20:20.698 //<- 2 sec passed, and that was the timeout. There was a race between timeout and OnNext(2). Timeout win. Correct.
The operation has timed out.
IObservable: Cleaned up on Thread: 12, 03:20:20.7