# @durable-streams/client 中文文档

规范地址：https://npmcn.com/package/%40durable-streams/client/

版本：0\.2\.7
npm 元数据更新时间：2026-09-10T17:20:30.643Z
中文译文更新时间：2026-11-13T00:34:54.208Z
软件包许可：Apache-2\.0

本文由上游 README 翻译，可能存在偏差，请结合原文核对 API、命令与配置。版权与许可归原作者所有。

原始 npm 页面：https://www.npmjs.com/package/%40durable-streams/client
原文对照：https://npmcn.com/package/%40durable-streams/client/?lang=original
项目主页：https://github.com/durable-streams/durable-streams#readme
源码仓库：https://github.com/durable-streams/durable-streams

## 关键信息（来自 npm 元数据）

- 包体积（解压后）：571 kB，29 个文件
- 模块格式：ESM + CJS
- TypeScript 类型：包内自带
- Node.js 要求：\>=18\.0\.0
- 运行依赖：1 个
- 首次发布：2025-12-23T17:50:18.715Z
- 最新版本发布：2026-09-10T17:20:30.275Z
- 近 30 天下载量：1787609（2026-10-13 至 2026-11-11，npm 官方下载统计）
- 已发布版本数：14（近 12 个月 14 个）

## 中文简介

持久流协议的脚本客户端

## 安装

```sh
npm install @durable-streams/client
```

也可以使用：`pnpm add @durable-streams/client`、`yarn add @durable-streams/client`、`bun add @durable-streams/client`

## README

# @可持久流/客户

持久流协议的TypeScript客户端.

## 安装

```bash
npm install @durable-streams/client
```

## 概览

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

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

## 关键特性

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

## 使用量

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

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

```typescript
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()` 提供多种消耗数据的方法:

```typescript
// 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`数字 :

```typescript
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()` 火和被遗忘

```typescript
// 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`

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

```typescript
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 已有数据)

```typescript
// 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))
})
```

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

```typescript
// 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>`

创建类似抓取流的会话 :

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

```typescript
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>>
}
```

### 生活模式

```typescript
// 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` 选项 :

```typescript
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),用于像认证符等动态值.

```typescript
// 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(),
  }
}
```

### 处理错误

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

累积所有字节直到更新.

```typescript
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的内容.

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

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

```typescript
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()` 帮助者 :
> 
> ```typescript
> 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()`数字 :**

```typescript
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`数字 :**

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

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

**管道到文件( Node.js) :**

```typescript
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()`数字 :**

```typescript
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`数字 :**

```typescript
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()`数字 :**

```typescript
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`数字 :**

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

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

**使用响应 API :**

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

### 订阅者

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

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

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

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

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

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

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

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

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

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

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

```typescript
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")
```

### 状态属性

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

在服务器上创建新流 .

```typescript
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 验证其存在) .

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

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

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

不创建控点而删除流 .

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

### 实例方法

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

获取此流的元数据 .

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

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

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

在服务器上创建此流 .

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

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

删除此流 .

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

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

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

```typescript
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, ...]`
-   **字节模式**二进制数据

**禁用批量 :**

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

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

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

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

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

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

```typescript
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()` 函数).

```typescript
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 风格的完全一对一的写作语义,有自动批量和管道衬线.

### 构造器

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

**参数:**

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

**选项 :**

```typescript
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)
  headers?: HeadersRecord // Extra headers for producer batch/close requests
  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` 调用.

```typescript
// 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>`

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

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

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

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

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

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

```typescript
// 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

```typescript
await producer.detach() // Stream remains open
```

#### `restart(): Promise<void>`

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

```typescript
await producer.restart()
```

### 属性

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

### 处理错误

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

```typescript
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 ()

```typescript
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"}',
})
```

**选项 :**

```typescript
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()

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

```typescript
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()` 在不关闭溪流的情况下阻止制片人.

#### 创建闭合流

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

```typescript
// 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` 属性表示永久关闭流时 :

```typescript
// 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`,然后停下来

```typescript
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`数字 :

```typescript
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)
  }
}
```

> **业绩说明:** `append()` 时间重叠的电话(无 默认情况下,将(等待)分批成一个POST。 狦 `await` 每个 在紧闭的回路内打电话 批量从不参与。 对于一个循环 可同步( 如 LLM 流), 偏好 `appendStream()` 编号 : `writable()`, (中文(简体) ), 火灾 `append()` 而不等待,只等待最后的应许 `close()`在结尾.

* * *

## 类型

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

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

## 常见问题

### @durable-streams/client 是什么？

持久流协议的脚本客户端

### 如何安装 @durable-streams/client？

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

### @durable-streams/client 支持 TypeScript 吗？

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

### @durable-streams/client 支持 ESM 还是 CommonJS？

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

### @durable-streams/client 需要什么版本的 Node\.js？

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

### @durable-streams/client 可以商用吗？

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

### @durable-streams/client 的下载量有多少？

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

### @durable-streams/client 还在维护吗？

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