WorkflowStream

用于处理工作流事件的响应式流。

WorkflowStream 扩展了标准 ReadableStream，为过滤、转换和消费工作流事件提供专门方法。它支持响应式模式，可用于构建复杂的事件处理流水线。

示例

// Get stream from workflow context
const stream = context.stream;

// Filter for specific events
const userEvents = stream.filter(UserEvent);

// Transform events
const processed = stream.map(event => (\{
  type: event.constructor.name,
  timestamp: Date.now(),
  data: event.data
\}));

// Consume events
for await (const event of stream.take(10)) \{
  console.log('Received:', event);
\}

扩展

ReadableStream<R>

类型参数

R

R = any

流经数据流的类型

实现

AsyncIterable<R>

构造函数

新工作流流<R>(subscribable, rootStream): WorkflowStream<R>

定义于：core/src/core/stream.ts:141

参数

可订阅

Subscribable<[R], void>

rootStream

ReadableStream<R>

WorkflowStream<R>

覆盖设置

ReadableStream<R>.constructor

构造函数

新工作流流<R>(subscribable, rootStream): WorkflowStream<R>

定义于：core/src/core/stream.ts:145

参数

可订阅

Subscribable<[R], void>

rootStream

null

WorkflowStream<R>

覆盖设置

ReadableStream<R>.constructor

构造函数

新工作流流<R>(subscribable, rootStream): WorkflowStream<R>

定义于：core/src/core/stream.ts:146

参数

可订阅

null

rootStream

null | ReadableStream<R>

WorkflowStream<R>

覆盖设置

ReadableStream<R>.constructor

方法

on()

开启<T>(event, handler): () => void

定义于：core/src/core/stream.ts:130

订阅特定的工作流事件。

类型参数

T

T

参数

事件

WorkflowEvent<T>

要监听的事件类型

处理器

(event) => void

处理事件的函数

取消订阅功能

(): void

void

示例

const unsubscribe = stream.on(UserEvent, (event) => \{
  console.log('User event:', event.data);
\});

// Later...
unsubscribe();

fromReadableStream()

static fromReadableStream<T>(stream): WorkflowStream<T>

定义于：core/src/core/stream.ts:192

从标准可读流创建一个工作流流。

类型参数

T

T = any

参数

流

ReadableStream<WorkflowEventData<any>>

要包装的可读流

WorkflowStream<T>

一个新的 WorkflowStream 实例

fromResponse()

static fromResponse(response, eventMap): WorkflowStream<WorkflowEventData<any>>

定义于：core/src/core/stream.ts:214

从HTTP响应创建一个工作流流。

参数

响应

Response

包含工作流事件的HTTP响应

eventMap

Record<string, WorkflowEvent<any>>

事件唯一标识符到事件构造函数的映射

WorkflowStream<WorkflowEventData<any>>

一个新的 WorkflowStream 实例

toResponse()

toResponse(init?, transformer?): R 继承 WorkflowEventData<any> ? Response : never

定义于：core/src/core/stream.ts:237

将数据流转换为HTTP响应。

参数

初始化？

ResponseInit

可选的ResponseInit参数

transformer？

JsonEncodeTransform = ...

可选的自定义转换器（默认为JSON编码）

R 继承 WorkflowEventData<any> ? Response : never

包含流数据的HTTP响应

forEach()

forEach(callback): Promise<void>

定义于：core/src/core/stream.ts:314

使用回调函数处理流中的每个项目。

参数

回调函数

(item) => void

为每个项目调用的函数

Promise<void>

当所有项目处理完成时解决的Promise

示例

await stream.forEach(event => \{
  console.log('Processing:', event);
\});

map()

map<T>(callback): WorkflowStream<T>

定义于：core/src/core/stream.ts:338

转换流中的每个项目。

类型参数

T

T

参数

回调函数

(item) => T

用于转换每个项目的函数

WorkflowStream<T>

一个包含转换后项目的新工作流流

示例

const timestamps = stream.map(event => (\{
  ...event,
  timestamp: Date.now()
\}));

take()

take(limit): WorkflowStream<R>

定义于：core/src/core/stream.ts:369

仅从流中获取前N个项目。

参数

限制

number

最大可获取项目数量

WorkflowStream<R>

一个限制为指定项目数量的新 WorkflowStream

示例

const firstTen = stream.take(10);
for await (const event of firstTen) \{
  console.log(event);
\}

filter()

调用签名

filter(predicate): WorkflowStream<R>

