@electric-ax/durable-streams-client-beta

中文文档

0.3.1发布于 2026/04/28Apache-2.0TypeScriptESM + CJS近 30 天下载 9568

持久流协议的脚本客户端

文档目录37 节

安装
概览
关键特性
使用量
只读: 使用 stream() (牵引式 API)
反应消耗方法
高通量写作:使用 IdempotentProducer (建议)
读取/写入:使用 DurableStream
从"现在"读取(skip 已有数据)
只读抓取( 没有实时更新)
API 密码
stream(options): Promise<StreamResponse>
DurableStream
生活模式
有 SSE 的二进制流
标题和参数
处理错误
流化反应方法
承诺帮助者
可读结构
订阅者
生命周期
状态属性
持久性结构方法
静态方法
实例方法
《同上》
构造器
方法
属性
处理错误
连续关闭(EOF)
作者边
阅读器侧面
处理错误
类型
许可证

@可持久流/客户

持久流协议的TypeScript客户端.

安装

npm install @durable-streams/client

概览

持久流客户端提供三个主要API:

stream() 函数 - 类似获取的只读 API 用于消耗流
DurableStream 类 - 溪流上读/写操作的手柄
IdempotentProducer 类 高通量制作人,精准地写出语义(推荐写作)

关键特性

完全正确重复数字 : IdempotentProducer 为 Kafka 风格的完全对等语义提供自动调值
自动剪接: 为高吞吐量, 多个写作自动分批进行
管道: 最高吞吐量默认情况下飞行中最多5个同时批次
流线读取数字 : stream() 财务报告和审定财务报表 DurableStream.stream() 提供丰富的消费选择(承诺、可读结构、订户)
可偿还:基于偏移的读取让您从任意点恢复
实时: 用于现场跟踪的长波和SSE模式,从任何冲抵中追赶

使用量

只读: 使用 `stream()` (牵引式 API)

这个 stream() 函数为从流中读取提供了简单而像获取的界面:

import { stream } from "@durable-streams/client"

// Connect and get a StreamResponse
const res = await stream<{ message: string }>({
  url: "https://streams.example.com/my-account/chat/room-1",
  headers: {
    Authorization: `Bearer ${process.env.DS_TOKEN!}`,
  },
  offset: savedOffset, // optional: resume from offset
  live: true, // default: auto-select best live mode
})

// Accumulate all JSON items until up-to-date
const items = await res.json()
console.log("All items:", items)

// Or stream live with a subscriber
res.subscribeJson(async (batch) => {
  for (const item of batch.items) {
    console.log("item:", item)
    saveOffset(batch.offset) // persist for resumption
  }
})

反应消耗方法

这个 StreamResponse 对象返回 stream() 提供多种消耗数据的方法:

// Promise helpers (accumulate until first upToDate)
const bytes = await res.body() // Uint8Array
const items = await res.json() // Array<TJson>
const text = await res.text() // string

// ReadableStreams
const byteStream = res.bodyStream() // ReadableStream<Uint8Array>
const jsonStream = res.jsonStream() // ReadableStream<TJson>
const textStream = res.textStream() // ReadableStream<string>

// Subscribers (with backpressure)
const unsubscribe = res.subscribeJson(async (batch) => {
  await processBatch(batch.items)
})
const unsubscribe2 = res.subscribeBytes(async (chunk) => {
  await processBytes(chunk.data)
})
const unsubscribe3 = res.subscribeText(async (chunk) => {
  await processText(chunk.text)
})

高通量写作:使用 `IdempotentProducer` (建议)

对于可靠且高通量的语义写作,使用 IdempotentProducer数字 :

import { DurableStream, IdempotentProducer } from "@durable-streams/client"

const stream = await DurableStream.create({
  url: "https://streams.example.com/events",
  contentType: "application/json",
})

const producer = new IdempotentProducer(stream, "event-processor-1", {
  autoClaim: true,
  onError: (err) => console.error("Batch failed:", err), // Errors reported here
})

// Fire-and-forget - don't await, errors go to onError callback
for (const event of events) {
  producer.append(event) // Objects serialized automatically for JSON streams
}

// IMPORTANT: Always flush before shutdown to ensure delivery
await producer.flush()
await producer.close()

对于高通量方案, append() 火和被遗忘

// Fire-and-forget - errors reported via onError callback
for (const event of events) {
  producer.append(event) // Returns void, adds to batch
}

// Always flush before shutdown to ensure delivery
await producer.flush()

为什么使用"同上"制片人?

