getWritable

Get a writable stream for sending data from a workflow

Retrieves a writable stream that is associated with the current workflow run. The stream allows step functions to write data that can be read outside the workflow.

Signature

function getWritable<W = any>(
  options?: WorkflowWritableStreamOptions
): WritableStream<W>

Type Parameters

W · type · default: "any"

The type of data that will be written to the stream.

Parameters

options · WorkflowWritableStreamOptions

Configuration options for the writable stream.

Show properties

namespace · string

An optional namespace to distinguish between multiple streams associated with the same workflow run.

If not provided, a default stream is used. Use different namespaces when you need multiple independent streams.

Returns

WritableStream · WritableStream<W>

A WritableStream that can be written to from within step functions.

The stream follows the Web Streams API standard.

Usage

Basic Stream Writing

Write data to the stream from a step:

import { getWritable, step } from 'workflow';

export async function workflowWithStream() {
  "use workflow";

  await step(async () => {
    "use step";
    
    const writable = getWritable<string>();
    const writer = writable.getWriter();
    
    await writer.write('Hello');
    await writer.write('World');
    await writer.close();
  });
}

Stream JSON Objects

Write structured data:

import { getWritable, step } from 'workflow';

interface LogEntry {
  timestamp: Date;
  message: string;
  level: 'info' | 'warn' | 'error';
}

export async function workflowWithLogs() {
  "use workflow";

  await step(async () => {
    "use step";
    
    const writable = getWritable<LogEntry>();
    const writer = writable.getWriter();
    
    await writer.write({
      timestamp: new Date(),
      message: 'Process started',
      level: 'info',
    });
    
    await writer.write({
      timestamp: new Date(),
      message: 'Processing data',
      level: 'info',
    });
    
    await writer.close();
  });
}

Multiple Namespaced Streams

Use different streams for different purposes:

import { getWritable, step } from 'workflow';

export async function workflowWithMultipleStreams() {
  "use workflow";

  await step(async () => {
    "use step";
    
    // Stream for logs
    const logStream = getWritable<string>({ namespace: 'logs' });
    const logWriter = logStream.getWriter();
    
    // Stream for metrics
    const metricStream = getWritable<number>({ namespace: 'metrics' });
    const metricWriter = metricStream.getWriter();
    
    await logWriter.write('Starting process');
    await metricWriter.write(100);
    
    await logWriter.write('Process complete');
    await metricWriter.write(200);
    
    await logWriter.close();
    await metricWriter.close();
  });
}

Stream Progress Updates

Send real-time progress to external consumers:

import { getWritable, step, sleep } from 'workflow';

interface Progress {
  current: number;
  total: number;
  message: string;
}

export async function workflowWithProgress(items: string[]) {
  "use workflow";

  for (let i = 0; i < items.length; i++) {
    await step(async () => {
      "use step";
      
      const stream = getWritable<Progress>();
      const writer = stream.getWriter();
      
      await writer.write({
        current: i + 1,
        total: items.length,
        message: `Processing ${items[i]}`,
      });
      
      await writer.close();
    });
    
    await processItem(items[i]);
    await sleep('1s');
  }
}

Pipe Transform Stream

Pipe data through a transform:

import { getWritable, step } from 'workflow';

export async function workflowWithTransform() {
  "use workflow";

  await step(async () => {
    "use step";
    
    const writable = getWritable<string>();
    
    // Create a transform stream to uppercase strings
    const transform = new TransformStream<string, string>({
      transform(chunk, controller) {
        controller.enqueue(chunk.toUpperCase());
      },
    });
    
    const writer = transform.writable.getWriter();
    
    // Pipe transformed data to the workflow stream
    transform.readable.pipeTo(writable);
    
    await writer.write('hello');
    await writer.write('world');
    await writer.close();
  });
}

Reading from Outside the Workflow

Consume the stream data from outside:

// In your application code (not in workflow)
import { getRun } from 'workflow/client';

const run = await getRun(runId);
const readable = run.readable({ namespace: 'logs' });

for await (const chunk of readable) {
  console.log('Received:', chunk);
}

Error Handling

Handle stream errors:

import { getWritable, step } from 'workflow';

export async function workflowWithErrorHandling() {
  "use workflow";

  await step(async () => {
    "use step";
    
    const writable = getWritable<string>();
    const writer = writable.getWriter();
    
    try {
      await writer.write('Data 1');
      await writer.write('Data 2');
      await writer.close();
    } catch (error) {
      await writer.abort(error);
      throw error;
    }
  });
}

Notes

• Can be called from both workflow functions and step functions
• The stream is automatically serialized and sent to the workflow server
• Data written to the stream can be read outside the workflow using getRun().readable()
• Multiple streams can be created using different namespaces
• The stream follows the Web Streams API standard
• Written data is automatically serialized using the workflow serialization format
• The stream is associated with the current workflow run ID

Related

• getStepMetadata - Get current step metadata
• getWorkflowMetadata - Get workflow run metadata