c# - 在 react 性管道中管理状态

Question

我正在构建一个需要扩展 (SelectMany) 然后展平(在本例中为 ToArray)的响应式管道，同时保持对在管道的开始。

这是我正在尝试的伪代码:

return Observable
    .Start(() => this.GetSearchResults(query))
    .SelectMany(results => results.Hits)     // results.Hits is a list of IDs. But there is also has a bool property that I want to keep through to the end of my pipeline
    .SelectMany(hit => GetById(hit.Id))      // asynchronously load each result
    .ToArray()                               // now need to pull all the results together into a containing data structure, and also include the bool flag from above in it
    .Select(resolvedResults => new ...);     // need access to both resolvedResults and the bool mentioned in the first comment above

所以我试图找到一种方法，干净地从管道末端的代码访问在管道开始时确定的某些状态。

我尝试的第一件事是使用匿名类型将 bool 与每个结果捆绑在一起。这很快就失控了，从性能的角度来看是一种浪费。

我尝试的第二件事是使用如下主题:

var state = new AsyncSubject<bool>();
return Observable
    .Start(() => this.GetSearchResults(query))
    .Do(results =>
        {
            state.OnNext(results.Flag);
            state.OnCompleted();
        }
    .SelectMany(results => results.Hits)
    .SelectMany(hit => GetById(hit.Id))
    .ToArray()
    .Zip(
        state,
        (results, state) => new ResultContainer(state, results));

这似乎工作正常，但我觉得有点恶心。

所以我想知道是否有一种更简洁的方法来管理响应式(Reactive)管道中的状态。

作为引用，这里是实际代码(而不仅仅是伪代码):

public IObservable<ISearchResults<IContact>> Search(string query, int maximumResultCount = 100, float minimumScore = 0.1F)
{
    Ensure.ArgumentNotNull(query, nameof(query));

    var moreHitsAvailable = new AsyncSubject<bool>();

    return Observable
        .Start(
            () => this.searchIndexService.Search<IContact>(query, maximumResultCount, minimumScore),
            this.schedulerService.DataStoreScheduler)
        .Do(
            results =>
            {
                moreHitsAvailable.OnNext(results.MoreHitsAreAvailable);
                moreHitsAvailable.OnCompleted();
            })
        .SelectMany(
            results => results
                .Hits
                .Select(
                    hit => new
                    {
                        Id = hit.Id,
                        ParsedId = ContactId.Parse(hit.Id)
                    }))
        .SelectMany(
            result => this
                .GetById(result.ParsedId)
                .Select(
                    contact => new
                    {
                        Id = result.Id,
                        Contact = contact
                    }))
        .Do(
            result =>
            {
                if (result.Contact == null)
                {
                    this.logger.Warn("Failed to find contact with ID '{0}' provided by the search index. Index may be out of date.", result.Id);
                }
            })
        .Select(result => result.Contact)
        .Where(contact => contact != null)
        .ToArray()
        .Zip(
            moreHitsAvailable,
            (results, more) => new SearchResults<IContact>(more, results.ToImmutableList()))
        .PublishLast()
        .ConnectUntilCompleted();
}

Answer 1

你可以弹出查询理解语法并做这样的事情

var x = from result in Observable.Start(() => this.GetSearchResults())
    let hasMore = result.MoreHitsAreAvailable
    from hit in result.Hits
    from contact in GetById(hit.Id)
    select new { hasMore , contact};

交给你如何处理重复的 hasMore 值。正如我们所知，它只是您可以分组的唯一不同值(所有 true 或所有 false)。