RxJS实现服务端轮询触发Object(...) is not a function错误如何解决？

核心错误原因 #

• RxJS的pipe方法要求传入操作符函数作为参数，你直接将from(fetch(url))生成的Observable实例传入timer.pipe()，框架尝试把Observable当函数调用，直接触发is not a function报错。
• 你没有对timer生成的轮询流做subscribe订阅，RxJS的流是冷流，不订阅永远不会执行。
• 没有处理后端返回的204无内容响应，直接调用response.json()会在204场景下抛出解析错误。
• 没有处理请求超时/慢响应的情况，若接口响应耗时超过500ms会出现请求堆积。
• 后端204响应逻辑错误，204状态码不允许携带响应体，你当前res.status(204).send({})的写法不符合HTTP规范。

修复后的代码 #

前端轮询代码 #

import { from, of, timer } from 'rxjs'
import { switchMap, map, tap } from 'rxjs/operators'
import { FRONT_PAGE } from './constants'

const poll_server = url => {
  // 返回Observable供外部订阅
  return timer(0, 500).pipe(
    // 用switchMap把定时事件转成请求，自动取消上一次未完成的请求
    switchMap(() => from(fetch(url))),
    // 打日志用tap操作符，不要自己写没用的pipe回调
    tap(() => console.log("Polling server..")),
    map(async response => {
      // 处理204空响应
      if (!response.ok) throw new Error('请求失败')
      if (response.status === 204) return {}
      return await response.json()
    })
  )
}

export const server_dispatch = action => {
  switch (action.type) {
    case FRONT_PAGE: {
      // 订阅流才能触发轮询执行，记得组件销毁时调用unsubscribe取消订阅避免内存泄漏
      poll_server('http://localhost:8080/warnings').subscribe({
        next: data => {
          // 这里处理返回的业务数据，更新你的状态即可
          console.log('轮询拿到数据：', data)
        },
        error: err => console.error('轮询出错：', err)
      })
      return of(action)
    }
    default:
      return of(action)
  }
}

后端接口修复 #

app.get('/warnings', (req, res) => {
  console.log("[GET] /warnings")
  const baseline = req.query.baseline ?? -1
  if (version > baseline) {
    res.send(warnings(alerts.filter(a => a.prediction)))
  } else if (game) {
    // 204直接用sendStatus，不要传响应体
    res.sendStatus(204)
  } else {
    res.sendStatus(404)
  }
})

额外注意点 #

• 轮询订阅后要在页面/组件销毁时调用unsubscribe取消订阅，避免内存泄漏和不必要的后台请求。
• 如果需要携带baseline参数，可在轮询逻辑里维护baseline变量，每次请求带上上一次返回的版本号，减少不必要的全量数据传输。

内容的提问来源于stack exchange，提问作者kakakakakakakk