在可通过mergeMap传递的RxJ中继续发生错误

我正在使用RxJs管道和mergeMap运算符进行一些并行HTTP获取。

在第一个请求失败时（假设/ urlnotexists抛出404错误），它将停止所有其他请求。

我希望它继续查询所有剩余的URL，而不必为此失败的请求调用所有剩余的mergeMap。

我曾尝试将throwError和RxJs的catchError一起玩，但没有成功。

index.js

const { from } = require('rxjs');
const { mergeMap, scan } = require('rxjs/operators');

const request = {
  get: url => {
    return new Promise((resolve, reject) => {
      setTimeout(() => {
        if (url === '/urlnotexists') { return reject(new Error(url)); }
        return resolve(url);
      }, 1000);
    });
  }
};

(async function() {
  await from([
    '/urlexists',
    '/urlnotexists',
    '/urlexists2',
    '/urlexists3',
  ])
    .pipe(
      mergeMap(async url => {
        try {
          console.log('mergeMap 1:', url);
          const val = await request.get(url);
          return val;
        } catch(err) {
          console.log('err:', err.message);
          // a throw here prevent all remaining request.get() to be tried
        }
      }),
      mergeMap(async val => {
        // should not pass here if previous request.get() failed 
        console.log('mergeMap 2:', val);
        return val;
      }),
      scan((acc, val) => {
        // should not pass here if previous request.get() failed 
        acc.push(val);
        return acc;
      }, []),
    )
    .toPromise()
    .then(merged => {
      // should have merged /urlexists, /urlexists2 and /urlexists3
      // even if /urlnotexists failed
      console.log('merged:', merged);
    })
    .catch(err => {
      console.log('catched err:', err);
    });
})();

$ node index.js
mergeMap 1: /urlexists
mergeMap 1: /urlnotexists
mergeMap 1: /urlexists2
mergeMap 1: /urlexists3
err: /urlnotexists
mergeMap 2: /urlexists
mergeMap 2: undefined <- I didn't wanted this mergeMap to have been called
mergeMap 2: /urlexists2
mergeMap 2: /urlexists3
merged: [ '/urlexists', undefined, '/urlexists2', '/urlexists3' ]

我希望发出并发的GET请求，并最后在一个对象中减小它们各自的值。

但是如果发生某些错误，我希望他们不要中断我的管道，而要记录它们。

任何建议？

如果您愿意放弃RXJS并仅使用async / await解决，则非常简单：

const urls = ['/urlexists', '/urlnotexists', '/urlexists2', '/urlexists3']
const promises = urls.map(url => request(url)
const resolved = await Promise.allSettled(promises)

// print out errors
resolved.forEach((r, i) => {
  if (r.status === "rejected') {
    console.log(`${urls[i]} failed: ${r.reason})
  }
})

// get the success results
const merged = resolved.filter(r => r.status === "resolved").map(r => r.value)
console.log('merged', merged)

此方法使用Promise.allSettled提出的辅助方法。如果您的环境没有此方法，则可以按照this answer中所示的方法来实现。

在可通过mergeMap传递的RxJ中继续发生错误

我正在使用RxJs管道和mergeMap运算符进行一些并行HTTP获取。

在第一个请求失败时（假设/ urlnotexists抛出404错误），它将停止所有其他请求。

我希望它继续查询所有剩余的URL，而不必为此失败的请求调用所有剩余的mergeMap。

我曾尝试将throwError和RxJs的catchError一起玩，但没有成功。

index.js

const { from } = require('rxjs');
const { mergeMap, scan } = require('rxjs/operators');

const request = {
  get: url => {
    return new Promise((resolve, reject) => {
      setTimeout(() => {
        if (url === '/urlnotexists') { return reject(new Error(url)); }
        return resolve(url);
      }, 1000);
    });
  }
};

(async function() {
  await from([
    '/urlexists',
    '/urlnotexists',
    '/urlexists2',
    '/urlexists3',
  ])
    .pipe(
      mergeMap(async url => {
        try {
          console.log('mergeMap 1:', url);
          const val = await request.get(url);
          return val;
        } catch(err) {
          console.log('err:', err.message);
          // a throw here prevent all remaining request.get() to be tried
        }
      }),
      mergeMap(async val => {
        // should not pass here if previous request.get() failed 
        console.log('mergeMap 2:', val);
        return val;
      }),
      scan((acc, val) => {
        // should not pass here if previous request.get() failed 
        acc.push(val);
        return acc;
      }, []),
    )
    .toPromise()
    .then(merged => {
      // should have merged /urlexists, /urlexists2 and /urlexists3
      // even if /urlnotexists failed
      console.log('merged:', merged);
    })
    .catch(err => {
      console.log('catched err:', err);
    });
})();

$ node index.js
mergeMap 1: /urlexists
mergeMap 1: /urlnotexists
mergeMap 1: /urlexists2
mergeMap 1: /urlexists3
err: /urlnotexists
mergeMap 2: /urlexists
mergeMap 2: undefined <- I didn't wanted this mergeMap to have been called
mergeMap 2: /urlexists2
mergeMap 2: /urlexists3
merged: [ '/urlexists', undefined, '/urlexists2', '/urlexists3' ]

我希望发出并发的GET请求，并最后在一个对象中减小它们各自的值。

但是如果发生某些错误，我希望他们不要中断我的管道，而要记录它们。

任何建议？

如果您愿意放弃RXJS并仅使用async / await解决，则非常简单：

const urls = ['/urlexists', '/urlnotexists', '/urlexists2', '/urlexists3']
const promises = urls.map(url => request(url)
const resolved = await Promise.allSettled(promises)

// print out errors
resolved.forEach((r, i) => {
  if (r.status === "rejected') {
    console.log(`${urls[i]} failed: ${r.reason})
  }
})

// get the success results
const merged = resolved.filter(r => r.status === "resolved").map(r => r.value)
console.log('merged', merged)

此方法使用Promise.allSettled提出的辅助方法。如果您的环境没有此方法，则可以按照this answer中所示的方法来实现。