完全一交货: 使用服务器调试 (producerId, epoch, seq) 二联
自动批量: 多个写入单 HTTP 请求
管道:同时飞行的多批次
僵尸围栏: Stale 生产商被拒绝,防止出现分裂大脑的情况
网络复原力: 在网络错误上重试的安全性( 服务器复制)

读取/写入:使用 `DurableStream`

简单的写操作或需要持续控件时:

import { DurableStream } from "@durable-streams/client"

// Create a new stream
const handle = await DurableStream.create({
  url: "https://streams.example.com/my-account/chat/room-1",
  headers: {
    Authorization: `Bearer ${process.env.DS_TOKEN!}`,
  },
  contentType: "application/json",
  ttlSeconds: 3600,
})

// Append data (simple API without exactly-once guarantees)
await handle.append(JSON.stringify({ type: "message", text: "Hello" }), {
  seq: "writer-1-000001",
})

// Read using the new stream() API
const res = await handle.stream<{ type: string; text: string }>()
res.subscribeJson(async (batch) => {
  for (const item of batch.items) {
    console.log("message:", item.text)
  }
})

从"现在"读取(skip 已有数据)

// HEAD gives you the current tail offset if the server exposes it
const handle = await DurableStream.connect({
  url,
  headers: { Authorization: `Bearer ${token}` },
})
const { offset } = await handle.head()

// Read only new data from that point on
const res = await handle.stream({ offset })
res.subscribeBytes(async (chunk) => {
  console.log("new data:", new TextDecoder().decode(chunk.data))
})

只读抓取( 没有实时更新)

// Read existing data only, stop when up-to-date
const res = await stream({
  url: "https://streams.example.com/my-stream",
  live: false,
})

const text = await res.text()
console.log("All existing data:", text)

API 密码

`stream(options): Promise<StreamResponse>`

创建类似抓取流的会话 :

const res = await stream<TJson>({
  url: string | URL,              // Stream URL
  headers?: HeadersRecord,        // Headers (static or function-based)
  params?: ParamsRecord,          // Query params (static or function-based)
  signal?: AbortSignal,           // Cancellation
  fetch?: typeof fetch,           // Custom fetch implementation
  backoffOptions?: BackoffOptions,// Retry backoff configuration
  offset?: Offset,                // Starting offset (default: start of stream)
  live?: LiveMode,                // Live mode (default: true)
  json?: boolean,                 // Force JSON mode
  onError?: StreamErrorHandler,   // Error handler
})

`DurableStream`

class DurableStream {
  readonly url: string
  readonly contentType?: string

  constructor(opts: DurableStreamConstructorOptions)

  // Static methods
  static create(opts: CreateOptions): Promise<DurableStream>
  static connect(opts: DurableStreamOptions): Promise<DurableStream>
  static head(opts: DurableStreamOptions): Promise<HeadResult>
  static delete(opts: DurableStreamOptions): Promise<void>

  // Instance methods
  head(opts?: { signal?: AbortSignal }): Promise<HeadResult>
  create(opts?: CreateOptions): Promise<this>
  delete(opts?: { signal?: AbortSignal }): Promise<void>
  close(opts?: CloseOptions): Promise<CloseResult> // Close stream (EOF)
  append(
    body: BodyInit | Uint8Array | string,
    opts?: AppendOptions
  ): Promise<void>
  appendStream(
    source: AsyncIterable<Uint8Array | string>,
    opts?: AppendOptions
  ): Promise<void>

  // Fetch-like read API
  stream<TJson>(opts?: StreamOptions): Promise<StreamResponse<TJson>>
}

生活模式

// true (default): auto-select best live mode
// - SSE for JSON streams, long-poll for binary
// - Promise helpers (body/json/text): stop after upToDate
// - Streams/subscribers: continue with live updates

// false: catch-up only, stop at first upToDate
const res = await stream({ url, live: false })

// "long-poll": explicit long-poll mode for live updates
const res = await stream({ url, live: "long-poll" })

// "sse": explicit SSE mode for live updates
const res = await stream({ url, live: "sse" })

有 SSE 的二进制流

对于二进制内容类型(例如, application/octet-stream),SSE模式要求 encoding 选项 :

const stream = await DurableStream.create({
  url: "https://streams.example.com/my-binary-stream",
  contentType: "application/octet-stream",
})

const response = await stream.read({
  live: "sse",
  encoding: "base64",
})

response.subscribe((chunk) => {
  console.log(chunk.data) // Uint8Array - automatically decoded from base64
})

