git.iko.soy/rowtype-yoga/purescript-yoga-om-strom
on main help sign in
rowtype-yoga/purescript-yoga-om-strom/README.md
444 lines 15 kB history raw
1# yoga-om-strom
3**Strom** is a powerful, ZIO ZStream-inspired streaming library for yoga-om that brings pull-based, composable, and resource-safe stream processing to PureScript.
5## Why "Strom"?
7Just as **Om** represents your effect type, **Strom** represents streaming effects. The name comes from German/Swedish meaning "stream" or "current" - fitting for a streaming library!
9## Features
11### 🎯 ZIO ZStream-Inspired
13Strom brings the best ideas from Scala's ZIO ZStream to PureScript:
15- **Pull-based streams** - Efficient, backpressure-aware processing
16- **Chunked execution** - Batch processing for optimal performance
17- **Resource safety** - Proper cleanup and error handling
18- **Type-safe** - Leverages Om's context and error tracking
20### 🚀 Comprehensive Operations
22- **Construction**: `fromArray`, `range`, `iterate`, `repeat`, `unfold`, `repeatOm`
23- **Transformation**: `map`, `mapM`, `mapParallel`, `bind` (or `>>=`), `scan`, `mapAccum`
24- **Selection**: `take`, `takeWhile`, `takeUntil`, `drop`, `dropWhile`, `filter`, `collect`, `changes`
25- **Combination**: `append`, `concat`, `merge`, `zip`, `zipWith`, `interleave`, `race`
26- **Grouping**: `grouped`, `chunked`, `partition`
27- **Execution**: `runCollect`, `runDrain`, `runFold`, `traverse_`, `for_`, `subscribe`
28- **Error Handling**: `catchAll`, `orElse`
30## Installation
32```bash
33spago install yoga-om-core yoga-om-strom
34```
36## Quick Start
38### Basic Stream Operations
40```purescript
41import Yoga.Om.Strom as Strom
43-- Simple transformation pipeline
44result <-
45 Strom.range 1 10
46 # Strom.map (_ * 2)
47 # Strom.filter (_ > 10)
48 # Strom.runCollect
50-- Output: [12, 14, 16, 18]
51```
53### Do-Notation (Comprehension-Style!)
55```purescript
56import Yoga.Om.Strom.Do (guard)
58-- Pythagoras triples
59triples <-
60 (do
61 a <- Strom.range 1 20
62 b <- Strom.range a 20
63 c <- Strom.range b 20
64 guard (a * a + b * b == c * c)
65 pure (Tuple a (Tuple b c))
66 ) # Strom.runCollect
68-- Output: [(3,4,5), (5,12,13), (6,8,10), (8,15,17), ...]
69```
71**See [DO_NOTATION.md](./DO_NOTATION.md) for comprehensive examples!**
73### Effectful Operations
75```purescript
76-- Map with monadic effects
77users <-
78 Strom.range 1 100
79 # Strom.mapM (\id -> fetchUser id) -- Om ctx err User
80 # Strom.filter (_.active)
81 # Strom.runCollect
82```
84### Parallel Processing
86```purescript
87-- Process up to 10 items concurrently
88results <-
89 Strom.fromArray urls
90 # Strom.mapParallel 10 fetchUrl
91 # Strom.runCollect
92```
94### Stateful Transformations
96```purescript
97-- Running total with scan
98totals <-
99 Strom.fromArray [1, 2, 3, 4, 5]
100 # Strom.scan (+) 0
101 # Strom.runCollect
102-- Output: [1, 3, 6, 10, 15]
104-- Stateful map with accumulator
105indexed <-
106 Strom.fromArray ["a", "b", "c"]
107 # Strom.mapAccum (\i x -> Tuple (i + 1) (show i <> ": " <> x)) 1
108 # Strom.runCollect
109-- Output: ["1: a", "2: b", "3: c"]
110```
112### Combining Streams
114```purescript
115-- Zip two streams
116pairs <-
117 Strom.zip
118 (Strom.range 1 5)
119 (Strom.fromArray ["a", "b", "c", "d", "e"])
120 # Strom.runCollect
121-- Output: [(1, "a"), (2, "b"), (3, "c"), (4, "d"), (5, "e")]
123-- Interleave deterministically
124mixed <-
125 Strom.interleave
126 (Strom.fromArray [1, 3, 5])
127 (Strom.fromArray [2, 4, 6])
128 # Strom.runCollect
129-- Output: [1, 2, 3, 4, 5, 6]
131-- Merge non-deterministically (race-based)
132merged <-
133 Strom.merge stream1 stream2
134 # Strom.runCollect
135```
137### Infinite Streams
139```purescript
140-- Take from infinite stream
141first10 <-
142 Strom.iterate (_ + 1) 0
143 # Strom.take 10
144 # Strom.runCollect
145-- Output: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
147-- Fibonacci sequence
148fibonacci <-
149 Strom.unfold
150 (\(Tuple a b) -> Just (Tuple a (Tuple b (a + b))))
151 (Tuple 0 1)
152 # Strom.take 10
153 # Strom.runCollect
154-- Output: [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
155```
157### Grouping and Batching
159```purescript
160-- Process in batches
161Strom.range 1 100
162 # Strom.grouped 10
163 # Strom.traverse_ (\batch -> do
164 Console.log $ "Processing batch of " <> show (length batch)
165 processBatch batch
166 )
167```
169### Error Handling
171```purescript
172-- Catch and recover
173results <-
174 riskyStream
175 # Strom.catchAll (\err -> Strom.succeed defaultValue)
176 # Strom.runCollect
178-- Provide alternative stream
179results <-
180 primaryStream `Strom.orElse` fallbackStream
181 # Strom.runCollect
182```
184## Real-World Example
186### Event Processing Pipeline
188```purescript
189type Event = { timestamp :: Int, userId :: String, action :: String }
190type Context = { logger :: String -> Aff Unit, db :: Database }
192processEvents :: Om Context () Unit
193processEvents = do
194 fetchEventStream
195 # Strom.filter (_.action /= "spam") -- Filter spam
196 # Strom.changes -- Deduplicate consecutive
197 # Strom.mapM enrichEvent -- Fetch additional data
198 # Strom.mapParallel 5 validateEvent -- Validate concurrently
199 # Strom.collect identity -- Keep only valid (Maybe)
200 # Strom.grouped 50 -- Batch for DB
201 # Strom.tapM (\batch -> logBatch batch) -- Log each batch
202 # Strom.traverse_ (\batch -> saveBatch batch) -- Save to DB
203```
205### Paginated API Fetching
207```purescript
208-- Unfold paginated results into a stream
209fetchAllPages :: Om Context () (Array Item)
210fetchAllPages = do
211 Strom.unfoldOm (\pageToken -> do
212 page <- fetchPage pageToken
213 case page.nextToken of
214 Nothing -> pure Nothing
215 Just token -> pure $ Just $ Tuple page.items token
216 ) initialToken
217 >>= Strom.fromArray -- Flatten pages
218 # Strom.runCollect
219```
221### Rate-Limited API Calls
223```purescript
224-- Process with delays between items
225processWithRateLimit :: Om Context () Unit
226processWithRateLimit = do
227 Strom.fromArray items
228 # Strom.mapM (\item -> do
229 result <- callAPI item
230 Om.delay (Milliseconds 100.0) -- Rate limit
231 pure result
232 )
233 # Strom.traverse_ handleResult
234```
236## Comparison with Other Approaches
238### vs Arrays
240```purescript
241-- ❌ Arrays: Load everything into memory
242processArrays :: Om ctx err (Array Result)
243processArrays = do
244 items <- fetchAllItems -- Loads ALL items
245 Array.traverse processItem items -- Sequential
247-- ✅ Strom: Efficient streaming
248processStream :: Om ctx err Unit
249processStream = do
250 Strom.unfoldOm fetchNextBatch initialState
251 >>= Strom.fromArray
252 # Strom.mapParallel 10 processItem -- Parallel!
253 # Strom.runDrain
254```
256### vs Bolson Events
258```purescript
259-- Bolson Events are push-based (FRP)
260-- Great for UI, but less control over backpressure
262-- Strom is pull-based
263-- Consumer controls when to pull next items
264-- Better for batch processing, APIs, file I/O
265```
267## API Overview
269### Construction
271- `empty :: Strom ctx err a` - Empty stream
272- `succeed :: a -> Strom ctx err a` - Single element
273- `fromArray :: Array a -> Strom ctx err a` - From array
274- `fromFoldable :: Foldable f => f a -> Strom ctx err a` - From any foldable
275- `fromOm :: Om ctx err a -> Strom ctx err a` - From Om effect
276- `fromAff :: Aff a -> Strom ctx err a` - From Aff
277- `range :: Int -> Int -> Strom ctx err Int` - Range of integers
278- `iterate :: (a -> a) -> a -> Strom ctx err a` - Infinite iteration
279- `repeat :: a -> Strom ctx err a` - Infinite repetition
280- `repeatOm :: Om ctx err a -> Strom ctx err a` - Infinite Om repetition
281- `unfold :: (b -> Maybe (Tuple a b)) -> b -> Strom ctx err a` - Unfold pattern
282- `unfoldOm :: (b -> Om ctx err (Maybe (Tuple a b))) -> b -> Strom ctx err a` - Effectful unfold
284### Running
286- `runCollect :: Strom ctx err a -> Om ctx err (Array a)` - Collect all elements
287- `runDrain :: Strom ctx err a -> Om ctx err Unit` - Run and discard
288- `runFold :: b -> (b -> a -> b) -> Strom ctx err a -> Om ctx err b` - Fold stream
289- `traverse_ :: (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit` - Traverse with effects
290- `for_ :: Strom ctx err a -> (a -> Om ctx err Unit) -> Om ctx err Unit` - Flipped traverse_
291- `subscribe :: (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err (Om ctx err Unit)` - Subscribe with cancellation
293### Transformations
295- `map :: (a -> b) -> Strom ctx err a -> Strom ctx err b` - Pure map
296- `mapM :: (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b` - Monadic map
297- `mapParallel :: Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b` - Parallel map
298- `bind :: (a -> Strom ctx err b) -> Strom ctx err a -> Strom ctx err b` - Monadic bind (use `>>=` operator)
299- `scan :: (b -> a -> b) -> b -> Strom ctx err a -> Strom ctx err b` - Running fold
300- `mapAccum :: (s -> a -> Tuple s b) -> s -> Strom ctx err a -> Strom ctx err b` - Stateful map
301- `tap :: (a -> Unit) -> Strom ctx err a -> Strom ctx err a` - Observe without modifying
302- `tapM :: (a -> Om ctx err Unit) -> Strom ctx err a -> Strom ctx err a` - Monadic tap
304### Selection
306- `take :: Int -> Strom ctx err a -> Strom ctx err a` - Take n elements
307- `takeWhile :: (a -> Boolean) -> Strom ctx err a -> Strom ctx err a` - Take while true
308- `takeUntil :: (a -> Boolean) -> Strom ctx err a -> Strom ctx err a` - Take until true (inclusive)
309- `drop :: Int -> Strom ctx err a -> Strom ctx err a` - Drop n elements
310- `dropWhile :: (a -> Boolean) -> Strom ctx err a -> Strom ctx err a` - Drop while true
311- `filter :: (a -> Boolean) -> Strom ctx err a -> Strom ctx err a` - Filter elements
312- `filterM :: (a -> Om ctx err Boolean) -> Strom ctx err a -> Strom ctx err a` - Monadic filter
313- `collect :: (a -> Maybe b) -> Strom ctx err a -> Strom ctx err b` - Filter + map
314- `collectM :: (a -> Om ctx err (Maybe b)) -> Strom ctx err a -> Strom ctx err b` - Monadic collect
315- `changes :: Eq a => Strom ctx err a -> Strom ctx err a` - Remove consecutive duplicates
317### Combining
319- `append :: Strom ctx err a -> Strom ctx err a -> Strom ctx err a` - Sequential concatenation
320- `concat :: Array (Strom ctx err a) -> Strom ctx err a` - Concat multiple streams
321- `merge :: Strom ctx err a -> Strom ctx err a -> Strom ctx err a` - Non-deterministic merge
322- `zip :: Strom ctx err a -> Strom ctx err b -> Strom ctx err (Tuple a b)` - Zip two streams
323- `zipWith :: (a -> b -> c) -> Strom ctx err a -> Strom ctx err b -> Strom ctx err c` - Zip with function
324- `interleave :: Strom ctx err a -> Strom ctx err a -> Strom ctx err a` - Deterministic interleave
325- `race :: Array (Strom ctx err a) -> Strom ctx err a` - Race multiple streams
327### Grouping
329- `grouped :: Int -> Strom ctx err a -> Strom ctx err (Array a)` - Fixed-size chunks
330- `chunked :: Int -> Strom ctx err a -> Strom ctx err (Array a)` - Alias for grouped
331- `partition :: (a -> Boolean) -> Strom ctx err a -> Tuple (Strom ctx err a) (Strom ctx err a)` - Split by predicate
333### Error Handling
335- `catchAll :: (Record err -> Strom ctx err2 a) -> Strom ctx err a -> Strom ctx err2 a` - Catch and recover
336- `orElse :: Strom ctx err a -> Strom ctx err a -> Strom ctx err a` - Alternative on failure
338## Performance Characteristics
340- **Chunked processing**: Operations process arrays internally for efficiency
341- **Pull-based**: Backpressure naturally handled - consumer controls pace
342- **Lazy**: Elements only computed when pulled
343- **Resource-safe**: Om's error handling ensures cleanup
345### Benchmark Results (Node.js v25)
347Performance benchmarks on a modern system (median of 10 runs):
349| Operation | Dataset | Median Time |
350|-----------|---------|-------------|
351| **Simple Ops** | | |
352| map | 2M elements | <1ms |
353| filter (50%) | 2M elements | <1ms |
354| map chain (3x) | 2M elements | <1ms |
355| pipeline (map+filter+map) | 2M elements | <1ms |
356| **Aggregation** | | |
357| fold (sum) | 1M elements | <1ms |
358| collect | 50k elements | <1ms |
359| scan (running sum) | 2M elements | <1ms |
360| **Construction** | | |
361| fromArray | 1M elements | <1ms |
362| iterateStrom | 10k elements | <1ms |
363| repeatStrom | 10k elements | <1ms |
364| unfold | 10k elements | <1ms |
365| iterateInfinite + take | 10k elements | **1ms** |
366| **Selection** | | |
367| take | 5k from 2M | <1ms |
368| takeWhile | ~50k from 1M | <1ms |
369| drop | 5k from 1M | <1ms |
370| dropWhile | 500k from 1M | <1ms |
371| **Transform** | | |
372| mapM (effect) | 50k elements | <1ms |
373| tap | 50k elements | <1ms |
374| collect (filter+map) | 50k elements | <1ms |
375| changes (dedup) | 50k elements | <1ms |
376| **Grouping** | | |
377| grouped (chunks of 100) | 50k elements | <1ms |
378| partition (even/odd) | 50k elements | <1ms |
379| **Combining** | | |
380| append | 2x500k streams | <1ms |
381| concat | 10x10k streams | <1ms |
382| zip | 2x500k streams | <1ms |
383| zipWith | 2x500k streams | <1ms |
384| bind (flatMap) | 5k×10 | <1ms |
385| **Concurrent (Pure)** | | |
386| mergeND (pure functions) | 2x1M streams | **485ms** ⚠️ |
387| mapPar (pure, concurrency=4) | 5k elements | **355ms** ⚠️ |
388| mergeND (with 1ms delays) | 2x100 elements | **~200ms** ✅ |
389| mapPar (with 10ms delays, conc=4) | 100 elements | **~250ms** ✅ |
391**Key Takeaways:**
392- ⚡ Most operations are **sub-millisecond** thanks to chunked processing and STArray optimizations
393- 🚀 Excellent throughput: **>2M elements/ms** for simple transformations
394- 📦 Efficient chunking (10,000 elements per chunk) minimizes overhead
395- ⚠️ **Concurrent operations** (`mergeND`, `mapPar`) have coordination overhead per chunk/group
396 - With **pure functions**: Overhead dominates (use sequential operations instead!)
397 - With **async I/O**: Parallelism provides huge speedup (e.g., 100 parallel 10ms tasks = 250ms vs 1000ms sequential)
398- 💡 **Use concurrent ops for**: API calls, database queries, file I/O
399- 💡 **Use sequential ops for**: Pure transformations, CPU-bound work
401## Architecture
403Strom is built on a simple but powerful abstraction:
405```purescript
406newtype Strom ctx err a = Strom
407 { pull :: Om ctx err (Step (Strom ctx err a) (Maybe (Chunk a)))
408 }
410type Chunk a = Array a
411```
413Each pull:
4141. Returns a `Step` - either `Done` (finished) or `Loop` (continue)
4152. May emit a `Chunk` of values (or `Nothing` if no values yet)
4163. Can perform Om effects (context access, error handling, etc.)
418This design enables:
419- Efficient batching
420- Natural backpressure
421- Resource safety
422- Composability
424## Documentation
426- **[INDEX.md](./INDEX.md)** - Start here! Navigation and overview
427- **[README.md](./README.md)** - This file (complete guide)
428- **[QUICKREF.md](./QUICKREF.md)** - One-page API reference
429- **[DO_NOTATION.md](./DO_NOTATION.md)** - Do-notation guide with examples
430- **[DEMO.md](./DEMO.md)** - API showcase with usage patterns
431- **[COMPARISON.md](./COMPARISON.md)** - Detailed comparison with ZIO ZStream
432- **[TESTING.md](./TESTING.md)** - Test coverage summary
433- **[Examples.purs](./src/Yoga/Om/Strom/Examples.purs)** - 18 runnable examples
435## See Also
437- **[yoga-om-core](../yoga-om-core)** - Core Om effect system (required)
438- **[yoga-om-node](../yoga-om-node)** - Node.js integrations
439- **[yoga-om-rom](../yoga-om-rom)** - Reactive Om: Bolson FRP integration (complementary, push-based)
441## Credits
443Inspired by [ZIO ZStream](https://zio.dev/reference/stream/zstream) from the Scala ZIO ecosystem.