XStream

把 ReadableStream<Uint8Array> 收成可异步迭代的记录。不传 transformStream 时按 SSE 拆：空行分隔事件，: 分隔字段名和值。

import { XStream } from 'ant-design-x-vue';

for await (const chunk of XStream({ readableStream })) {
  console.log(chunk);
}

readableStream 不是 ReadableStream 时抛出 The options.readableStream must be an instance of ReadableStream.

默认协议 - SSE

默认协议 - SSE

用内存里的 SSE 片段拼出 Hello, world!，不发网络请求。

<script setup lang="tsx">
defineOptions({ name: 'AXSdkXStreamDefaultProtocolV2' });

import { computed, ref } from 'vue';
import { TagsOutlined } from '@ant-design/icons-vue';
import { Button, Flex } from 'ant-design-vue';
import { Bubble, ThoughtChain, XStream } from 'ant-design-x-vue';

const contentChunks = ['He', 'llo', ', w', 'or', 'ld!'];

function mockReadableStream() {
  const sseChunks = contentChunks.map(
    (chunk, i) => `event: message\ndata: {"id":"${i}","content":"${chunk}"}\n\n`,
  );
  return new ReadableStream({
    async start(controller) {
      for (const chunk of sseChunks) {
        await new Promise((resolve) => setTimeout(resolve, 100));
        controller.enqueue(new TextEncoder().encode(chunk));
      }
      controller.close();
    },
  });
}

const lines = ref<Record<string, string>[]>([]);
const content = computed(() =>
  lines.value.map((line) => JSON.parse(line.data).content).join(''),
);

async function readStream() {
  lines.value = [];
  for await (const chunk of XStream({ readableStream: mockReadableStream() })) {
    lines.value = [...lines.value, chunk as Record<string, string>];
  }
}

defineRender(() => (
  <Flex vertical gap={16}>
    <Button type="primary" style={{ alignSelf: 'flex-start', width: 'fit-content' }} onClick={readStream}>
      Mock Default Protocol - SSE
    </Button>
    {content.value ? <Bubble content={content.value} /> : null}
    <ThoughtChain
      items={
        lines.value.length
          ? [
              {
                key: 'log',
                title: 'Mock Default Protocol - Log',
                status: 'success',
                icon: <TagsOutlined />,
                content: (
                  <pre style={{ overflow: 'auto', margin: 0 }}>
                    {lines.value.map((i) => i.data).join('\n')}
                  </pre>
                ),
              },
            ]
          : []
      }
    />
  </Flex>
));
</script>

隐藏源代码

自定义协议

自定义协议

传入 transformStream，按自定义协议（SIP/SDP 文本）逐块读取。

<script setup lang="tsx">
defineOptions({ name: 'AXSdkXStreamCustomProtocolV2' });

import { ref } from 'vue';
import { TagsOutlined } from '@ant-design/icons-vue';
import { Button, Flex } from 'ant-design-vue';
import { ThoughtChain, XStream } from 'ant-design-x-vue';

const sipHeaders = [
  'INVITE sip:[email protected] SIP/2.0\n',
  'Via: SIP/2.0/UDP [host];branch=123456\n',
  'Content-Type: application/sdp\n\n',
];

const sdp = [
  'v=0\n',
  'o=alice 2890844526 2890844526 IN IP4 [host]\n',
  's=\n',
  'c=IN IP4 [host]\n',
  't=0 0\n',
  'm=audio 49170 RTP/AVP 0\n',
  'a=rtpmap:0 PCMU/8000\n',
  'm=video 51372 RTP/AVP 31\n',
  'a=rtpmap:31 H261/90000\n',
  'm=video 53000 RTP/AVP 32\n',
  'a=rtpmap:32 MPV/90000\n\n',
];

function mockReadableStream() {
  return new ReadableStream({
    async start(controller) {
      for (const chunk of sipHeaders.concat(sdp)) {
        await new Promise((resolve) => setTimeout(resolve, 100));
        controller.enqueue(new TextEncoder().encode(chunk));
      }
      controller.close();
    },
  });
}

const lines = ref<string[]>([]);

async function readStream() {
  lines.value = [];
  for await (const chunk of XStream({
    readableStream: mockReadableStream(),
    transformStream: new TransformStream<string, string>({
      transform(chunk, controller) {
        controller.enqueue(chunk);
      },
    }),
  })) {
    lines.value = [...lines.value, chunk as string];
  }
}

defineRender(() => (
  <Flex vertical gap={16}>
    <Button type="primary" style={{ alignSelf: 'flex-start', width: 'fit-content' }} onClick={readStream}>
      Mock Custom Protocol - SIP
    </Button>
    <ThoughtChain
      items={
        lines.value.length
          ? [
              {
                key: 'log',
                title: 'Mock Custom Protocol - Log',
                status: 'success',
                icon: <TagsOutlined />,
                content: (
                  <pre style={{ overflow: 'auto', margin: 0 }}>
                    <code>{lines.value.join('')}</code>
                  </pre>
                ),
              },
            ]
          : []
      }
    />
  </Flex>
));
</script>

隐藏源代码

API

XStreamOptions

属性	说明	类型	默认值	版本
readableStream	二进制可读流	ReadableStream<Uint8Array>	-	-
transformStream	自定义 TransformStream；不传时使用默认 SSE 转换	TransformStream<string, T>	sseTransformStream	-

type SSEFields = 'data' | 'event' | 'id' | 'retry';
type SSEOutput = Partial<Record<SSEFields, any>>;

type XReadableStream<R = SSEOutput> = ReadableStream<R> & AsyncGenerator<R>;

function XStream<Output = SSEOutput>(
  options: XStreamOptions<Output>,
): XReadableStream<Output>;

1.x 页面仍在 XStream 流。2.0 的调用形式相同，实现来自 @ant-design-x-vue/x-sdk。