Rxjs如何使用mergeMap？

Question

任务

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

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

我有另一个 observable isAuth$ 作为我的流源。我需要得到 http3$.

的回复

上下文

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

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

编码

创建一个可观察对象来处理从 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$ 响应，但我得到了一个奇怪的可观察值。
在这种情况下，如何正确处理 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
    },
);

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

Rxjs如何使用mergeMap？

Rxjs how to use mergeMap?

typescript

rxjs

mapping

httprequest