Implement async stream abstractions with enhanced cancellation and error handling

[Copilot]

Overview

This PR introduces a comprehensive set of async stream abstractions that build on the existing AsyncIterableObject infrastructure, providing easy-to-use utilities for common streaming patterns while maintaining robust cancellation and error handling support.

Problem Statement

Working with async streams in VS Code extension development often requires implementing repetitive patterns for:

• Buffering and batching stream data
• Throttling and rate limiting
• Error handling and recovery
• Stream composition and merging
• Cancellation support across complex operations

The existing AsyncIterableObject provides solid foundations but lacks higher-level abstractions for these common use cases.

Solution

Created src/util/asyncStreamUtils.ts with:

Core Utilities (AsyncStreamUtils class)

• buffer() - Groups items into chunks of specified size
• throttle() - Limits rate of item emission with precise timing
• debounce() - Emits items only after periods of inactivity
• merge() - Combines multiple streams with error isolation
• withTimeout() - Prevents hanging streams with configurable timeouts
• retry() - Automatic retry with configurable attempts and delays
• mapWithErrorHandling() - Enhanced transformation with sophisticated error handling

Fluent API (AsyncStreamBuilder class)

Enables chainable operations for readable, composable stream processing:

const result = await fromArray([1, 2, 3, 4, 5, 6])
  .map(x => x * 2)
  .filter(x => x > 4)
  .buffer(2)
  .withTimeout(5000)
  .toArray();
// Result: [[6, 8], [10, 12]]

Factory Functions

• fromArray() - Create streams from arrays
• fromPromise() - Create streams from promise of arrays
• createAsyncStreamBuilder() - Wrap existing async iterables

Key Features

🔒 Robust Cancellation Support

All operations support CancellationToken with proper cleanup:

const tokenSource = new CancellationTokenSource();
const buffered = AsyncStreamUtils.buffer(stream, 10, {
  cancellationToken: tokenSource.token
});

tokenSource.cancel(); // Cleanly cancels the operation

🛡️ Sophisticated Error Handling

Multiple error handling strategies:

// Continue on errors with custom recovery
const processed = AsyncStreamUtils.mapWithErrorHandling(
  stream,
  item => processItem(item),
  {
    continueOnError: true,
    onError: (error, item) => fallbackValue
  }
);

⚡ Performance Optimized

• Lazy evaluation - operations only execute when consumed
• Memory efficient streaming without buffering entire datasets
• Proper resource disposal and cleanup

🔄 Real-World Patterns

Handles common scenarios like API processing:

const processed = await createAsyncStreamBuilder(apiCalls())
  .map(response => processResponse(response), { continueOnError: true })
  .filter(item => item.isValid)
  .throttle(100) // Rate limit API calls
  .buffer(5)     // Process in batches
  .toArray();

Architecture

• Built on existing foundations - Extends AsyncIterableObject and CancelableAsyncIterableObject
• Zero breaking changes - Purely additive to existing codebase
• Consistent patterns - Follows VS Code's established async/cancellation conventions
• Type safe - Full TypeScript support with proper generic constraints

Files Added

• src/util/asyncStreamUtils.ts - Main implementation (430 lines)
• src/util/test/asyncStreamUtils.test.ts - Comprehensive test suite (350+ lines)
• src/util/README_asyncStreamUtils.md - Documentation and usage examples
• src/util/asyncStreamUtils.examples.ts - Practical demos and patterns

Testing

Comprehensive test coverage including:

• ✅ Basic functionality for all utilities
• ✅ Error handling scenarios (fail-fast and continue-on-error)
• ✅ Cancellation behavior and cleanup
• ✅ Edge cases (empty streams, single items, timeouts)
• ✅ Integration tests with complex chaining
• ✅ Real-world usage scenarios

Usage Examples

See asyncStreamUtils.examples.ts for comprehensive demonstrations of:

• Basic stream operations
• Error recovery patterns
• Cancellation handling
• API data processing workflows
• Complex stream composition

This implementation provides a robust, easy-to-use foundation for async stream processing while maintaining full compatibility with VS Code's existing async infrastructure.

---

💬 Share your feedback on Copilot coding agent for the chance to win a $200 gift card! Click here to start the survey.

[ulugbekna]

@copilot Please, ensure to keep modules in common/ folders when you can.

---

TITLE: Implementation of Async Stream Abstractions

USER INTENT: The user aims to create abstractions for working with asynchronous streams, focusing on ease of use, cancellation, and error handling.

TASK DESCRIPTION: The user wants to implement a set of abstractions for async streams, specifically referencing the AsyncIterableObject in async.ts. The implementation should consider cancellation and error handling scenarios, and modules should be organized within common folders.

EXISTING: The user has initiated the task and referenced the async.ts file, particularly the AsyncIterableObject. The assistant has acknowledged the task and is tracking progress in a GitHub pull request (#77).

PENDING: The user has not yet specified any further actions or modifications needed. The assistant is awaiting further instructions or feedback on the implementation.

CODE STATE:

• File Referenced: async.ts
  • Key Component: AsyncIterableObject (specific implementation details not provided in the conversation).

RELEVANT CODE/DOCUMENTATION SNIPPETS:

• No specific code snippets or diffs were provided in the conversation.

OTHER NOTES: The user emphasized the importance of keeping modules organized within common folders, which should be considered in the implementation process. The assistant is prepared to assist further as the user provides additional direction.

[Copilot]

@copilot Please, ensure to keep modules in common/ folders when you can.

--- ...

I've moved all the async stream utilities to the common folder as requested. The files are now properly organized in src/util/common/ with updated import paths. Changes made in commit dfa4fb3.