Xrequest自定义配置的时候无法关闭streams的锁

[mhfe123]

重现步骤

// 这是使用fetch的方案
const [provider] = React.useState(
      new CustomProvider<CustomMessage, CustomInput, CustomOutput>({
        request: XRequest<CustomInput, CustomOutput>(
          url,
          {
            manual: true,
            headers: {
              Accept: "text/event-stream",
              "access-token": localStorage.getItem("access-token") || "",
            },
            callbacks: {
              onSuccess: () => {
                setIsSending(false);
                console.log("onSuccess");
              },
              onError: (error) => {
                setIsSending(false);
                console.error("onError", error);
              },
              onUpdate: () => {
                console.log("onUpdate");
              },
            },
            // 自定义流转换器
            transformStream: new TransformStream<string, CustomOutput>({
              transform(chunk, controller) {
                console.log("transformStream", chunk);
                controller.enqueue({ data: chunk });
              },
            }),
            fetch: async (url, options) => {
              const response = await fetch(url, options);
              if (!response.body) {
                throw new Error("Response body is null");
              }
              const sseChunks: string[] = [];
              const rawStream = response.body;
              const textStream = rawStream.pipeThrough(
                new TextDecoderStream("utf-8")
              );
              const lineStream = textStream.pipeThrough(
                new TransformStream({
                  transform(chunk, controller) {
                    // SSE流按换行符分割，保证每次拿到完整的一行SSE数据
                    chunk
                      .split("\n")
                      .forEach((line) => controller.enqueue(line));
                  },
                })
              );

              // ✅ 最终：读取管道处理后的「新流」，无锁冲突！
              const reader = lineStream.getReader();
              const reSSELine = /^(data|event|retry|id)(\s*)/; // 匹配SSE合法字段名
              try {
                while (true) {
                  const { done, value } = await reader.read();
                  if (done) break;

                  let sseLine = value.trim();
                  // ✅ 核心兼容逻辑：修复SSE行格式，补上缺失的冒号
                  if (sseLine && !sseLine.includes(":")) {
                    // 情况1：有字段名但无冒号 → 补上冒号
                    if (reSSELine.test(sseLine)) {
                      sseLine = sseLine.replace(reSSELine, "$1:");
                    }
                    // 情况2：纯文本无字段名 → 补上 data: 前缀+冒号
                    else {
                      sseLine = `data: ${sseLine}`;
                    }
                  }
                  // ✅ 情况3：替换中文全角冒号为英文半角冒号
                  sseLine = sseLine.replace(/：/g, ":");

                  // 解析业务数据
                  if (sseLine.startsWith("data:")) {
                    const jsonStr = sseLine.slice(5).trim();
                    sseChunks.push(jsonStr);
                    // const data = JSON.parse(jsonStr);
                    // console.log("解析后的业务数据：", data);
                  }
                }
              } catch (err) {
                console.error("SSE读取异常：", err);
                setIsSending(false);
                if (reader) reader.releaseLock();
              } finally {
                setIsSending(false);
                console.log("SSE读取完毕");
                if (reader && !reader.closed) reader.releaseLock();
              }
              return new Response(
                new ReadableStream({
                  start(controller) {
                    for (const chunk of sseChunks) {
                      controller.enqueue(new TextEncoder().encode(chunk));
                    }
                    controller.close();
                  },
                }),
                {
                  headers: {
                    "Content-Type": "text/event-stream",
                  },
                }
              );
            },
          }
        ),
      })
    );

// 这是没有用fetch的方案
new CustomProvider<CustomMessage, CustomInput, CustomOutput>({
      request: XRequest<CustomInput, CustomOutput>(
         url,
        {
          manual: true,
          headers: {
            Accept: "text/event-stream",
            "access-token": localStorage.getItem("access-token") || "",
          },
          callbacks: {
            onSuccess: () => {
              console.log("onSuccess");
            },
            onError: (error) => {
              console.error("onError", error);
            },
            onUpdate: () => {
              console.log("onUpdate");
            },
          },
          // 自定义流转换器：解析完整SSE格式数据，兼容Windows和Unix换行符
          transformStream: new TransformStream<string, CustomOutput>({
            transform(chunk, controller) {
              // 处理SSE事件：event: metadata\ndata: {...}\nid: 0\n\n
              // 统一转换Windows换行符(\r\n)为Unix换行符(\n)，并移除单独的\r字符
              const normalizedChunk = chunk
                .replace(/\r\n/g, "\n")
                .replace(/\r/g, "");

              // 分割成独立的SSE事件（每个事件以\n\n结束）
              const events = normalizedChunk
                .split("\n\n")
                .filter((event) => event.trim());

              for (const event of events) {
                // 提取data字段内容
                const dataLines = event
                  .split("\n")
                  .filter((line) => line.startsWith("data: "))
                  .map((line) => line.replace("data: ", ""))
                  .join("");

                if (dataLines) {
                  try {
                    // 解析JSON数据
                    const jsonData = JSON.parse(dataLines);
                    // 转换为CustomOutput格式
                    controller.enqueue({ data: JSON.stringify(jsonData) });
                  } catch (error) {
                    // 如果解析失败，直接使用原始数据
                    controller.enqueue({ data: dataLines });
                  }
                } else {
                  console.log("No data field in event:", event);
                  // 如果没有data字段，则直接使用原始数据
                  controller.enqueue({ data: event });
                }
              }
            },
          }),
        }
      ),
    });

