ArtStream 流

转换可读数据流。

何时使用

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

使用说明

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

import { ArtStream } from '@artmate/chat'

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

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

代码演示

默认协议 - SSE

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

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

<script setup lang="ts">
import type { ThoughtChainItemProps } from '@artmate/chat'
import { ArtStream, Bubble, ThoughtChain } from '@artmate/chat'
import { Discount } from '@element-plus/icons-vue'
import { ElButton, ElIcon, ElSpace } from 'element-plus'
import { computed, ref } from 'vue'

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 ArtStream({
    readableStream: mockReadableStream(),
  })) {
    console.log(chunk)
    lines.value = [...lines.value, chunk]
  }
}

const items = computed<ThoughtChainItemProps[]>(() => {
  if (!content.value) return []
  return [
    {
      title: 'Mock Default Protocol - Log',
      status: 'success',
      content: lines.value,
    },
  ]
})
</script>

<template>
  <ElSpace alignment="flex-start" :size="20" wrap>
    <ElSpace alignment="flex-start" direction="vertical">
      <ElButton type="primary" @click="readStream">Mock Default Protocol - SSE</ElButton>
      <Bubble v-if="content" :content="content" />
    </ElSpace>
    <ThoughtChain :items="items">
      <template #icon="{ info }">
        <ElIcon size="20" :color="info.status ? 'white' : 'block'">
          <Discount />
        </ElIcon>
      </template>
      <template #content="{ info }">
        <pre v-if="content">
          <code v-for="(item, index) in info.content" :key="index">{{ item.data }}</code>
        </pre>
      </template>
    </ThoughtChain>
  </ElSpace>
</template>

<style lang="scss" scoped>
pre {
  width: auto;
  margin: 0;
  overflow: scroll;

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

查看源代码

自定义协议

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

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

<script setup lang="ts">
import type { ThoughtChainItemProps } from '@artmate/chat'
import { ArtStream, ThoughtChain } from '@artmate/chat'
import { Discount } from '@element-plus/icons-vue'
import { ElButton, ElIcon } from 'element-plus'
import { computed, ref } from '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() {
  // 🌟 Read the stream
  for await (const chunk of ArtStream({
    readableStream: mockReadableStream(),
    transformStream: new TransformStream<string, string>({
      transform(chunk, controller) {
        controller.enqueue(chunk)
      },
    }),
  })) {
    lines.value = [...lines.value, chunk]
  }
}

const items = computed<ThoughtChainItemProps[]>(() => {
  if (!lines.value.length)
    return []
  return [
    {
      title: 'Mock Custom Protocol - Log',
      status: 'success',
      content: lines.value,
    },
  ]
})
</script>

<template>
  <div class="demo">
    <ElButton type="primary" @click="readStream">
      Mock Custom Protocol - SIP
    </ElButton>
    <ThoughtChain :items="items">
      <template #icon="{ info }">
        <ElIcon size="20" :color="info.status ? 'white' : 'block'">
          <Discount />
        </ElIcon>
      </template>
      <template #content="{ info }">
        <pre v-if="info.content">
          <code v-for="(item, index) in info.content" :key="index">{{ item }}</code>
        </pre>
      </template>
    </ThoughtChain>
  </div>
</template>

<style lang="scss" scoped>
.demo {
  display: flex;
  flex-wrap: wrap;
  gap: 10px;
  overflow-x: auto;
}
</style>

查看源代码

API

ArtStreamOptions

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