XStream 流

转换可读数据流。

何时使用

• 将 SSE 协议的 ReadableStream 转换为 Record
• 将任何协议的 ReadableStream 解码并读取

使用说明

常见的 ReadableStream 实例，如 await fetch(...).body 使用示例:

import { XStream } from '@ant-design/x';

async function request() {
  const response = await fetch();
  // .....

  for await (const chunk of XStream({
    readableStream: response.body,
  })) {
    console.log(chunk);
  }
}

代码演示

默认协议 - SSE

SSE - https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events

XStream 默认的 transformStream 是用于 SSE 协议的流转换器。readableStream 接收一个 new ReadableStream(...) 实例，常见的如 await fetch(...).body

<script setup lang="tsx">
import { TagsOutlined } from '@ant-design/icons-vue';
import { Button, Flex } from 'ant-design-vue';
import { Bubble, ThoughtChain, XStream } from 'ant-design-x-vue';
import { computed, ref } from 'vue';

defineOptions({ name: 'AXXStreamDefaultProtocol' });

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

function mockReadableStream() {
  const sseChunks: string[] = [];

  for (let i = 0; i < contentChunks.length; i++) {
    const sseEventPart = `event: message\ndata: {"id":"${i}","content":"${contentChunks[i]}"}\n\n`;
    sseChunks.push(sseEventPart);
  }

  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() {
  // 🌟 Read the stream
  for await (const chunk of XStream({
    readableStream: mockReadableStream(),
  })) {
    console.log(chunk);
    lines.value = [...lines.value, chunk];
  }
}

defineRender(() => {
  return (
    <Flex gap={8}>
      <div>
        {/* -------------- Emit -------------- */}
        <Button type="primary" onClick={readStream} style={{ marginBottom: '16px' }}>
          Mock Default Protocol - SSE
        </Button>
        {/* -------------- Content Concat -------------- */}
        {content.value && <Bubble content={content.value} />}
      </div>
      <div>
        <ThoughtChain
          items={
            lines.value.length
              ? [
                {
                  title: 'Mock Default Protocol - Log',
                  status: 'success',
                  icon: <TagsOutlined />,
                  content: (
                    <pre style={{ overflow: 'scroll' }}>
                      {lines.value.map((i) => (
                        <code key={i.data}>{i.data}</code>
                      ))}
                    </pre>
                  ),
                },
              ]
              : []
          }
        />
      </div>
    </Flex>
  )
});
</script>

<style scoped>
pre {
  width: 'auto';
  margin: 0;

  code {
    display: block;
    padding: 12px 16px;
    font-size: 14px;
  }
}
</style>

隐藏源代码

自定义协议

在本示例中，我们将演示如何解析 SIP 协议, 该协议常用于 P2P 音视频会话协商。

传入 transformStream 流转换器，该参数需接收一个 new TransformStream(...) 实例。

<script setup lang="tsx">
import { TagsOutlined } from '@ant-design/icons-vue';
import { Button, Flex } from 'ant-design-vue';
import { ThoughtChain, XStream } from 'ant-design-x-vue';
import { ref } from 'vue';

defineOptions({ name: 'AXXStreamCustomProtocol' });

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() {
  // 🌟 Read the stream
  for await (const chunk of XStream({
    readableStream: mockReadableStream(),
    transformStream: new TransformStream<string, string>({
      transform(chunk, controller) {
        controller.enqueue(chunk);
      },
    }),
  })) {
    lines.value = [...lines.value, chunk];
  }
}

defineRender(() => {
  return (
    <Flex gap={8}>
      <div>
        <Button type="primary" onClick={readStream} style={{ marginBottom: 16 }}>
          Mock Custom Protocol - SIP
        </Button>
      </div>
      {/* -------------- Log -------------- */}
      <div>
        <ThoughtChain
          items={
            lines.value.length
              ? [
                {
                  title: 'Mock Custom Protocol - Log',
                  status: 'success',
                  icon: <TagsOutlined />,
                  content: (
                    <pre style={{ overflow: 'scroll' }}>
                      <code>{lines.value.join('')}</code>
                    </pre>
                  ),
                },
              ]
              : []
          }
        />
      </div>
    </Flex>
  )
});
</script>

隐藏源代码

API

XStreamOptions

属性	说明	类型	默认值	版本
readableStream	ReadableStream 实例	ReadableStream<'Uint8Array'>	-	-
transformStream	自定义的 transformStream 用于转换流的处理	TransformStream<string, T>	sseTransformStream	-