streams

Implement non-blocking I/O with AsyncStream and MessageChannel in Pulp.

13|1|Updated Apr 6, 2026

Search Tags:#async #callbacks #io #backpressure #transport #stream #pulp

Ready to run

npx skills add https://github.com/danielraffel/pulp --skill streams-danielraffel

💬 Prompt for ChatGPT / Claude Chat▼

Please help me install this Agent Skill.
Skill: streams
Source: https://github.com/danielraffel/pulp/tree/main/.agents/skills/streams
Command: npx skills add https://github.com/danielraffel/pulp --skill streams-danielraffel

SYSTEM DOCUMENTATION & REQUIREMENTS

What problem does it solve?

This Skill provides a structured approach to using the Pulp streaming framework to read, write, and transport data across different I/O backends without deadlocks or race conditions, reducing complexity in async code.

Core Features & Use Cases

Dispatch callbacks onto your own loop to avoid library cycles and race conditions
Backpressure handling with write_async and on_drain to prevent unbounded buffers
Cancelling drains and ensuring clean completion of queued writes
Extend with new transports (WebSocket, OSC, etc.) by implementing the transport interfaces
Message channels for structured communication over any transport

Quick Start

Instantiate an AsyncStream with your event loop and select a transport to begin reading and writing data.

Frequently Asked Questions about streams

High-intent search queries and answers about installing and using this skill.

FAQPage Schema

How do I handle backpressure in async I/O streams?▼

Backpressure in async I/O streams is handled by returning a WouldBlock status during read or write operations, which pauses execution and prevents unbounded buffer growth while waiting for the transport to drain.

What is the best way to prevent race conditions in async callbacks?▼

Preventing race conditions in async callbacks is achieved by using an executor closure to dispatch callbacks onto your own event loop, avoiding library cycles and ensuring safe cross-thread dispatch.

How do I add custom transport protocols to an async streaming framework?▼

Adding custom transport protocols to an async streaming framework requires implementing the required channel interfaces, enabling new transports like WebSocket or OSC to integrate seamlessly with the unified AsyncStream model.

Can I use a unified stream model for both local file I/O and TCP transports?▼

Yes, you can use a unified AsyncStream and MessageChannel model for local file I/O, inter-process pipes, and TCP or HTTP transports, providing a consistent interface for reading and writing data across different backends.

How do I safely cancel queued writes in a non-blocking I/O system?▼

Safely canceling queued writes in a non-blocking I/O system involves cancelling drains and ensuring clean completion of pending operations, which maintains data integrity and prevents deadlocks during shutdown.

When do I need a MessageChannel for structured communication?▼

You need a MessageChannel for structured communication when transmitting data over any transport layer, as it provides a unified interface that abstracts the underlying I/O backend and ensures reliable cross-thread dispatch.