在控制器中等待 Rx Observable 结果永远不会产生任何结果答案

【问题标题】：Await Rx Observable result in Controller never produces any result在控制器中等待 Rx Observable 结果永远不会产生任何结果
【发布时间】：2020-10-06 14:11:32
【问题描述】：

我有一个带有 POST 方法的 Web API 控制器，我想用它来刷新我的数据并返回它。我的 Crawler 获取 HTML，对其进行解析，并向 SourceObservable 发出一个值，其中包含已解析的数据。一旦Repository.Save(accreditationData) 实际保存，它就可以正常工作。但是返回这个 accreditationDataObservable 不起作用，所以我的 Action 永远不会结束响应，Postman 看起来像是无限期地等待：

[HttpPost]
public async Task<AccreditationData<AllMantainedTableRow>> Post()
{
    var savedSubject = new Subject<bool>();

    AllMantainedCrawler.SourceObservable.Subscribe(accreditationData => {
        Repository.Save(accreditationData);
        savedSubject.OnNext(true);
    });

    var accreditationDataObservable = AllMantainedCrawler.SourceObservable.TakeUntil(savedSubject.AsObservable());

    AllMantainedCrawler.SourceSubject.OnNext(new Uri("my URL here"));

    return await accreditationDataObservable;
}

我也尝试了Take(1) 而不是TakeUntil 并且还返回ToTask() 而不是可观察的，但得到了相同的结果。有什么建议吗？

【问题讨论】：

你为什么使用 Observable 而不是 Task？这会影响其他地方吗？
Crawler 有这个 Observable，这是监听新鲜数据的方式。所以我想设置订阅，使用OnNext 请求抓取并返回可观察到的第一个值

标签： c# .net observable system.reactive

【解决方案1】：

AllMantainedCrawler.SourceSubject.OnNext(new Uri("my URL here"));

此行强制爬虫发出。因此savedSubject 也会发出。

var accreditationDataObservable = AllMantainedCrawler.SourceObservable.TakeUntil(savedSubject.AsObservable());

如果 saveSubject 发出上述行完成 accreditationDataObservable 但在最后一行你尝试等待已经完成的 accreditationDataObservable。

如果您删除 OnNext 行并让爬虫发出，您可以等待 accreditationDataObservable 完成。

【讨论】：

【解决方案2】：

永远不要在Subscribe 中调用OnNext。总有办法避免它。

使用您的代码，如果没有OnNext，这似乎是等效的：

public async Task<AccreditationData<AllMantainedTableRow>> Post()
{
    var accreditationDataObservable = AllMantainedCrawler.SourceObservable.Take(1).Do(x => Repository.Save(x));
    AllMantainedCrawler.SourceSubject.OnNext(new Uri("my URL here"));
    return await accreditationDataObservable;
}

但是，我无法理解 AllMantainedCrawler.SourceSubject.OnNext(new Uri("my URL here")); 在这里做什么，因为您没有在问题中提供足够的细节。如果你能提供完整的代码，我想我可以帮助你。

如前所述，我不知道SourceSubject 与SourceObservable 有何关系，但如果我可以假设accreditationData 只是Uri，那么这将起作用：

public async Task<AccreditationData<AllMantainedTableRow>> Post()
{
    return await Observable.Start(() => Repository.Save(new Uri("my URL here")));
}

【讨论】：