第一个或最后一个IOObservable扩展

Question

我想运行IObservable<T>，查找与谓词匹配的元素，如果找不到，则返回IObservable<T>的最后一个元素。我不想存储IObservable<T>的全部内容，也不想在IObservable中循环两次，所以我设置了一个扩展方法

public static class ObservableExtensions
{
    public static IObservable<T> FirstOrLastAsync<T>(this IObservable<T> source, Func<T, bool> pred)
    {
        return Observable.Create<T>(o =>
        {
            var hot = source.Publish();
            var store = new AsyncSubject<T>();
            var d1 = hot.Subscribe(store);
            var d2 = hot.FirstAsync(x => pred(x)).Amb(store).Subscribe(o);
            var d3 = hot.Connect();
            return new CompositeDisposable(d1, d2, d3);
        });
    }
    public static T FirstOrLast<T>(this IObservable<T> source, Func<T, bool> pred)
    {
        return source.FirstOrLastAsync(pred).Wait();
    }
}

Async方法从传入的可能较冷的元素创建一个热的可观察对象。它订阅一个AsyncSubject<T>来记住最后一个元素，并订阅一个查找该元素的IObservable<T>。然后，它从这些IObservable<T>中的任何一个中获取第一个元素，该元素首先通过.Amb返回值（AsyncSubject<T>在获得.OnCompleted消息之前不会返回值）。

我的问题如下：

• 使用不同的Observable方法可以写得更好或更简洁吗
• 所有这些一次性用品都需要包含在CompositeDisposable中吗
• 当热可观察对象在没有找到匹配元素的情况下完成时，FirstAsync抛出异常和AsyncSubject传播其值之间是否存在竞争条件
• 如果是，我是否需要将行更改为：

var d2 = hot.Where(x => pred(x)).Take(1).Amb(store).Subscribe(o);

我是RX的新手，这是我在IOobservable上的第一个扩展。

编辑

我最终选择了

public static class ObservableExtensions
{
    public static IObservable<T> FirstOrLastAsync<T>(this IObservable<T> source, Func<T, bool> pred)
    {
        var hot = source.Publish().RefCount();
        return hot.TakeLast(1).Amb(hot.Where(pred).Take(1).Concat(Observable.Never<T>()));
    }
    public static T FirstOrLast<T>(this IObservable<T> source, Func<T, bool> pred)
    {
        return source.FirstOrLastAsync(pred).First();
    }
}

Answer 1

您可以将您想要的两个案例放在一起。如果你的观测源是冷的，你可以做一个Publish|Refcount。

    public static IObservable<T> FirstOrLast<T>(this IObservable<T> source, Func<T, bool> predicate)
    {
        return source.TakeLast(1).Amb(source.Where(predicate).Take(1));
    }

测试：

        var source = Observable.Interval(TimeSpan.FromSeconds(0.1))
                               .Take(10)
                               .Publish()
                               .RefCount();
        FirstOrLast(source, i => i == 5).Subscribe(Console.WriteLine); //5
        FirstOrLast(source, i => i == 11).Subscribe(Console.WriteLine); //9

Answer 2

我尝试过生成一个"更简单"的查询，但到目前为止什么都没有。

如果我坚持你的基本结构，我可以稍微改进一下。试试这个：

public static IObservable<T> FirstOrLastAsync<T>(
    this IObservable<T> source, Func<T, bool> pred)
{
    return Observable.Create<T>(o =>
    {
        var hot = source.Publish();
        var store = new AsyncSubject<T>();
        var d1 = hot.Subscribe(store);
        var d2 =
            hot
                .Where(x => pred(x))
                .Concat(store)
                .Take(1)
                .Subscribe(o);
        var d3 = hot.Connect();
        return new CompositeDisposable(d1, d2, d3);
    });
}

它并没有太好，但我喜欢它比使用Amb更好。我想它只是稍微干净一点。