Add stream support with generators for functional reactive programming

[Copilot]

Overview

This PR adds comprehensive stream support to aspipes, enabling functional reactive programming (FRP) patterns with async generators. Now you can process endless event streams and wait for particular events using composable pipeline operations.

Note: This PR has been merged with the main branch, which includes composable pipes functionality. The merge combines both feature sets seamlessly.

What's New

Stream Processing Functions

Added five new generator-based aspipe functions in stream.js:

• map(iterable, fn) - Transform each item in an async generator
• filter(iterable, predicate) - Filter items based on a condition
• take(iterable, n) - Take first n items (essential for endless streams)
• scan(iterable, reducer, initial) - Accumulate values, yielding intermediate results
• reduce(iterable, reducer, initial) - Reduce stream to a single value

Key Capabilities

Processing Endless Streams:

import { createAsPipes } from 'aspipes';
import { createStreamPipes, eventStream } from 'aspipes/stream';

const { pipe, asPipe } = createAsPipes();
const { map, filter, take } = createStreamPipes(asPipe);

// Process an infinite event stream
async function* infiniteEvents() {
  let id = 0;
  while (true) {
    yield { id: id++, type: id % 3 === 0 ? 'special' : 'normal' };
  }
}

// Take first 5 "special" events from the endless stream
const result = pipe(infiniteEvents())
  | filter(e => e.type === 'special')
  | take(5);

const stream = await result.run();
for await (const event of stream) {
  console.log(event); // Logs first 5 special events
}

Mouse Event Tracking:

// Track drag movements between mousedown and mouseup
const events = [
  { type: 'mousedown', x: 10, y: 10 },
  { type: 'mousemove', x: 15, y: 15 },
  { type: 'mousemove', x: 20, y: 20 },
  { type: 'mouseup', x: 20, y: 20 },
];

let isDragging = false;
const trackDrag = e => {
  if (e.type === 'mousedown') isDragging = true;
  if (e.type === 'mouseup') isDragging = false;
  return isDragging && e.type === 'mousemove';
};

const result = pipe(eventStream(events))
  | filter(trackDrag)
  | map(e => ({ x: e.x, y: e.y }));

const dragPositions = await collect(await result.run());
// Returns: [{ x: 15, y: 15 }, { x: 20, y: 20 }]

Double-Click Detection:

// Detect clicks within a time window using scan
const trackDoubleClicks = (state, event) => {
  if (event.type !== 'click') return { lastClick: null, isDouble: false };
  
  const timeDiff = state.lastClick ? event.time - state.lastClick.time : Infinity;
  const isDouble = timeDiff < 250; // 250ms threshold
  
  return { lastClick: event, isDouble, event: isDouble ? event : null };
};

const result = pipe(eventStream(clickEvents))
  | scan(trackDoubleClicks, { lastClick: null, isDouble: false })
  | filter(state => state.isDouble)
  | map(state => state.event);

Files Added

• stream.js - Core stream processing functions and helpers
• stream.test.js - 15 comprehensive tests covering all stream functionality
• examples.js - 5 practical examples demonstrating stream capabilities
• frp-demo.js - Comprehensive FRP demonstration with mouse events, double-clicks, and system monitoring

Files Modified

• README.md - Merged documentation for both stream processing (sections E, F) and composable pipes (section D from main)
• index.js - Includes composable pipes support from main branch
• test.js - Includes 3 new composable pipes tests from main branch
• package.json - Updated to include stream tests and exports

Documentation

Updated README.md with extensive documentation including:

• Stream function API reference (sections E and F)
• FRP examples with async generators
• Mouse event processing patterns
• Guide for working with endless streams
• Composable pipes example (section D, from main branch)

Testing

All tests pass (40 total):

• ✅ 22 original tests (unchanged)
• ✅ 15 new stream tests including:
  • Basic stream operations
  • Endless stream processing with take()
  • Mouse drag and click composables
  • Async operations in streams
  • Stateful stream processing with scan()
• ✅ 3 composable pipes tests (from main branch merge)

Merge Notes

This PR has been successfully merged with the main branch. The merge resolved conflicts in README.md by:

• Combining both stream processing examples (this PR) and composable pipes example (from main)
• Reorganizing README sections to include both feature sets
• Incorporating enhanced index.js with composable pipes support
• All features work together seamlessly with no conflicts

Compatibility

• No breaking changes to existing API
• Works seamlessly with existing pipe and asPipe functions
• Requires Node.js >=18.0.0 (already specified)
• Stream module available as aspipes/stream export

This implementation enables aspipes to be used as a functional reactive programming library, perfect for processing event streams, mouse/keyboard interactions, and any scenario requiring stateful stream transformations. The composable pipes feature from main further enhances reusability and abstraction.

Original prompt

Stream support- Add tests with generators to se if we can use aspipes as functional reactive programming- meaning taking an endless stream of events and pipe it to a pipeline that waits for a particular event. Create a few generator aspipe functions such as map, reduce, scan and filter and take. And make sure we can use them on an endless stream of events, maybe a mousedrag or mouseclick composable pipe.

---

✨ Let Copilot coding agent set things up for you — coding agent works faster and does higher quality work when set up for your repo.

[irony]

@copilot resolve the conflicts

[Copilot]

@copilot resolve the conflicts

Conflicts resolved in commit df8ec87. The merge combines both the stream support features from this PR with the composable pipes features from main. All 40 tests pass.

[Copilot]

Pull Request Overview

This PR adds comprehensive stream support to aspipes, enabling functional reactive programming (FRP) patterns with async generators. The implementation allows processing endless event streams and waiting for particular events using composable pipeline operations, successfully merging with existing composable pipes functionality from the main branch.

Key changes:

• Added five new generator-based aspipe functions (map, filter, take, scan, reduce) for stream processing
• Implemented comprehensive testing suite with 15 new tests covering stream operations and FRP patterns
• Created extensive documentation and examples demonstrating mouse event tracking, double-click detection, and system monitoring patterns

Reviewed Changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
stream.js	Core stream processing functions with async generator support for FRP
stream.test.js	Comprehensive test suite covering all stream functionality and reactive patterns
package.json	Updated exports, scripts, and keywords to include stream module support
frp-demo.js	Comprehensive FRP demonstration with practical examples
examples.js	Five practical examples showing stream capabilities
README.md	Enhanced documentation with stream processing sections E and F

---

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

[Copilot]

[nitpick] The assignment within parentheses pattern (result = pipe(...)) is used consistently throughout the test file but creates unnecessary complexity. Consider separating the assignment from the pipe operation for better readability: result = pipe(numbers()); result | map(x => x * 2);

[Copilot]

[nitpick] Same assignment-in-parentheses pattern as previous comment. This pattern is repeated throughout the entire test file and reduces code clarity.

[Copilot]

The logic for handling the first item when initialValue is undefined is duplicated between scan and reduce functions. Consider extracting this into a shared helper function to reduce code duplication.

[Copilot]

This is the exact same logic as in the scan function above. The duplication should be addressed by extracting the common pattern.