Monad structure

[devphilip21]

Summary

Add monad structure to Stream with flatMap, switchMap, and exhaustMap operators for async stream composition.

Changes

New Operators

Operator	Description	Use Case
flatMap (alias: mergeMap)	Merges all inner streams concurrently	Parallel processing, order doesn't matter
switchMap	Switches to new stream, cancels previous	Search autocomplete, latest selection
exhaustMap	Ignores new signals while inner stream active	Prevent duplicate submissions, drag sessions

Files Added

• src/operators/flat-map.ts - flatMap/mergeMap implementation
• src/operators/switch-map.ts - switchMap implementation
• src/operators/exhaust-map.ts - exhaustMap implementation
• src/operators/flat-map.spec.ts - flatMap tests (including monad laws)
• src/operators/switch-map.spec.ts - switchMap tests
• src/operators/exhaust-map.spec.ts - exhaustMap tests

Files Modified

• src/operators/index.ts - Added exports for new operators

Implementation Details

flatMap (mergeMap)

import { flatMap } from "cereb/operators";

// All inner streams are active concurrently
source$.pipe(
  flatMap(signal => fetchData(signal.value.id))
);

• Subscribes to multiple inner streams concurrently
• Completes only when source and all inner streams complete
• Cleans up all active inner subscriptions on unsubscribe

switchMap

import { switchMap } from "cereb/operators";

// Only show results for the latest search query
searchInput$.pipe(
  switchMap(signal => fetchResults(signal.value.query))
);

• Unsubscribes from previous inner stream when new signal arrives
• Only the most recent inner stream is active at any time
• Automatically handles race conditions

exhaustMap

import { exhaustMap } from "cereb/operators";

// Ignore additional clicks while form is submitting
submitButton$.pipe(
  exhaustMap(signal => submitForm(signal.value))
);

• Ignores new signals while an inner stream is active
• Processes next signal only after inner stream completes
• Ideal for preventing duplicate requests

Monad Laws Verification

Tests verify that flatMap satisfies the three monad laws:

1. Left Identity: pure(a).flatMap(f) === f(a)
2. Right Identity: m.flatMap(pure) === m
3. Associativity: m.flatMap(f).flatMap(g) === m.flatMap(x => f(x).flatMap(g))

Test Coverage

• 27 test cases added
• flatMap: 10 tests (basic behavior + 3 monad laws)
• switchMap: 8 tests
• exhaustMap: 9 tests

Test Scenarios

• Basic mapping and flattening
• Concurrent inner stream handling
• Index passing to project function
• Error handling (project function, inner stream)
• Cleanup on unsubscribe
• Completion signal propagation
• Async stream handling

Breaking Changes

None. This PR only adds new operators.

Migration Guide

No changes required for existing code. New operators are opt-in.

// Before: Manual inner stream management
let currentUnsub: Unsubscribe | null = null;
source$.on(signal => {
  if (currentUnsub) currentUnsub();
  currentUnsub = innerStream(signal).on(/* ... */);
});

// After: Using switchMap
source$.pipe(
  switchMap(signal => innerStream(signal))
).on(/* ... */);

Checklist

• flatMap implementation
• switchMap implementation
• exhaustMap implementation
• Monad laws tests
• Error handling tests
• Cleanup tests
• Async stream tests
• Export from index.ts
• All tests passing (109 tests)

Related

• Spec: 001-monad-structure.md