是否可以向Rx.Net Timeout操作符添加自定义消息

Question

我正在尝试从设备中读取数据流，并在其中找到有效数据。它是一组零，中间有一些数据。类似于：

0,0,0,1,2,3,2,1,0,0,0,1,2,2,1,00,0,0,0,0,0依此类推

处理后发出的是一组非零数据的总和，例如

9，9

我的代码中有两个超时运算符。不幸的是，他们都有相同的信息：

System.TimeoutException: The operation has timed out.

因此，我无法向用户提供更多的上下文，就好像错误发生在哪里一样。例如，数据流变为静音，或者我们有一个数据流在运行，但我们只得到零。

有没有一种方法可以添加我自己的消息，而不是常规的TimeoutException消息？

这是我的伪代码：

var _handle = 
_source.Select(x => x.Item1)
.Timeout(TimeSpan.FromSeconds(5)) // If the stream goes silent here the timeout is raised.
.EmitSumOfDataNonZeroPoints()
.Timeout(TimeSpan.FromSeconds(15)) // If no valid group of nonzero data is found the timeout is raised.
.Subscribe(
someProcessing => {},
ex => {
_errorInObservableMessage = ex.Message;
},
() => {}
);

Answer 1

Timeout运算符有一个重载，您可以在其中传递一个other可观察到的值，该值将在超时时继续：

// Applies a timeout policy for each element in the observable sequence. If the
// next element isn't received within the specified timeout duration starting from
// its predecessor, the other observable sequence is used to produce future messages
// from that point on.
public static IObservable<TSource> Timeout<TSource>(this IObservable<TSource> source,
TimeSpan dueTime, IObservable<TSource> other);

您可以直接使用此重载，也可以将其用作接受timeoutMessage参数的自定义Timeout运算符的构建块：

/// <summary>In case of timeout propagates a TimeoutException with custom message.</summary>
public static IObservable<TSource> Timeout<TSource>(
this IObservable<TSource> source, TimeSpan dueTime, string timeoutMessage)
{
return source.Timeout(dueTime,
Observable.Throw<TSource>(new TimeoutException(timeoutMessage)));
}