WorkflowStream

Defined in: core/src/core/stream.ts:106

A reactive stream for processing workflow events.

WorkflowStream extends the standard ReadableStream to provide specialized methods for filtering, transforming, and consuming workflow events. It supports reactive patterns and can be used to build complex event processing pipelines.

Example

// 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);
\}

Extends

• ReadableStream<R>

Type Parameters

R

R = any

The type of data flowing through the stream

Implements

• AsyncIterable<R>

Constructors

Constructor

new WorkflowStream<R>(subscribable, rootStream): WorkflowStream<R>

Defined in: core/src/core/stream.ts:141

Parameters

subscribable

Subscribable<[R], void>

rootStream

ReadableStream<R>

Returns

WorkflowStream<R>

Overrides

ReadableStream<R>.constructor

Constructor

new WorkflowStream<R>(subscribable, rootStream): WorkflowStream<R>

Defined in: core/src/core/stream.ts:145

Parameters

subscribable

Subscribable<[R], void>

rootStream

null

Returns

WorkflowStream<R>

Overrides

ReadableStream<R>.constructor

Constructor

new WorkflowStream<R>(subscribable, rootStream): WorkflowStream<R>

Defined in: core/src/core/stream.ts:146

Parameters

subscribable

null

rootStream

null | ReadableStream<R>

Returns

WorkflowStream<R>

Overrides

ReadableStream<R>.constructor

Methods

on()

on<T>(event, handler): () => void

Defined in: core/src/core/stream.ts:130

Subscribe to specific workflow events.

Type Parameters

T

T

Parameters

event

WorkflowEvent<T>

The event type to listen for

handler

(event) => void

Function to handle the event

Returns

Unsubscribe function

(): void

Returns

void

Example

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

// Later...
unsubscribe();

---

fromReadableStream()

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

Defined in: core/src/core/stream.ts:192

Create a WorkflowStream from a standard ReadableStream.

Type Parameters

T

T = any

Parameters

stream

ReadableStream<WorkflowEventData<any>>

The ReadableStream to wrap

Returns

WorkflowStream<T>

A new WorkflowStream instance

---

fromResponse()

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

Defined in: core/src/core/stream.ts:214

Create a WorkflowStream from an HTTP Response.

Parameters

response

Response

The HTTP Response containing workflow events

eventMap

Record<string, WorkflowEvent<any>>

Map of event unique IDs to event constructors

Returns

WorkflowStream<WorkflowEventData<any>>

A new WorkflowStream instance

---

toResponse()

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

Defined in: core/src/core/stream.ts:237

Convert the stream to an HTTP Response.

Parameters

init?

ResponseInit

Optional ResponseInit parameters

transformer?

JsonEncodeTransform = ...

Optional custom transformer (defaults to JSON encoding)

Returns

R extends WorkflowEventData<any> ? Response : never

HTTP Response containing the stream data

---

forEach()

forEach(callback): Promise<void>

Defined in: core/src/core/stream.ts:314

Process each item in the stream with a callback function.

Parameters

callback

(item) => void

Function to call for each item

Returns

Promise<void>

Promise that resolves when all items are processed

Example

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

---

map()

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

Defined in: core/src/core/stream.ts:338

Transform each item in the stream.

Type Parameters

T

T

Parameters

callback

(item) => T

Function to transform each item

Returns

WorkflowStream<T>

A new WorkflowStream with transformed items

Example

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

---

take()

take(limit): WorkflowStream<R>

Defined in: core/src/core/stream.ts:369

Take only the first N items from the stream.

Parameters

limit

number

Maximum number of items to take

Returns

WorkflowStream<R>

A new WorkflowStream limited to the specified number of items

Example

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

---

filter()

Call Signature

filter(predicate): WorkflowStream<R>

Defined in: core/src/core/stream.ts:404

Filter the stream to include only items matching the predicate.

Parameters

predicate

R extends WorkflowEventData<any> ? WorkflowEvent<InferWorkflowEventData<R<R>>> : never

Event type, function, or value to filter by

Returns

WorkflowStream<R>

A new WorkflowStream containing only matching items

Example

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

Call Signature

filter(predicate): WorkflowStream<R>

Defined in: core/src/core/stream.ts:409

Filter the stream to include only items matching the predicate.

Parameters

predicate

R

Event type, function, or value to filter by

Returns

WorkflowStream<R>

A new WorkflowStream containing only matching items

Example

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

Call Signature

filter(predicate): WorkflowStream<R>

Defined in: core/src/core/stream.ts:410

Filter the stream to include only items matching the predicate.

Parameters

predicate

(event) => boolean

Event type, function, or value to filter by

Returns

WorkflowStream<R>

A new WorkflowStream containing only matching items

Example

// 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()

Call Signature

until(predicate): WorkflowStream<R>

Defined in: core/src/core/stream.ts:452

Continue the stream until the predicate is met, then terminate.

Parameters

predicate

R extends WorkflowEventData<any> ? WorkflowEvent<InferWorkflowEventData<R<R>>> : never

Event type, function, or value to stop at

Returns

WorkflowStream<R>

A new WorkflowStream that terminates when the predicate is met

Example

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

Call Signature

until(predicate): WorkflowStream<R>

Defined in: core/src/core/stream.ts:457

Continue the stream until the predicate is met, then terminate.

Parameters

predicate

(item) => boolean

Event type, function, or value to stop at

Returns

WorkflowStream<R>

A new WorkflowStream that terminates when the predicate is met

Example

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

Call Signature

until(item): WorkflowStream<R>

Defined in: core/src/core/stream.ts:458

Continue the stream until the predicate is met, then terminate.

Parameters

item

R

Returns

WorkflowStream<R>

A new WorkflowStream that terminates when the predicate is met

Example

// 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[]>

Defined in: core/src/core/stream.ts:494

Collect all items from the stream into an array.

Returns

Promise<R[]>

Promise resolving to an array of all stream items

Example

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