如何为响应式扩展制作自定义扩展

Question

很容易找到一个示例如何制作自定义的LINQ扩展方法。但我找不到如何制作自定义Rx扩展方法的示例。

有人能指向一个资源或发布一个示例吗？我正在使用最新版本（ver 2.2.5）。

我有兴趣制作自定义的“Cast”或自定义的“Select”。我可以反射现有扩展的源代码，但仍然不清楚。

谢谢。

- Brent Arias

Answer 1

实现Rx中自定义扩展的最常见方式（除了组合现有操作之外）是通过Observable.Create。强烈建议使用这种方法，因为Create会处理许多问题，如果你自己实现，则非常复杂。

无论如何，一定要仔细阅读Rx设计指南，其中解释了你应该遵循的规则和约定——其中许多Observable.Create已经为你处理了。

下面是一个Cast实现的示例：

public static class ObservableExtensions
{
    public static IObservable<TResult> Cast<TSource, TResult>(
        this IObservable<TSource> source)
    {
        return Observable.Create<TResult>(o =>
            source.Subscribe(x => {
                TResult cast;
                try
                {
                    cast = (TResult)(object)x;                        
                }
                catch (InvalidCastException ex)
                {
                    o.OnError(ex);
                    return;
                }
                o.OnNext(cast);
            },
            o.OnError,
            o.OnCompleted));
    }
}

我将这段代码保留原样，因为双重转换的要求并不明显（实际上，IEnumerable 的 Cast 函数出于同样的原因也会执行相同的操作）。你可以通过进行以下微调来将类型参数减少到一个，前提是需要进行 object 转换：

public static class ObservableExtensions
{
    public static IObservable<TResult> Cast<TResult>(
        this IObservable<object> source)
    {
        return Observable.Create<TResult>(o =>
            source.Subscribe(x => {
                TResult cast;
                try
                {
                    cast = (TResult)x;
                }
                catch (InvalidCastException ex)
                {
                    o.OnError(ex);
                    return;
                }
                o.OnNext(cast);
            },
            o.OnError,
            o.OnCompleted));
    }       
}

你可以查看 IntroToRx 了解更多关于Observable.Create的内容。

- James World

---

只需要一个更正：o.OnNext 不应该在 try 块中。 - Dave Sexton

是的，如果订阅者引起了“InvalidCastException”，它将被处理不当。已修复，谢谢。 - James World

晚来一步，但关于你的第一个实现：如果即使没有人再使用构建的转换可观察对象，订阅源也从未被取消订阅，这不会泄漏资源吗？ - domin

不，这是一个微妙的细节 - 创建逻辑正在传回 IDisposable 以直接从源中取消订阅源。 - James World

Answer 2

通常可以通过组合现有的操作符来创建新的Rx操作符。例如，可以使用内置的Select操作符实现Cast操作符，如下所示：

public static IObservable<TResult> Cast<TSource, TResult>(
    this IObservable<TSource> source)
{
    return source.Select(x => (TResult)(object)x);
}

- Theodor Zoulias

---

+1 这值得明确说明 - 我默认 OP 所在的地方不适用这种方法或需要更精确的控制。尽可能使用现有的运算符是一个好主意，因为这样可以更容易地避免实现失误。 - James World