“合并”流的流以产生每个流的最新值的流

Question

我有一个IObservable<IObservable<T>>，其中每个内部IObservable<T> 是一个值流，后跟一个最终的OnCompleted 事件。

我想将其转换为IObservable<IEnumerable<T>>，这是一个由任何未完成的内部流中的最新值组成的流。每当从内部流之一生成新值（或内部流过期）时，它应该生成一个新的IEnumerable<T>

最容易用大理石图表示（我希望它足够全面）：

input ---.----.---.----------------
         |    |   '-f-----g-|      
         |    'd------e---------|
         'a--b----c-----|          

result ---a--b-b--c-c-c-e-e-e---[]-
               d  d d e f g        
                    f f            

（[] 是空的IEnumerable<T> 而-| 代表 OnCompleted）

您可以看到它有点类似于CombineLatest 操作。 我一直在玩 Join 和 GroupJoin 无济于事，但我觉得这几乎肯定是前进的正确方向。

我想在这个运算符中使用尽可能少的状态。

更新

我已更新此问题，不仅包括单值序列 - 结果 IObservable<IEnumerable<T>> 应仅包括每个序列的最新值 - 如果序列未产生值，则不应包括在内。

Answer 1

这是基于您昨天解决方案的版本，针对新要求进行了调整。基本思想是将引用放入易腐烂的集合中，然后在内部序列产生新值时更新引用的值。

我还进行了修改以正确跟踪内部订阅并在外部 observable 取消订阅时取消订阅。

如果任何流产生错误，也进行了修改以将其全部删除。

最后，我修复了一些可能违反 Rx 指南的竞争条件。如果你的内部 observables 从不同的线程同时触发，你可以同时调用obs.OnNext，这是一个很大的禁忌。因此，我使用相同的锁对每个内部 observable 进行了门控以防止这种情况发生（请参阅Synchronize 调用）。请注意，正因为如此，您可能会使用常规的双链表而不是 PerishableCollection，因为现在使用该集合的所有代码都在锁内，因此您不需要 @987654324 的线程保证@。

// Acts as a reference to the current value stored in the list
private class BoxedValue<T>
{
    public T Value;
    public BoxedValue(T initialValue) { Value = initialValue; }
}

public static IObservable<IEnumerable<T>> MergeLatest<T>(this IObservable<IObservable<T>> source)
{
    return Observable.Create<IEnumerable<T>>(obs =>
    {
        var collection = new PerishableCollection<BoxedValue<T>>();
        var outerSubscription = new SingleAssignmentDisposable();
        var subscriptions = new CompositeDisposable(outerSubscription);
        var innerLock = new object();

        outerSubscription.Disposable = source.Subscribe(duration =>
        {
            BoxedValue<T> value = null;
            var lifetime = new DisposableLifetime(); // essentially a CancellationToken
            var subscription = new SingleAssignmentDisposable();

            subscriptions.Add(subscription);
            subscription.Disposable = duration.Synchronize(innerLock)
                .Subscribe(
                    x =>
                    {
                        if (value == null)
                        {
                            value = new BoxedValue<T>(x);
                            collection.Add(value, lifetime.Lifetime);
                        }
                        else
                        {
                            value.Value = x;
                        }
                        obs.OnNext(collection.CurrentItems().Select(p => p.Value.Value));
                    },
                    obs.OnError, // handle an error in the stream.
                    () => // on complete
                    {
                        if (value != null)
                        {
                            lifetime.Dispose(); // removes the item
                            obs.OnNext(collection.CurrentItems().Select(p => p.Value.Value));
                            subscriptions.Remove(subscription); // remove this subscription
                        }
                    }
            );
        });

        return subscriptions;
    });
}

Answer 2

此解决方案适用于单项流，但不幸的是会将每个项累积到内部流中，直到完成。

public static IObservable<IEnumerable<T>> MergeLatest<T>(this IObservable<IObservable<T>> source)
{
    return Observable.Create<IEnumerable<T>>(obs =>
    {
        var collection = new PerishableCollection<T>();
        return source.Subscribe(duration =>
        {
            var lifetime = new DisposableLifetime(); // essentially a CancellationToken
            duration
                .Subscribe(
                    x => // on initial item
                    {
                        collection.Add(x, lifetime.Lifetime);
                        obs.OnNext(collection.CurrentItems().Select(p => p.Value));
                    },
                    () => // on complete
                    {
                        lifetime.Dispose(); // removes the item
                        obs.OnNext(collection.CurrentItems().Select(p => p.Value));
                    }
            );
        });
    });
}

Answer 3

Another solution given by Dave Sexton，Rxx 的创建者 - 它使用 Rxx.CombineLatest，在实现上似乎与 Brandon 的解决方案非常相似：

public static IObservable<IEnumerable<T>> CombineLatestEagerly<T>(this IObservable<IObservable<T>> source)
{
  return source
    // Reify completion to force an additional combination:
    .Select(o => o.Select(v => new { Value = v, HasValue = true })
                  .Concat(Observable.Return(new { Value = default(T), HasValue = false })))
    // Merge a completed observable to force combination with the first real inner observable:
    .Merge(Observable.Return(Observable.Return(new { Value = default(T), HasValue = false })))
    .CombineLatest()
    // Filter out completion notifications:
    .Select(l => l.Where(v => v.HasValue).Select(v => v.Value));
}