定义于：core/src/core/stream.ts:404

过滤数据流，仅包含符合谓词条件的条目。

参数

谓词

R 扩展 WorkflowEventData<any> ? WorkflowEvent<InferWorkflowEventData<R<R>>> : never

按事件类型、函数或值进行筛选

WorkflowStream<R>

一个仅包含匹配项的新 WorkflowStream

示例

// Filter by event type
const userEvents = stream.filter(UserEvent);

// Filter by function
const importantEvents = stream.filter(event => event.priority === 'high');

// Filter by specific value
const specificEvent = stream.filter(myEventInstance);

调用签名

filter(predicate): WorkflowStream<R>

定义于：core/src/core/stream.ts:409

过滤数据流，仅包含符合谓词条件的项目。

参数

谓词

R

按事件类型、函数或值进行筛选

WorkflowStream<R>

一个仅包含匹配项的新 WorkflowStream

示例

// Filter by event type
const userEvents = stream.filter(UserEvent);

// Filter by function
const importantEvents = stream.filter(event => event.priority === 'high');

// Filter by specific value
const specificEvent = stream.filter(myEventInstance);

调用签名

filter(predicate): WorkflowStream<R>

定义于：core/src/core/stream.ts:410

过滤数据流，仅包含符合谓词条件的项目。

参数

谓词

(event) => boolean

按事件类型、函数或值进行筛选

WorkflowStream<R>

一个仅包含匹配项的新 WorkflowStream

示例

// Filter by event type
const userEvents = stream.filter(UserEvent);

// Filter by function
const importantEvents = stream.filter(event => event.priority === 'high');

// Filter by specific value
const specificEvent = stream.filter(myEventInstance);

until()

调用签名

until(predicate): WorkflowStream<R>

定义于：core/src/core/stream.ts:452

持续流式处理直到满足谓词条件，然后终止。

参数

谓词

R 扩展 WorkflowEventData<any> ? WorkflowEvent<InferWorkflowEventData<R<R>>> : never

事件类型、函数或停止值

WorkflowStream<R>

当满足条件时终止的新工作流流

示例

// Stop at completion event
const processingEvents = stream.until(CompletionEvent);

// Stop when condition is met
const beforeError = stream.until(event => event.type === 'error');

// Stop at specific event instance
const beforeSpecific = stream.until(myEventInstance);

调用签名

until(predicate): WorkflowStream<R>

定义于：core/src/core/stream.ts:457

继续流式处理直到满足谓词条件，然后终止。

参数

谓词

(item) => boolean

事件类型、函数或停止值

WorkflowStream<R>

当满足条件时终止的新工作流流

示例

// Stop at completion event
const processingEvents = stream.until(CompletionEvent);

// Stop when condition is met
const beforeError = stream.until(event => event.type === 'error');

// Stop at specific event instance
const beforeSpecific = stream.until(myEventInstance);

调用签名

until(item): WorkflowStream<R>

定义于：core/src/core/stream.ts:458

持续流式处理直到满足谓词条件，然后终止。

参数

项目

R

WorkflowStream<R>

当满足条件时终止的新工作流流

示例

// Stop at completion event
const processingEvents = stream.until(CompletionEvent);

// Stop when condition is met
const beforeError = stream.until(event => event.type === 'error');

// Stop at specific event instance
const beforeSpecific = stream.until(myEventInstance);

toArray()

toArray(): Promise<R[]>

定义于：core/src/core/stream.ts:494

将流中的所有项目收集到一个数组中。

Promise<R[]>

Promise 解析为所有流项目的数组

示例

const events = await stream.take(5).toArray();
console.log('Collected events:', events);

WorkflowStream

示例

扩展

类型参数

R

实现

构造函数

构造函数

参数

可订阅

rootStream

返回

覆盖设置

构造函数

参数

可订阅

rootStream

返回

覆盖设置

构造函数

参数

可订阅

rootStream

返回

覆盖设置

方法

on()

类型参数

T

参数

事件

处理器

返回

返回

示例

fromReadableStream()

类型参数

T

参数

流

返回

fromResponse()

参数

响应

eventMap

返回

toResponse()

参数

初始化？

transformer？

返回

forEach()

参数

回调函数

返回

示例

map()

类型参数

T

参数

回调函数

返回

示例

take()

参数

限制

返回

示例

filter()

调用签名

参数

谓词

返回

示例

调用签名

参数

谓词

返回

示例

调用签名