流媒体与上传

Core 的 callApiStream 逐行读取 NDJSON 响应并在 Observable 中按顺序发出.

在 Core 中直接使用 callApiStream.

当服务器返回 NDJSON（每行一个 JSON）时使用 callApiStream。如果没有 Accept header，则将 application/x-ndjson 设置为默认值，其他媒体类型可以通过每个请求的 headers 覆盖.

export const streamEvents = () => {
  return client.callApiStream<{
    eventName: string;
    content: unknown;
  }>({
    url: "/api/event/stream",
    method: "get",
    // headers: { Accept: "application/x-ndjson" },
  });
};

NDJSON 响应格式.

text

{"status":"STARTED","progress":0}
{"status":"PROGRESS","progress":35}
{"status":"PROGRESS","progress":80}
{"status":"DONE","progress":100}

一次响应是firstValueFrom以换行符为基准分割响应并对每一行进行 JSON.parse。期待的是 NDJSON，而不是标准 SSE 的 data: frame 格式.

Observable 订阅和取消.

tsx

const subscription = streamEvents().subscribe({
  next: (event) => {
    setProgress(event.content);
  },
  error: (error) => {
    setErrorMessage(error.message);
  },
  complete: () => {
    setRunning(false);
  },
});

return () => subscription.unsubscribe();

调用 unsubscribe() 时，内部 ReadableStream reader 也会被取消。在 React effect 或任务清理中取消订阅.

可选：使用生成的 stream 服务.

在使用 OpenAPI generator 的项目中，可以使生成的服务调用 stream 端点的 callApiStream。在界面上以与普通生成服务相同的方式导入和订阅.

tsx

const subscription = jobProgress({
  path: { jobId },
}).subscribe({
  next: (event) => {
    setProgress(event.progress || 0);
  },
  error: (error) => {
    setErrorMessage(error.message);
  },
  complete: () => {
    setRunning(false);
  },
});

return () => subscription.unsubscribe();

uploadFile：将文件 PUT 到指定 URL.

uploadFile.ts

service.uploadFile({
  file,
  url: signedUploadUrl,
  headers: {
    "Content-Type": file.type || "application/octet-stream",
    "If-None-Match": "*",
  },
}).subscribe();

uploadFile header 操作.

uploadFile 内部有默认 header 合并代码，但实际的 fromFetch 调用使用传入的 headers。所需的 Content-Type、If-None-Match 等 header 需在调用处直接传递.