客户端在返回基础64数据事件之前会自动解码. 除下列内容外,其他任何内容类型都需要这样做: text/* 或者说 application/json 当使用 SSE 模式时.

标题和参数

标题和params支持静态值和函数(sync或async),用于像认证符等动态值.

// Static headers
{
  headers: {
    Authorization: "Bearer my-token",
    "X-Custom-Header": "value",
  }
}

// Function-based headers (sync)
{
  headers: {
    Authorization: () => `Bearer ${getCurrentToken()}`,
    "X-Tenant-Id": () => getCurrentTenant(),
  }
}

// Async function headers (for refreshing tokens)
{
  headers: {
    Authorization: async () => {
      const token = await refreshToken()
      return `Bearer ${token}`
    }
  }
}

// Mix static and function headers
{
  headers: {
    "X-Static": "always-the-same",
    Authorization: async () => `Bearer ${await getToken()}`,
  }
}

// Query params work the same way
{
  params: {
    tenant: "static-tenant",
    region: () => getCurrentRegion(),
    token: async () => await getSessionToken(),
  }
}

处理错误

import { stream, FetchError, DurableStreamError } from "@durable-streams/client"

const res = await stream({
  url: "https://streams.example.com/my-stream",
  headers: {
    Authorization: "Bearer my-token",
  },
  onError: async (error) => {
    if (error instanceof FetchError) {
      if (error.status === 401) {
        const newToken = await refreshAuthToken()
        return { headers: { Authorization: `Bearer ${newToken}` } }
      }
    }
    if (error instanceof DurableStreamError) {
      console.error(`Stream error: ${error.code}`)
    }
    return {} // Retry with same params
  },
})

流化反应方法

这个 StreamResponse 对象提供了多种方式来消耗流数据。所有方法都尊重 live 模式设置.

承诺帮助者

这些方法积累数据直到流更新,然后解析.

`body(): Promise<Uint8Array>`

累积所有字节直到更新.

const res = await stream({ url, live: false })
const bytes = await res.body()
console.log("Total bytes:", bytes.length)

// Process as needed
const text = new TextDecoder().decode(bytes)

`json(): Promise<Array<TJson>>`

累计所有JSON项目直至更新. 只有JSON的内容.

const res = await stream<{ id: number; name: string }>({
  url,
  live: false,
})
const items = await res.json()

for (const item of items) {
  console.log(`User ${item.id}: ${item.name}`)
}

`text(): Promise<string>`

累积出所有文本直到更新.

const res = await stream({ url, live: false })
const text = await res.text()
console.log("Full content:", text)

可读结构

Web Streams API 用于向其他流流提供管道或使用流式API. 可读结构可以使用任意一种 getReader() 或者说 for await...of 语法.

Safari/iOS兼容性: 客户端确保所有返回的流可以通过定义同步插入 [Symbol.asyncIterator] 中,当缺少时。允许 for await...of 在不要求全球多填充的情况下消费 instanceof ReadableStream 行为.

衍生流: 通过 .pipeThrough() 或类似的转换不自动补丁。使用导出 asAsyncIterableReadableStream() 帮助者 :
import { asAsyncIterableReadableStream } from "@durable-streams/client"

const derived = res.bodyStream().pipeThrough(myTransform)
const iterable = asAsyncIterableReadableStream(derived)
for await (const chunk of iterable) { ... }

`bodyStream(): ReadableStream<Uint8Array> & AsyncIterable<Uint8Array>`

Raw 字节作为可读结构 .

使用 getReader()数字 :

const res = await stream({ url, live: false })
const readable = res.bodyStream()

const reader = readable.getReader()
while (true) {
  const { done, value } = await reader.read()
  if (done) break
  console.log("Received:", value.length, "bytes")
}

使用 for await...of数字 :

const res = await stream({ url, live: false })

for await (const chunk of res.bodyStream()) {
  console.log("Received:", chunk.length, "bytes")
}

管道到文件( Node.js) :

import { Readable } from "node:stream"
import { pipeline } from "node:stream/promises"

const res = await stream({ url, live: false })
await pipeline(
  Readable.fromWeb(res.bodyStream()),
  fs.createWriteStream("output.bin")
)

`jsonStream(): ReadableStream<TJson> & AsyncIterable<TJson>`

单个JSON项目作为可读结构.

使用 getReader()数字 :

const res = await stream<{ id: number }>({ url, live: false })
const readable = res.jsonStream()

const reader = readable.getReader()
while (true) {
  const { done, value } = await reader.read()
  if (done) break
  console.log("Item:", value)
}

使用 for await...of数字 :

const res = await stream<{ id: number; name: string }>({ url, live: false })

for await (const item of res.jsonStream()) {
  console.log(`User ${item.id}: ${item.name}`)
}

`textStream(): ReadableStream<string> & AsyncIterable<string>`

文本块作为可读结构 .

使用 getReader()数字 :

const res = await stream({ url, live: false })
const readable = res.textStream()

const reader = readable.getReader()
while (true) {
  const { done, value } = await reader.read()
  if (done) break
  console.log("Text chunk:", value)
}

使用 for await...of数字 :

const res = await stream({ url, live: false })

for await (const text of res.textStream()) {
  console.log("Text chunk:", text)
}

使用响应 API :

const res = await stream({ url, live: false })
const textResponse = new Response(res.textStream())
const fullText = await textResponse.text()

订阅者

订户以回压方式提供回调消费. 下一个块是不会得到的直到你的回电承诺。返回一个未订阅的函数 .

`subscribeJson(callback): () => void`

订阅带有元数据的 JSON 批次。提供后压意识消耗.

const res = await stream<{ event: string }>({ url, live: true })

const unsubscribe = res.subscribeJson(async (batch) => {
  // Process items - next batch waits until this resolves
  for (const item of batch.items) {
    await processEvent(item)
  }
  await saveCheckpoint(batch.offset)
})

// Later: stop receiving updates
setTimeout(() => {
  unsubscribe()
}, 60000)

`subscribeBytes(callback): () => void`

订阅带有元数据的字节块 .

const res = await stream({ url, live: true })

const unsubscribe = res.subscribeBytes(async (chunk) => {
  console.log("Received bytes:", chunk.data.length)
  console.log("Offset:", chunk.offset)
  console.log("Up to date:", chunk.upToDate)

  await writeToFile(chunk.data)
  await saveCheckpoint(chunk.offset)
})

`subscribeText(callback): () => void`

订阅带有元数据的文本块 .

const res = await stream({ url, live: true })

const unsubscribe = res.subscribeText(async (chunk) => {
  console.log("Text:", chunk.text)
  console.log("Offset:", chunk.offset)

  await appendToLog(chunk.text)
})

生命周期

`cancel(reason?: unknown): void`

取消流会话。中止任何待决请求.

const res = await stream({ url, live: true })

// Start consuming
res.subscribeBytes(async (chunk) => {
  console.log("Chunk:", chunk)
})

// Cancel after 10 seconds
setTimeout(() => {
  res.cancel("Timeout")
}, 10000)

`closed: Promise<void>`

保证在会议结束或取消时解决.

const res = await stream({ url, live: false })

// Start consuming in background
const consumer = res.text()

// Wait for completion
await res.closed
console.log("Stream fully consumed")

状态属性

const res = await stream({ url })

res.url // The stream URL
res.contentType // Content-Type from response headers
res.live // The live mode (true, "long-poll", "sse", or false)
res.startOffset // The starting offset passed to stream()
res.offset // Current offset (updates as data is consumed)
res.cursor // Cursor for collapsing (if provided by server)
res.upToDate // Whether we've caught up to the stream head
res.streamClosed // Whether the stream is permanently closed (EOF)

持久性结构方法

静态方法

`DurableStream.create(opts): Promise<DurableStream>`

在服务器上创建新流 .

const handle = await DurableStream.create({
  url: "https://streams.example.com/my-stream",
  headers: {
    Authorization: "Bearer my-token",
  },
  contentType: "application/json",
  ttlSeconds: 3600, // Optional: auto-delete after 1 hour
})

await handle.append('{"hello": "world"}')

`DurableStream.connect(opts): Promise<DurableStream>`

连接到已有的流( 通过 HEAD 验证其存在) .

const handle = await DurableStream.connect({
  url: "https://streams.example.com/my-stream",
  headers: {
    Authorization: "Bearer my-token",
  },
})

console.log("Content-Type:", handle.contentType)

`DurableStream.head(opts): Promise<HeadResult>`

在不创建手柄的情况下获取流元数据 .

const metadata = await DurableStream.head({
  url: "https://streams.example.com/my-stream",
  headers: {
    Authorization: "Bearer my-token",
  },
})

console.log("Offset:", metadata.offset)
console.log("Content-Type:", metadata.contentType)

`DurableStream.delete(opts): Promise<void>`

不创建控点而删除流 .

await DurableStream.delete({
  url: "https://streams.example.com/my-stream",
  headers: {
    Authorization: "Bearer my-token",
  },
})

实例方法

`head(opts?): Promise<HeadResult>`

获取此流的元数据 .

const handle = new DurableStream({
  url,
  headers: { Authorization: `Bearer ${token}` },
})
const metadata = await handle.head()

console.log("Current offset:", metadata.offset)

`create(opts?): Promise<this>`

在服务器上创建此流 .

const handle = new DurableStream({
  url,
  headers: { Authorization: `Bearer ${token}` },
})
await handle.create({
  contentType: "text/plain",
  ttlSeconds: 7200,
})

`delete(opts?): Promise<void>`

删除此流 .

const handle = new DurableStream({
  url,
  headers: { Authorization: `Bearer ${token}` },
})
await handle.delete()

`append(body, opts?): Promise<void>`

将数据附加到流中。默认, 启用自动批量: 多个 append() pOST在飞行中发出的电话将合并为一个请求。这大大提高了高频写作的吞吐量.

const handle = await DurableStream.connect({
  url,
  headers: { Authorization: `Bearer ${token}` },
})

// Append string
await handle.append("Hello, world!")

// Append with sequence number for ordering
await handle.append("Message 1", { seq: "writer-1-001" })
await handle.append("Message 2", { seq: "writer-1-002" })

// For JSON streams, append objects directly (serialized automatically)
await handle.append({ event: "click", x: 100, y: 200 })

// Batching happens automatically - these may be sent in a single request
await Promise.all([
  handle.append({ event: "msg1" }),
  handle.append({ event: "msg2" }),
  handle.append({ event: "msg3" }),
])

断接行为 :

JSON 模式 (单位:千美元)contentType: "application/json":多个值作为 JSON 数组发送 [val1, val2, ...]
字节模式二进制数据

禁用批量 :

如果您需要确保每个附件立即发送( 如精确的时间或调试) :

const handle = new DurableStream({
  url,
  batching: false, // Disable automatic batching
})

`appendStream(source, opts?): Promise<void>`

从可同步或可读结构中追加流出数据。这种方法支持来自任何来源的管道.

const handle = await DurableStream.connect({
  url,
  headers: { Authorization: `Bearer ${token}` },
})

// From async generator
async function* generateData() {
  for (let i = 0; i < 100; i++) {
    yield `Line ${i}\n`
  }
}
await handle.appendStream(generateData())

// From ReadableStream
const readable = new ReadableStream({
  start(controller) {
    controller.enqueue("chunk 1")
    controller.enqueue("chunk 2")
    controller.close()
  },
})
await handle.appendStream(readable)

// Pipe from a fetch response body
const response = await fetch("https://example.com/data")
await handle.appendStream(response.body!)

`writable(opts?): WritableStream<Uint8Array | string>`

创建可以接收管状数据的可写入结构。用于流组分 :

const handle = await DurableStream.connect({ url, auth })

// Pipe from any ReadableStream
await someReadableStream.pipeTo(handle.writable())

// Pipe through a transform
const readable = inputStream.pipeThrough(new TextEncoderStream())
await readable.pipeTo(handle.writable())

`stream(opts?): Promise<StreamResponse>`

开始一个已读会话( 与独立会话相同) stream() 函数).

const handle = await DurableStream.connect({
  url,
  headers: { Authorization: `Bearer ${token}` },
})

const res = await handle.stream<{ message: string }>({
  offset: savedOffset,
  live: true,
})

res.subscribeJson(async (batch) => {
  for (const item of batch.items) {
    console.log(item.message)
  }
})

《同上》

这个 IdempotentProducer class 提供 Kafka 风格的完全一对一的写作语义,有自动批量和管道衬线.

构造器

new IdempotentProducer(stream: DurableStream, producerId: string, opts?: IdempotentProducerOptions)

参数:

stream - 写给
producerId - 这个生产者的稳定标识符(例如“订单-服务-1”)
opts - 可选配置

选项 :

interface IdempotentProducerOptions {
  epoch?: number // Starting epoch (default: 0)
  autoClaim?: boolean // On 403, retry with epoch+1 (default: false)
  maxBatchBytes?: number // Max bytes before sending batch (default: 1MB)
  lingerMs?: number // Max time to wait for more messages (default: 5ms)
  maxInFlight?: number // Concurrent batches in flight (default: 5)
  signal?: AbortSignal // Cancellation signal
  fetch?: typeof fetch // Custom fetch implementation
  onError?: (error: Error) => void // Error callback for batch failures
}

方法

`append(body): void`

将数据附加到溪流(fire-and-forget). 对于JSON流,可以直接通过对象. 加入内部批次后立即返回。错误报告通过 onError 调用.

// For JSON streams - pass objects directly
producer.append({ event: "click", x: 100 })

// Or strings/bytes
producer.append("message data")
producer.append(new Uint8Array([1, 2, 3]))

// All appends are fire-and-forget - use flush() to wait for delivery
await producer.flush()

`flush(): Promise<void>`

立即发送任何待处理的批次,并等待所有飞行中批次完成.

// Always call before shutdown
await producer.flush()

`close(finalMessage?): Promise<CloseResult>`

流出待处理信件并关闭基础流线 (欧统局)。这是结束制作会的典型方式:

模糊所有待处理信件
可选地将最后信件在解析时附加为关闭
关闭流( 任何制片人都不允许再添加附件)

《同上》:在网络故障上进行重试的安全性 - 使用出厂头来分解.

// Close stream (EOF)
const result = await producer.close()
console.log("Final offset:", result.finalOffset)

// Close with final message (atomic append + close)
const result = await producer.close('{"done": true}')

`detach(): Promise<void>`

阻止制片人而不关闭下流. 当 :

将信件交给另一个制作人
保持溪流打开供将来写
停止这个制片人,但不向读者显示 EOF

await producer.detach() // Stream remains open

`restart(): Promise<void>`

递增时代并重置序列. 重新启动制作方以建立新会话时使用此功能 .

await producer.restart()

属性

epoch: number - 这个制片人的时代
nextSeq: number - 下一个要指定的序列号
pendingCount: number - 当前待处理批次中的信件
inFlightCount: number - 正在飞行的蝙蝠

处理错误

错误通过 onError 自此以来 append() 是火和遗忘:

import {
  IdempotentProducer,
  StaleEpochError,
  SequenceGapError,
} from "@durable-streams/client"

const producer = new IdempotentProducer(stream, "my-producer", {
  onError: (error) => {
    if (error instanceof StaleEpochError) {
      // Another producer has a higher epoch - this producer is "fenced"
      console.log(`Fenced by epoch ${error.currentEpoch}`)
    } else if (error instanceof SequenceGapError) {
      // Sequence gap detected (should not happen with proper usage)
      console.log(`Expected seq ${error.expectedSeq}, got ${error.receivedSeq}`)
    }
  },
})

producer.append("data") // Fire-and-forget, errors go to onError
await producer.flush() // Wait for all batches to complete

连续关闭(EOF)

持久流支持永久关闭流来信号 EOF( 文件结束) 。一旦关闭,则不再允许附加,但数据仍然完全可以读取.

作者边

使用持久 Stream.close ()

const stream = await DurableStream.connect({ url })

// Simple close (no final message)
const result = await stream.close()
console.log("Final offset:", result.finalOffset)

// Atomic append-and-close with final message
const result = await stream.close({
  body: '{"status": "complete"}',
})

选项 :

interface CloseOptions {
  body?: Uint8Array | string // Optional final message
  contentType?: string // Content type (must match stream)
  signal?: AbortSignal // Cancellation
}

interface CloseResult {
  finalOffset: Offset // The offset after the last byte
}

同上:

close() 无机体:同上——可以多次拨打
close({ body }) 与身体:非一能-抛出 StreamClosedError 如果已经关闭。使用 IdempotentProducer.close(finalMessage) 和身体亲近.

使用同上强生产器. close()

可靠地关闭最后信件( 安全重试) :

const producer = new IdempotentProducer(stream, "producer-1", {
  autoClaim: true,
})

// Write some messages
producer.append('{"event": "start"}')
producer.append('{"event": "data"}')

// Close with final message (idempotent, safe to retry)
const result = await producer.close('{"event": "end"}')

重要内容: IdempotentProducer.close() 关闭流线不仅仅是制片人使用 detach() 在不关闭溪流的情况下阻止制片人.

创建闭合流

创建立即关闭的流( 用于缓存回复、错误、单发数据 ):

// Empty closed stream
const stream = await DurableStream.create({
  url: "https://streams.example.com/cached-response",
  contentType: "application/json",
  closed: true,
})

// Closed stream with initial content
const stream = await DurableStream.create({
  url: "https://streams.example.com/error-response",
  contentType: "application/json",
  body: '{"error": "Service unavailable"}',
  closed: true,
})

阅读器侧面

检测关闭

这个 streamClosed 属性表示永久关闭流时 :

// StreamResponse properties
const res = await stream({ url, live: true })
console.log(res.streamClosed) // false initially

// In subscribers - batch/chunk metadata includes streamClosed
res.subscribeJson((batch) => {
  console.log("Items:", batch.items)
  console.log("Stream closed:", batch.streamClosed) // true when EOF reached
})

// In HEAD requests
const metadata = await stream.head()
console.log("Stream closed:", metadata.streamClosed)

生活模式行为

当一条溪流被关闭时 :

长波: 随即返回 streamClosed: true (没有等待)
证监会: 发送 streamClosed: true ,然后关闭连接
订阅者: 接收最后批次 streamClosed: true,然后停下来

const res = await stream({ url, live: true })

res.subscribeJson((batch) => {
  for (const item of batch.items) {
    process(item)
  }

  if (batch.streamClosed) {
    console.log("Stream complete, no more data will arrive")
    // Connection will close automatically
  }
})

处理错误

试图附加到封闭的溪流中 StreamClosedError数字 :

import { StreamClosedError } from "@durable-streams/client"

try {
  await stream.append("data")
} catch (error) {
  if (error instanceof StreamClosedError) {
    console.log("Stream is closed at offset:", error.finalOffset)
  }
}

类型

从软件包导出的关键类型 :

Offset - 流位置的不透明字符串
StreamResponse - 溪流反应对象(包括 streamClosed (财产)
ByteChunk - 说吧 { data: Uint8Array, offset: Offset, upToDate: boolean, streamClosed: boolean, cursor?: string }
JsonBatch<T> - 说吧 { items: T[], offset: Offset, upToDate: boolean, streamClosed: boolean, cursor?: string }
TextChunk - 说吧 { text: string, offset: Offset, upToDate: boolean, streamClosed: boolean, cursor?: string }
HeadResult - 来自总部请求的元数据(包括 streamClosed (财产)
CloseOptions - 关闭流的选项
CloseResult - 关闭溪流(包括 finalOffset(中文(简体) )
IdempotentProducer - 完全一档制作班
StaleEpochError - 制作者时代被打倒时扔出(僵尸围栏)
SequenceGapError - 序列号出故障时抛出
StreamClosedError - 试图接入封闭流时投掷(包括: finalOffset(中文(简体) )
DurableStreamError - 带有代码的协议级别错误
FetchError - 运输/网络错误

许可证

阿帕奇 - 2.0

版本与发布记录

@electric-ax/durable-streams-client-beta 共发布 2 个版本，首次发布于 2026/04/22；过去 12 个月发布了 2 个版本。

版本	发布日期	距今
`0.3.1`latest	2026/04/28	6 个月前
`0.3.0`	2026/04/22	6 个月前

关于 @electric-ax/durable-streams-client-beta 的常见问题

@electric-ax/durable-streams-client-beta 是什么？

持久流协议的脚本客户端

如何安装 @electric-ax/durable-streams-client-beta？

在项目目录运行 npm install @electric-ax/durable-streams-client-beta，或使用 pnpm add / yarn add / bun add。它还提供命令行工具 intent，可以用 npx intent 直接运行。

@electric-ax/durable-streams-client-beta 支持 TypeScript 吗？

支持。包内自带 TypeScript 类型声明（package.json 声明了 types 入口），安装后即可获得类型提示，无需再安装 @types 包。

@electric-ax/durable-streams-client-beta 支持 ESM 还是 CommonJS？

同时提供 ES Module 与 CommonJS 入口，import 和 require() 都可以使用，打包工具会自动选择合适的版本。

@electric-ax/durable-streams-client-beta 需要什么版本的 Node.js？

package.json 的 engines 字段要求 Node.js >=18.0.0。版本不符时 npm 默认只给出警告，开启 engine-strict 后会拒绝安装。

@electric-ax/durable-streams-client-beta 可以商用吗？

@electric-ax/durable-streams-client-beta 使用 Apache-2.0 许可。宽松许可：允许商业使用、修改和再分发，并附带明确的专利授权；分发时需保留许可与 NOTICE 文件，修改过的文件要注明改动。以上为概括，不构成法律意见，请以许可原文为准。

@electric-ax/durable-streams-client-beta 的下载量有多少？

根据 npm 官方下载统计，@electric-ax/durable-streams-client-beta 在 2026/10/14 – 2026/11/12 这 30 天内被下载 9,568 次，日均约 319 次。下载量会把 CI 构建、镜像同步和作为其他包依赖的安装一并计入，适合横向比较热度，不能直接等同于使用人数。

@electric-ax/durable-streams-client-beta 还在维护吗？

最新版本 0.3.1 发布于 2026/04/28（6 个月前）；至今共发布 2 个版本，其中过去 12 个月发布了 2 个；首次发布于 2026/04/22。发布频率只是参考信号：功能稳定的工具可能很少需要更新，建议结合源码仓库的 issue 和提交记录判断。

npm install @electric-ax/durable-streams-client-beta

pnpmpnpm add @electric-ax/durable-streams-client-beta
yarnyarn add @electric-ax/durable-streams-client-beta
bunbun add @electric-ax/durable-streams-client-beta

命令行npx intent

在你的项目目录中运行此命令。

近 30 天下载

9,568

约 9568 次日均 319 次

统计周期 2026/10/14 – 2026/11/12
数据来自 npm 官方下载统计，今天更新

最新版本0.3.1发布于 6 个月前

开源许可Apache-2.0宽松许可

包体积559 kB28 个文件 · 解压后

运行依赖2 个

模块格式ESM + CJS无副作用，可 tree-shake

TypeScript自带类型

Node.js>=18.0.0

发布版本2 个近 12 个月 2 个

首次发布2026/04/226 个月前

元数据更新2026/04/28

每一个软件包背后，
都有分享与创造的人。

@electric-ax/durable-streams-client-beta

@可持久流/客户

安装

概览

关键特性

使用量

只读: 使用 stream() (牵引式 API)

反应消耗方法

高通量写作:使用 IdempotentProducer (建议)

读取/写入:使用 DurableStream

从"现在"读取(skip 已有数据)

只读抓取( 没有实时更新)

API 密码

stream(options): Promise<StreamResponse>

DurableStream

生活模式

有 SSE 的 二进制流

标题和参数

处理错误

流化反应方法

承诺帮助者

body(): Promise<Uint8Array>

json(): Promise<Array<TJson>>

text(): Promise<string>

可读结构

bodyStream(): ReadableStream<Uint8Array> & AsyncIterable<Uint8Array>

jsonStream(): ReadableStream<TJson> & AsyncIterable<TJson>

textStream(): ReadableStream<string> & AsyncIterable<string>

订阅者

subscribeJson(callback): () => void

subscribeBytes(callback): () => void

subscribeText(callback): () => void

生命周期

cancel(reason?: unknown): void

closed: Promise<void>

状态属性

持久性结构方法

静态方法

DurableStream.create(opts): Promise<DurableStream>

DurableStream.connect(opts): Promise<DurableStream>

DurableStream.head(opts): Promise<HeadResult>

DurableStream.delete(opts): Promise<void>

实例方法

head(opts?): Promise<HeadResult>

create(opts?): Promise<this>

delete(opts?): Promise<void>

append(body, opts?): Promise<void>

appendStream(source, opts?): Promise<void>

writable(opts?): WritableStream<Uint8Array | string>

stream(opts?): Promise<StreamResponse>

《同上》

构造器

方法

append(body): void

flush(): Promise<void>

close(finalMessage?): Promise<CloseResult>

detach(): Promise<void>

restart(): Promise<void>

属性

处理错误

连续关闭(EOF)

作者边

使用持久 Stream.close ()

使用同上强生产器. close()

创建闭合流

阅读器侧面

检测关闭

生活模式行为

处理错误

类型

许可证

版本与发布记录

关于 @electric-ax/durable-streams-client-beta 的常见问题

@electric-ax/durable-streams-client-beta 是什么？

如何安装 @electric-ax/durable-streams-client-beta？

@electric-ax/durable-streams-client-beta 支持 TypeScript 吗？

@electric-ax/durable-streams-client-beta 支持 ESM 还是 CommonJS？

@electric-ax/durable-streams-client-beta 需要什么版本的 Node.js？

@electric-ax/durable-streams-client-beta 可以商用吗？

@electric-ax/durable-streams-client-beta 的下载量有多少？

只读: 使用 `stream()` (牵引式 API)

高通量写作:使用 `IdempotentProducer` (建议)

读取/写入:使用 `DurableStream`

`stream(options): Promise<StreamResponse>`

`DurableStream`

有 SSE 的二进制流

`body(): Promise<Uint8Array>`

`json(): Promise<Array<TJson>>`

`text(): Promise<string>`

`bodyStream(): ReadableStream<Uint8Array> & AsyncIterable<Uint8Array>`

`jsonStream(): ReadableStream<TJson> & AsyncIterable<TJson>`

`textStream(): ReadableStream<string> & AsyncIterable<string>`

`subscribeJson(callback): () => void`

`subscribeBytes(callback): () => void`

`subscribeText(callback): () => void`

`cancel(reason?: unknown): void`

`closed: Promise<void>`

`DurableStream.create(opts): Promise<DurableStream>`

`DurableStream.connect(opts): Promise<DurableStream>`

`DurableStream.head(opts): Promise<HeadResult>`

`DurableStream.delete(opts): Promise<void>`

`head(opts?): Promise<HeadResult>`

`create(opts?): Promise<this>`

`delete(opts?): Promise<void>`

`append(body, opts?): Promise<void>`

`appendStream(source, opts?): Promise<void>`

`writable(opts?): WritableStream<Uint8Array | string>`

`stream(opts?): Promise<StreamResponse>`

`append(body): void`

`flush(): Promise<void>`

`close(finalMessage?): Promise<CloseResult>`

`detach(): Promise<void>`

`restart(): Promise<void>`