RxJS HTTP轮询拉取例子

页面上经常需要每几秒钟轮询服务器接口更新页面状态，这种场景多发生在一些监控页面里，如果不用RxJS可能我们这写轮询代码

【国内直连ChatGPT 29元起】
国内直连ChatGPT，Plus会员每月29元起，支持最新o1模型探索更多领域，无需注册OpenAI账号。

const do_request = () => {
  fetch('https://www.qttc.net/api')
    .then((r) => {
      console.log(r);
    });
};

setInterval(do_request, 2000);

以上代码可能会产生一个问题，如果服务器响应比较慢，在上一个请求还未响应之前，下一个请求就已经发出，可能造成的结果就是在请求数不断增加，这不仅没有意义，还会造成性能问题，所以通常这么优化

const request_interval_sec = 2000; 

const do_request = () => {
  fetch('https://www.qttc.net/api')
    .then((r) => {
      handle_response(r);
      setTimeout(do_request, request_interval_sec);
    })
    .catch((err) => {
      handle_error(err);
      
      setTimeout(do_request, request_interval_sec);
    })
};

setTimeout(do_request, request_interval_sec);

以上代码必须等待上次的请求响应之后才会间隔2秒发出下次请求，但设计上看着感觉很不舒服，用RxJS可以这么写

import { from, timer } from 'rxjs';
import { map, exhaustMap, tap } from 'rxjs/operators';
import axios from 'axios';

const wait_for_sec = 2;

timer(0, wait_for_sec * 1000)
  .pipe(
    exhaustMap(() => from(axios
      .get('https://www.qttc.net/api')
      .then(r => r.data)
      .catch(r => r.response))
    ),
    map(r => r.status),
    tap((status_code) => {
      console.log(`status code: ${status_code}`);
    }),
  )
  .subscribe({
    error(err) {
      handle_error(err);
    },
    complete() { console.log('done'); }
  });

如果需要某个条件达到时停止，可以配合takeWhile使用，比如响应状态码为200就停止轮询

import { from, timer } from 'rxjs';
import { map, exhaustMap, tap, takeWhile } from 'rxjs/operators';
import axios from 'axios';

const wait_for_sec = 2;

timer(0, wait_for_sec * 1000)
  .pipe(
    exhaustMap(() => from(axios
      .get('https://www.qttc.net/api')
      .then(r => r.data)
      .catch(r => r.response))
    ),
    map(r => r.status),
    tap((status_code) => {
      if (status_code === 200) {
        console.log('\x1b[36m%s\x1b[0m', `status code: ${status_code}, successfully`);
      } else {
        console.log('\x1b[33m%s\x1b[0m', `status code: ${status_code}, wait for retry in ${wait_for_sec} sec ...`);
      }
    }),
    takeWhile(status_code => status_code !== 200)
  )
  .subscribe({
    error(err) {
      handle_error(err);
    },
    complete() { console.log('done'); }
  });

以上代码会一直轮询，直到服务器响应200状态码之后停止轮询，配合RxJS丰富的操作函数集，可以在pipe里进行很多有意思的玩法。