当所有源都包含给定键时，连接未知数量的源

Question

给定如下源提供商:

IObservable<ISource> Sources();

每个ISource如下所示:

IObservable<IEnumerable<string>> ObserveData(string filter)

我想返回:

IObservable<IEnumerable<string>> Results

当给定字符串从所有isource返回时。我想要所有源的交点

如果添加了一个新源，那么所有内容都应该重新评估。

我正在努力想出一个通用的解决方案。我见过的大多数解决方案都有一些众所周知的来源。如有任何意见，欢迎。

好吧，在思考了一会儿之后，我想出了我的答案。可能它可以改进，但它似乎对我有用，所以我会把它贴在这里供参考，以防有人有类似的问题。感谢ibebbs和Shlomo花时间回复，非常感谢。

 //Arrange
        var s1 = Substitute.For<ISource>();
        s1.ObserveData(Arg.Any<string>()).Returns(Observable.Return(new[] { "a", "b", "c", "d" }));
        var s2 = Substitute.For<ISource>();
        s2.ObserveData(Arg.Any<string>()).Returns(Observable.Return(new[] { "b", "xx", "c", "d" }));
        var s3 = Substitute.For<ISource>();
        s3.ObserveData(Arg.Any<string>()).Returns(Observable.Return(new[] { "yy", "b", "ff", "d" }));
        var expected = new[] { "b", "d" };
        var sources = new[] { s1, s2, s3 }.ToObservable();
        var scheduler = new TestScheduler();
        var observer = scheduler.CreateObserver<IList<string>>();
        //Act
        sources.Buffer(TimeSpan.FromMilliseconds(500), scheduler)
            .Select(s => Observable.CombineLatest(s.Select(x => x.ObserveData("NoFilter"))))
            .Switch()
            .Select(x =>IntersectAll(x))
            .Do(x => Console.WriteLine($"Recieved {string.Join("," , x)}"))
            .Subscribe(observer);
        scheduler.AdvanceBy(TimeSpan.FromMilliseconds(500).Ticks);
        //Assert
        observer.Messages.AssertEqual(
            OnNext<IList<string>>(0, s => s.SequenceEqual(expected)),
            OnCompleted<IList<string>>(0));

对于IntersectAll，请参见IEnumerable.Intersect()

Answer 1

好的，第二次尝试，我很确定这是你需要的(测试夹具包含在底部):

public interface ISource
{
    IObservable<IEnumerable<string>> ObserveData(string filter);
}
public static class ArbitrarySources
{
    public static IObservable<IEnumerable<string>> Intersection(this IObservable<ISource> sourceObservable, string filter)
    {
        return sourceObservable
            .SelectMany((source, index) => source.ObserveData(filter).Select(values => new { Index = index, Values = values }))
            .Scan(ImmutableDictionary<int, IEnumerable<string>>.Empty, (agg, tuple) => agg.SetItem(tuple.Index, tuple.Values))
            .Select(dictionary => dictionary.Values.Aggregate(Enumerable.Empty<string>(), (agg, values) => agg.Any() ? agg.Intersect(values) : values).ToArray());       
    }
}
public class IntersectionTest
{
    internal class Source : ISource
    {
        private readonly IObservable<IEnumerable<string>> _observable;
        public Source(IObservable<IEnumerable<string>> observable)
        {
            _observable = observable;
        }
        public IObservable<IEnumerable<string>> ObserveData(string filter)
        {
            return _observable;
        }
    }
    [Fact]
    public void ShouldIntersectValues()
    {
        TestScheduler scheduler = new TestScheduler();
        var sourceA = new Source(scheduler.CreateColdObservable(
            new Recorded<Notification<IEnumerable<string>>>(TimeSpan.FromSeconds(1).Ticks, Notification.CreateOnNext<IEnumerable<string>>(new string[] { "a", "b" })),
            new Recorded<Notification<IEnumerable<string>>>(TimeSpan.FromSeconds(3).Ticks, Notification.CreateOnNext<IEnumerable<string>>(new string[] { "a", "b", "c" }))
        ));
        var sourceB = new Source(scheduler.CreateColdObservable(
            new Recorded<Notification<IEnumerable<string>>>(TimeSpan.FromSeconds(1).Ticks, Notification.CreateOnNext<IEnumerable<string>>(new string[] { "a", "c" })),
            new Recorded<Notification<IEnumerable<string>>>(TimeSpan.FromSeconds(3).Ticks, Notification.CreateOnNext<IEnumerable<string>>(new string[] { "b", "c" }))
        ));
        var sources = scheduler.CreateColdObservable(
            new Recorded<Notification<ISource>>(TimeSpan.FromSeconds(1).Ticks, Notification.CreateOnNext<ISource>(sourceA)),
            new Recorded<Notification<ISource>>(TimeSpan.FromSeconds(2).Ticks, Notification.CreateOnNext<ISource>(sourceB))
        );
        var observer = scheduler.Start(() => sources.Intersection("test"), 0, 0, TimeSpan.FromSeconds(6).Ticks);
        IEnumerable<string>[] actual = observer.Messages
            .Select(message => message.Value)
            .Where(notification => notification.Kind == NotificationKind.OnNext && notification.HasValue)
            .Select(notification => notification.Value)
            .ToArray();
        IEnumerable<string>[] expected = new []
        {
            new [] { "a", "b" },
            new [] { "a" },
            new [] { "a", "c" },
            new [] { "b", "c" }
        };
        Assert.Equal(expected.Length, actual.Length);
        foreach (var tuple in expected.Zip(actual, (e, a) => new { Expected = e, Actual = a }))
        {
            Assert.Equal(tuple.Expected, tuple.Actual);
        }
    }
}

这种方法还有一个额外的好处，即在添加新源时不需要重新查询现有源，而是在任何源每次发出一个值时重新计算交集。

Answer 2

这个怎么样:

public IObservable<IEnumerable<string>> From(this IObservable<ISource> sources, string filter)
{
    return sources
        .Scan(Observable.Empty<IEnumerable<string>>(), (agg, source) => Observable.Merge(agg, source.ObserveData(filter)))
        .Switch();
}

请注意，每次从sources发出新源时，之前发出的所有源都将再次调用它们的ObserveData方法。因此，这个解决方案不能很好地扩展，但确实满足了你的"如果添加了一个新源，那么一切都应该重新评估"的要求