未触发 rxjs 错误订阅

Question

我有一个 .net Core web-api 应用程序。 我故意从控制器返回一个异常，但我没有看到在订阅的错误处理中处理该异常。 这是我的代码（手工复制，因此可能会出现拼写错误）： 控制器：

public IActionResult getSomeErrorAsTest()
{
    try
    {
        throw new Exception("Serer error");
    }
    catch(Exception ex)
    {
        return StatusCode(StatusCodes.Status500InternalServerError, new List<string>());
        //throw ex;
    }
}

角度服务：

export class MyService
{
    MyDataSubject = new Subject<any[]>();
    MyDataChanged :Observable>any[]> = this.MyDataSubject.asObservable();
    
    subscribe :Subscription;
    constructor(private httpClient : HttpClient)
    {
        this.subscribe = timer(0, 30000).pipe(
        switchMap(()=>
            this.getData())).subscribe();
    }
    getData()
    {
        return this.httpClient.get<any[]>(<controller url>)
        .pipe(
            tap(res =>
            {
                this.MyDataSubject.next(res);
            }),
            catchError(error =>
                {
                    debugger;//I would expect to catch the debugger here, but nothing happens
                    return throwError(error);
                })
            )
    }
}   

组件：

export class MyComponent (private mySrv : MyService)
{
    getMyData()
    {
        let sub =this.mySrv.MyDataChanged.subscribe(result => doSomething(),
                                                    error=> popUpAlert());
    }
}

Answer 1

正如@fridoo 指出的那样，您的组件正在订阅一个永远不会发出任何错误的可观察流。这是因为你有两个独立的 observables。

您实际上不需要维护主题并手动发出。您可以直接从getData 方法实现轮询和公开可观察的更改。

您甚至可以考虑将getData() 设为私有，只允许消费者使用MyDataChanged 可观察对象。 （也许重命名为 data$）

export class MyService {
    data$: Observable<any[]> = timer(0, 30000).pipe(
        switchMapTo(this.getData()),
        distinctUntilChanged(compareFn)
    );
    
    constructor(private httpClient: HttpClient) { }

    private getData() {
        return this.httpClient.get(<controller url>).pipe(
            retry(2)
        );
    }

    compareFn(oldVal: any[], newVal: any[]) {
        // add logic to determine if emission
        // is considered to be different.
    }
}   

请注意，您不必订阅您的服务。你也不需要有一个单独的主题。这还有一个懒惰的好处，因为在实际订阅了 data$ observable 之前，http 调用不会开始轮询。

您可以选择在 getData() 方法或组件中捕获错误。

使用的运算符：

1. switchMapTo - 由于您没有使用从计时器传递的参数，因此您可以使用这种更简洁的方法来代替常规的switchMap

2. distinctUntilChanged - 将过滤掉相同的连续排放。您提供一个“比较函数”来确定一个值是否被认为与以前的不同。

3. retry - 如果发生错误，可以方便地重试 http 调用。

现在，在您的组件中：

export class MyComponent {

    constructor(private mySrv: MyService){ }

    getMyData() {
        const sub = this.mySrv.data$.subscribe(
            result => doSomething(),
            error => popUpAlert()
        );
    }
}

如果我没有提到在模板中利用 async 管道而不是在控制器中管理订阅通常更简单，那我就失职了。

这种情况看起来像：

export class MyComponent {

    constructor(private mySrv: MyService){ }

    data$ = this.mySrv.data$.pipe(
        // tap(data => doSomething(data)), <-- hopefully you don't even need this
        catchError(error => { 
            popUpAlert(error);
        })
    );
}

然后是模板：

<ul>
    <li *ngFor="let item of data$ | async> {{ item.property }}</li>
</ul>