使用“ RX”的意外行为在.net

Question

我观察到我没想到的行为。这是一些要说明的代码：

class ReactiveService
{
    private readonly ISubject<WorkPayload> observableWork
        = new Subject<WorkPayload>();

    void WorkBeingDone(WorkPayload workPayload)
    {
        this.observableWork.OnNext(workPayload);
    }

    IObservable<WorkPayload> WorkBeingDone() => this.observableWork;
}

class WorkService
{
    private readonly ReactiveService reactiveService;

    public WorkService(ReactiveService reactiveService)
        => this.reactiveService = reactiveService;

    Task DoWork()
    {
        this.reactiveService.WorkBeingDone(new WorkPayload());
        // ... lots of logic here ...
        return Task.CompletedTask;
    }
}

class WorkObservingService
{
    private readonly ReactiveService reactiveService;

    public WorkObservingService(ReactiveService reactiveService)
        => this.reactiveService = reactiveService;

    async Task DoSomethingWhenWorkHappens()
    {
        // Now here what I want to do is that on each work being observed,
        // some asynchronous operation is triggered, currently I'm doing it
        // like this, but I'm open to any other suggestion.
        this.reactiveService.WorkBeingDone().Select(
            w => Observable.FromAsync(async () =>
            {
                // Do the asynchronous stuff with w...
            })).Concat().Subscribe();
        // Prevent the method from returning or something keep
        // the subscription alive...
    }
}

似乎发生的事情是，当我尝试在observable.fromasync中做一些异步工作时，它会一直升起，返回到> > Workservice并将调用撞向dowork。但实际上，这件事不应该关心可观察工作的观察者发生的情况。那是预期的吗？如何防止这种情况发生？

Answer 1

可观察到的序列是等待的，因此您可以简单地执行此操作：

async Task DoSomethingWhenWorkHappens()
{
    await this.reactiveService.WorkBeingDone().Select(
        w => Observable.FromAsync(async () =>
        {
            // Do the asynchronous stuff with w...
        })).Concat().DefaultIfEmpty();
}

通过序列传播的任何错误都将传递给结果task，在A

链条末尾的defaultifempty运算符的目的是防止invalidoperationException，以防序列发射零元素。 等待旨在传播序列中的最后一个元素，因此当序列不包含任何元素时，它会感到困惑。

另一个选项是 elide async and等待，以及等待，只是将序列转换为任务（.totask（）），然后直接返回。

似乎发生的事情是，当我尝试在observable.fromasync中进行一些异步工作时，它会一直升至Workservice并将呼叫撞向dowork。

我没有运行您的代码，但是我的理论是observable.fromasync中的错误被升级为崩溃过程的未手动异常。我不希望您可以通过在中包装/catch将其处理在dowork中。发生这种情况是因为裸体.subscribe（），它缺少所有三个处理程序（OnNext，onerror anderor 和on Completed >）。通过特别省略onerror处理程序，您对RX说“嘿！我不知道如何处理可能的错误！” 。猜猜是什么，RX不知道如何处理此错误，因此它可以说是最明智和负责任的事情，并在threadpool上重新启动它，从而导致您的应用程序崩溃。这不是很友善，但另一方面，它可能比抑制错误并将其静静地扔进内存孔更好。