当前行为

我在使用Xrequest做自定义的封装的时候，因为我当前使用的后端数据可能会有点不标准返回内容，需要我做一个手动处理，我在Xrequest中只使用transformStream的时候会造成第二次输入请求的时候流被锁定的问题，如果配合fetch使用的话又会导致它要等接口请求完成以后才会进行输出，有没有什么好的解决方案
报错信息：onError TypeError: Failed to execute 'getReader' on 'ReadableStream': ReadableStreamDefaultReader constructor can only accept readable streams that are not yet locked to a reader

预期行为

因为我看demo中只有单独Xrequest的场景，没有实际Provider中的场景，不是很清楚是我写法上有问题还是什么原因导致的

上下文

No response

版本

2.1.3

您在哪些浏览器上遇到了这个问题？

No response

[dosubot]

Hi @mhfe123! I'm Dosu and I’m helping the x team.

你的问题本质是 ReadableStream 只能被一个 reader 锁定，第二次请求如果复用了同一个 stream 或 transformStream 实例，就会报锁定错误（getReader 只能用于未锁定的流）参考。

在 Provider 场景下，常见原因是 Provider 或 XRequest 实例被 useState 持久化，导致流对象被复用。每次请求都应该创建新的 ReadableStream 和 TransformStream 实例，不能复用，否则会锁定参考。

关于自定义 fetch，如果你在 fetch 里把整个 response 读完再返回新流，会导致只有请求结束后才输出数据，无法实现流式增量输出。正确做法是直接把 response.body 作为流传递给 XRequest，并用 transformStream 按协议分割和处理数据（比如 NDJSON 按 \n 分割，SSE 按 \n\n 分割），这样 XRequest 会边读边触发 onUpdate，实现实时输出参考。

推荐做法：

• 每次请求都 new 一个 TransformStream 实例，不能复用。
• fetch 返回的 Response.body 直接传给 XRequest，不要提前读完。
• transformStream 里用 buffer 分割数据，保证每条消息单独 enqueue，onUpdate 能多次触发。
• Provider 里不要把流对象或 transformStream 持久化在 state，建议每次请求都 new。

相关 demo 可参考 custom-transformer 和 custom-protocol。

如果还有具体 Provider 场景代码，可以贴出来一起分析。

To reply, just mention @dosu.

---

^{How did I do? Good | Irrelevant | Incorrect | Verbose | Hallucination | Report 🐛 | Other}

   

[mhfe123]

现在问题是我如果把transformStream去掉就不会有锁定问题，然后加上去就会有锁定问题，看文档里也没有说明Provider 的这个持久化情况是咋回事，就在文档Custom Chat Provider自定义这个给的例子基础上加一个transformStream的配置就会出现锁定情况，我可以通过创建新的provide来解决这个问题，但是这样就感觉不太优雅，如果框架能自己提供这个能力就更好了

[dosubot]

你遇到的锁定问题，是因为 transformStream 实例被持久化（比如写在 Provider 或 useState 里），导致每次请求都复用同一个 TransformStream/ReadableStream。Web Streams API 规定，一个流只能被一个 reader 锁定，复用会报错 参考。

正确做法是：每次请求都创建新的 transformStream 实例。你可以把 transformStream 配置成一个工厂函数（每次调用返回新的 TransformStream），或者在请求发起时 new 一个新的 TransformStream。

比如：

const [provider] = React.useState(
  new CustomProvider({
    request: XRequest(url, {
      manual: true,
      // 推荐写法：transformStream 用函数返回新实例
      transformStream: () => new TransformStream({
        transform(chunk, controller) {
          // 你的自定义处理逻辑
          controller.enqueue({ data: chunk });
        },
      }),
      // 其他配置...
    }),
  })
);

或者在每次请求前 new：

const request = XRequest(url, {
  manual: true,
  transformStream: new TransformStream({ ... }), // 不要持久化在 Provider/useState
});

官方 demo 没有用 transformStream，所以没遇到这个问题，但实际业务场景下必须保证每次请求都是新流、新 transformStream，否则就会锁定 源码分析。

目前文档确实没有说明 Provider 持久化和 transformStream 的坑，建议后续补充。如果你流量很大，transformStream 里可以做批量/节流，避免 React 频繁 setState 导致性能问题，但这和锁定无关 相关讨论。

如需更复杂的流拆分，可以参考 custom-transformer demo。

核心原则：每次请求都 new，不要复用流对象或 transformStream。

To reply, just mention @dosu.

---

^{How did I do? Good | Irrelevant | Incorrect | Verbose | Hallucination | Report 🐛 | Other}

   

[mhfe123]

好的，谢谢

[kimteayon]

我把文档补齐