XStream

Collect ReadableStream\<Uint8Array> into asynchronously iterable records. If transformStream is not passed, split according to SSE: events are separated by blank lines and : separates field names and values; all three separators are configurable. Leaving the async loop early cancels the underlying stream and releases the reader lock.

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

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

If readableStream is not a ReadableStream, throw The options.readableStream must be an instance of ReadableStream.

Default protocol — SSE

Default protocol — SSE

Assemble Hello, world! from in-memory SSE chunks with no network request.

<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>

Hide source

Custom protocol

Custom protocol

Pass transformStream to read in chunks according to a custom protocol (SIP/SDP text).

<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>

Hide source

API

XStreamOptions

Prop	Description	Type	Default	Version
readableStream	Binary readable stream	ReadableStream<Uint8Array>	-	-
transformStream	Custom TransformStream; if not passed, the default SSE transform is used	TransformStream<string, T>	sseTransformStream	-
streamSeparator	Stream event separator; ignored when transformStream is set	string	\n\n	2.2.0
partSeparator	Field separator inside one event; ignored when transformStream is set	string	\n	2.2.0
kvSeparator	Separator between field names and values; ignored when transformStream is set	string	:	2.2.0

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>;

The 1.x page is still at XStream Stream. The usage in 2.x is the same, implemented from @ant-design-x-vue/x-sdk.