如何在Redux Toolkit中创建无初始数据拉取的流式端点

我需要在Redux Toolkit中使用流式连接，官方文档的示例通过初始拉取获取历史消息记录：

export const api = createApi({
  baseQuery: fetchBaseQuery({ baseUrl: '/' }),
  endpoints: (build) => ({
    getMessages: build.query<Message[], Channel>({
      // retrieve initial data
      query: (channel) => `messages/${channel}`,
      async onCacheEntryAdded(
        arg,
        { updateCachedData, cacheDataLoaded, cacheEntryRemoved }
      ) {
        // create a websocket connection when the cache subscription starts
        const ws = new WebSocket('ws://localhost:8080')
        try {
          // wait for the initial query to resolve before proceeding
          await cacheDataLoaded

          // when data is received from the socket connection to the server,
          // if it is a message and for the appropriate channel,
          // update our query result with the received message
          const listener = (event: MessageEvent) => {
            const data = JSON.parse(event.data)
            if (!isMessage(data) || data.channel !== arg) return

            updateCachedData((draft) => {
              draft.push(data)
            })
          }

          ws.addEventListener('message', listener)
        } catch {
          // no-op in case `cacheEntryRemoved` resolves before `cacheDataLoaded`,
          // in which case `cacheDataLoaded` will throw
        }
        // cacheEntryRemoved will resolve when the cache subscription is no longer active
        await cacheEntryRemoved
        // perform cleanup steps once the `cacheEntryRemoved` promise resolves
        ws.close()
      },
    }),
  }),
})

export const { useGetMessagesQuery } = api

但我的端点仅支持流式更新，无初始数据响应，请问如何实现无需初始拉取的流式端点？

我已尝试的方案：

• 移除query属性
• 在query的responseHandler中返回undefined

核心思路 #

RTK Query的build.query要求必须配置query字段，不能直接移除。我们可以通过配置一个"占位符"query返回空初始数据，同时调整onCacheEntryAdded逻辑，跳过对初始请求结果的等待，直接建立流式连接。

完整实现代码 #

export const api = createApi({
  baseQuery: fetchBaseQuery({ baseUrl: '/' }),
  endpoints: (build) => ({
    getMessages: build.query<Message[], Channel>({
      // 配置占位符query，返回空数组作为初始数据
      query: () => ({
        url: '/dummy-stream-endpoint',
        // 直接返回空数组，避免发起实际无效请求
        responseHandler: () => []
      }),
      async onCacheEntryAdded(
        arg,
        { updateCachedData, cacheEntryRemoved }
      ) {
        // 初始化缓存为空数组（确保组件订阅时拿到正确初始状态）
        updateCachedData(() => [])
        
        // 建立WebSocket连接
        const ws = new WebSocket('ws://localhost:8080')

        // 监听流式消息并更新缓存
        const messageListener = (event: MessageEvent) => {
          const data = JSON.parse(event.data)
          if (!isMessage(data) || data.channel !== arg) return

          updateCachedData((draft) => {
            draft.push(data)
          })
        }

        ws.addEventListener('message', messageListener)

        // 缓存订阅结束时清理WebSocket连接
        await cacheEntryRemoved
        ws.close()
      },
    }),
  }),
})

export const { useGetMessagesQuery } = api

关键细节说明 #

1. 保留query字段：build.query强制要求存在query配置，直接移除会导致报错。用占位符query返回空数组，既满足语法要求，又不会产生无效请求。
2. 移除cacheDataLoaded等待：因为没有初始请求，cacheDataLoaded Promise永远不会resolve（或抛出错误），所以直接跳过该步骤，直接建立流式连接。
3. 初始化缓存：在onCacheEntryAdded开头调用updateCachedData(() => [])，确保组件订阅该query时，初始状态是符合类型定义的空数组，避免渲染错误。
4. 资源清理：通过cacheEntryRemoved监听订阅结束事件，及时关闭WebSocket，避免内存泄漏。

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