Rxjs如何使用mergeMap？

Question

任务

我有 3 个可观察的 http1$ http2$ http3$ 是 http 请求。

const http1$ = this.myService.someFnReturnAnObservale({});

我有另一个可观察的isAuth$ 作为我的流源。 我需要得到http3$的回复。

上下文

为了使http3$，http2$必须先响应一些东西。 取决于isAuth$ 的值，如果它发出true，那么http2$ 也必须等待http1$ 响应，如果它发出false，则不需要http1$。

我的意思是成功或错误，所以我可以对其应用一些逻辑，并决定是否流式传输到下一个 observable。

编码

• 创建一个 observable 来处理从 http2$ 到 http3$ 的逻辑

const http23$ = http2$.pipe(
    first(),
    map(res => {
       // some logic, return boolean flag
    }),
    catchError(err => {
        return of(false); // in case http2$ response with error, also return false
    }),
    map(res => {
        // if response (which is the flag above) is false
        // pop an message
        return res;
    }),
    filter(res => res === true), // only subscribe to http3$ after this
    mergeMap(res => http3$),
);

• isAuth$ 有 2 个逻辑，所以：

const initWithoutAuth$ = of(http23$); // no need http1$
const initWithAuth$ = http1$.pipe(
    first(),
    map(res => {
        // some logic, return boolean flag
    }),
    catchError(err => {
        return of(false); // in case http1$ response with error, also return false
    }),
    map(res => {
        // if response (which is the flag above) is false
        // pop an message

        return res;
    }),
    filter(res => res === true), // only subscribe to http23$ after this
    mergeMap(res => http23$)
);

• 每当isAuth$ 发出新值时，我想取消当前流并开始使用另一个流

isAuth$.pipe(
    switchMap(res => // does this really cancel curernt stream (http requests) to start over within the other one?
        res ? initWithAuth$ : initWithoutAuth$
    )
).subscribe(
    res => {
        console.log(`init`, res); // not the desired result yet
    },
    err => {}
);

• 当我在最后一步订阅时，我希望 http3$ 有响应，但我得到了一个奇怪的 observable。
• 在这种情况下如何正确处理 http observable 响应错误？ （每当我尝试处理响应并捕获前一个错误时，都会出现重复的代码）

Answer 1

我认为错误的是这一行

const initWithoutAuth$ = of(http23$);

通过这一行，您基本上说initWithoutAuth$ 是一个通知（发出）另一个 Observable 的 Observable，即它通知 http23$。

在我看来 initWithoutAuth$ 应该简单地设置为 http23$ 而不是 of(http23$)。

如果你这样做，你也可以从initWithAuth$和initWithoutAuth$中删除catchError，并将其添加到最终的Observable链中，像这样

isAuth$.pipe(
    switchMap(res => // does this really cancel curernt stream (http requests) to start over within the other one?
        res ? initWithAuth$ : initWithoutAuth$
    ),
    catchError(err => {
        return of(false); // in case http1$ or http2$ response with error, also return false
    }),
).subscribe(
    res => {
        console.log(`init`, res); // not the desired result yet
    },
);

最后一点。我认为您在每次 http 调用后使用的 first 运算符不是必需的。 http 客户端返回一个 Observable，它最多只通知 1 个值然后完成，因此 first 运算符是多余的。