git.iko.soy/rowtype-yoga/purescript-yoga-om-strom
on main help sign in
rowtype-yoga/purescript-yoga-om-strom/DEMO.md
383 lines 9.6 kB history raw
1# Strom API Demo
3This document showcases the Strom API design and usage patterns.
5## Core Design Philosophy
7Strom brings ZIO ZStream's powerful streaming model to PureScript's Om effect system:
91. **Pull-based** - Consumer controls the pace (backpressure-aware)
102. **Chunked** - Processes data in batches for efficiency
113. **Type-safe** - Leverages Om's context (`ctx`) and error (`err`) tracking
124. **Resource-safe** - Integrates with Om's error handling
14## Type Signature
16```purescript
17newtype Strom ctx err a = Strom
18 { pull :: Om ctx err (Step (Strom ctx err a) (Maybe (Chunk a)))
19 }
21type Chunk a = Array a
22```
24Each `pull` operation:
25- Returns a `Step` (either `Done` or `Loop` to continue)
26- May emit a chunk of values or `Nothing`
27- Can perform Om effects with context and error handling
29## API Showcase
31### 1. Construction - Building Streams
33```purescript
34-- Empty stream
35empty :: Strom ctx err a
37-- Single value
38succeed 42 :: Strom ctx err Int
40-- From collections
41fromArray [1, 2, 3, 4, 5] :: Strom ctx err Int
42fromFoldable (List.fromFoldable [1, 2, 3]) :: Strom ctx err Int
44-- From effects
45fromOm (fetchUser 123) :: Strom ctx err User
46fromAff httpCall :: Strom ctx err Response
48-- Ranges and sequences
49range 1 100 :: Strom ctx err Int
50iterate (_ + 1) 0 :: Strom ctx err Int -- infinite!
51repeat "hello" :: Strom ctx err String -- infinite!
53-- Unfold patterns
54unfold (\n -> if n > 10 then Nothing else Just (Tuple n (n + 1))) 0
55unfoldOm fetchNextPage initialToken -- paginated APIs!
57-- Repeat with effects
58repeatOm (fetchLatestData) :: Strom ctx err Data
59```
61### 2. Running - Consuming Streams
63```purescript
64-- Collect all elements
65runCollect :: Strom ctx err a -> Om ctx err (Array a)
66result <- Strom.range 1 10 # Strom.runCollect
67-- [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
69-- Fold (reduce)
70runFold :: b -> (b -> a -> b) -> Strom ctx err a -> Om ctx err b
71sum <- Strom.range 1 100 # Strom.runFold 0 (+)
72-- 5050
74-- Run for effects, discard values
75runDrain :: Strom ctx err a -> Om ctx err Unit
76Strom.repeat doSomething # Strom.take 10 # Strom.runDrain
78-- Process each element
79traverse_ :: (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit
80events # Strom.traverse_ handleEvent
82-- Subscribe (returns cancellation)
83subscribe :: (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err (Om ctx err Unit)
84cancel <- Strom.subscribe logEvent eventStream
85-- later: cancel
86```
88### 3. Transformations - Modifying Streams
90```purescript
91-- Pure map
92Strom.range 1 10
93 # Strom.map (_ * 2)
94 # Strom.map show
95-- ["2", "4", "6", ..., "20"]
97-- Effectful map
98users
99 # Strom.mapM enrichUserData
100 # Strom.mapM validateUser
102-- Parallel map (up to N concurrent)
103urls
104 # Strom.mapParallel 10 fetchUrl
105 # Strom.runCollect
107-- Bind (nested streams, use >>= operator)
108departments
109 >>= (\dept -> Strom.fromArray dept.employees)
111-- Scan (running aggregation, emits intermediates)
112Strom.fromArray [1, 2, 3, 4, 5]
113 # Strom.scan (+) 0
114-- [1, 3, 6, 10, 15]
116-- Stateful map with accumulator
117items
118 # Strom.mapAccum (\index item -> Tuple (index + 1) { index, item }) 0
120-- Tap (observe without modifying)
121stream
122 # Strom.tap (\x -> log x) -- pure observation
123 # Strom.tapM (\x -> saveToDb x) -- effectful
124```
126### 4. Selection - Filtering Streams
128```purescript
129-- Take n elements
130Strom.iterate (_ + 1) 0
131 # Strom.take 10
132-- [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
134-- Take while predicate
135numbers
136 # Strom.takeWhile (_ < 100)
138-- Take until predicate (inclusive)
139numbers
140 # Strom.takeUntil (_ == 42)
142-- Drop operations
143numbers
144 # Strom.drop 5
145 # Strom.dropWhile (_ < 10)
147-- Filter
148numbers
149 # Strom.filter (_ `mod` 2 == 0)
151-- Effectful filter
152users
153 # Strom.filterM isActiveUser
155-- Collect (filter + map in one)
156strings
157 # Strom.collect parseInt -- :: String -> Maybe Int
158-- Only successful parses!
160-- Remove consecutive duplicates
161events
162 # Strom.changes
163```
165### 5. Combination - Merging Streams
167```purescript
168-- Sequential concatenation
169stream1 `Strom.append` stream2
170Strom.concat [s1, s2, s3, s4]
172-- Non-deterministic merge (race-based)
173Strom.merge fastStream slowStream
174-- Elements appear as they're produced
176-- Zip (pairs elements)
177Strom.zip
178 (Strom.range 1 5)
179 (Strom.fromArray ["a", "b", "c", "d", "e"])
180-- [(1, "a"), (2, "b"), (3, "c"), (4, "d"), (5, "e")]
182-- Zip with function
183Strom.zipWith (\_) stream1 stream2
185-- Deterministic interleave
186Strom.interleave odds evens
187-- Takes one from each alternately
189-- Race (first to emit wins)
190Strom.race [source1, source2, source3]
191```
193### 6. Grouping - Batching Streams
195```purescript
196-- Fixed-size chunks
197Strom.range 1 100
198 # Strom.grouped 10
199-- [[1..10], [11..20], ..., [91..100]]
201-- Chunked (alias for grouped)
202events
203 # Strom.chunked 50
204 # Strom.traverse_ processBatch
206-- Partition (split by predicate)
207let Tuple evens odds = Strom.partition (\n -> n `mod` 2 == 0) numbers
208evensResult <- Strom.runCollect evens
209oddsResult <- Strom.runCollect odds
210```
212### 7. Error Handling
214```purescript
215-- Catch and recover
216riskyStream
217 # Strom.catchAll (\err -> Strom.succeed fallbackValue)
219-- Alternative on failure
220primaryStream `Strom.orElse` backupStream
221```
223## Real-World Example: Event Processing Pipeline
225```purescript
226type Event = { id :: String, userId :: String, action :: String, timestamp :: Int }
227type Context = { logger :: String -> Aff Unit, db :: Database }
229processEventStream :: Om Context () Unit
230processEventStream = do
231 eventSource
232 # Strom.filter (\e -> e.action /= "spam")
233 # Strom.changes
234 # Strom.mapM enrichEvent
235 # Strom.mapParallel 5 validateEvent
236 # Strom.collect identity
237 # Strom.mapAccum
238 (\seq event -> Tuple (seq + 1) { seq, event })
239 0
240 # Strom.tapM (\{ seq, event } -> logEvent seq event)
241 # Strom.grouped 50
242 # Strom.traverse_ saveBatch
243```
245## Real-World Example: Paginated API
247```purescript
248type Page = { items :: Array Item, nextToken :: Maybe String }
250fetchAllPages :: Om Context () (Array Item)
251fetchAllPages = do
252 Strom.unfoldOm (\token -> do
253 page <- fetchPage token
254 case page.nextToken of
255 Nothing -> pure Nothing
256 Just next -> pure $ Just $ Tuple page.items next
257 ) initialToken
258 >>= Strom.fromArray
259 # Strom.runCollect
260```
262## Real-World Example: Rate-Limited Processing
264```purescript
265processWithRateLimit :: Om Context () Unit
266processWithRateLimit = do
267 items
268 # Strom.mapM (\item -> do
269 result <- callAPI item
270 Om.delay (Milliseconds 100.0) -- Rate limit
271 pure result
272 )
273 # Strom.grouped 10 -- Process in batches
274 # Strom.traverse_ logBatch
275```
277## Comparison: Strom vs Arrays
279```purescript
280-- ❌ Arrays: Everything in memory
281items <- fetchAllItems -- Could be 1M items!
282results <- Array.traverse processItem items -- Sequential
283saveAll results
285-- ✅ Strom: Stream processing
286fetchAllItemsStream
287 # Strom.mapParallel 10 processItem -- Parallel!
288 # Strom.grouped 100 -- Batch saves
289 # Strom.traverse_ saveBatch -- Incremental
290```
292## Comparison: Strom vs Bolson Events
294- **Bolson Events**: Push-based (FRP), great for UI
295- **Strom**: Pull-based, great for data processing, APIs, I/O
297Both complement each other! You can even convert between them via `yoga-om-streams`.
299## Performance Characteristics
301- **Chunked**: Processes arrays internally (1000 elements by default)
302- **Lazy**: Values computed only when pulled
303- **Backpressure**: Consumer controls pace naturally
304- **Resource-safe**: Om's error handling ensures cleanup
305- **Parallel**: `mapParallel` processes up to N concurrently
307## Advanced Patterns
309### Fibonacci Generator
311```purescript
312fibonacci = Strom.unfold
313 (\(Tuple a b) -> Just (Tuple a (Tuple b (a + b))))
314 (Tuple 0 1)
316first20 <- fibonacci # Strom.take 20 # Strom.runCollect
317```
319### Infinite Data Source with Cancellation
321```purescript
322liveDataFeed :: Om Context () Unit
323liveDataFeed = do
324 cancel <- Strom.repeatOm fetchLatestData
325 # Strom.mapM processData
326 # Strom.subscribe handleData
328 -- Let it run
329 Om.delay (Milliseconds 60000.0)
331 -- Cancel after 1 minute
332 cancel
333```
335### Multi-Source Aggregation
337```purescript
338aggregateFromSources :: Om Context () (Array Result)
339aggregateFromSources = do
340 Strom.race
341 [ sourceA # Strom.map toResult
342 , sourceB # Strom.map toResult
343 , sourceC # Strom.map toResult
344 ]
345 # Strom.take 100
346 # Strom.runCollect
347```
349## Type Safety Benefits
351```purescript
352-- Context is tracked
353fetchUser :: Int -> Om { db :: Database } _ User
355-- Won't compile - missing db in context!
356❌ Strom.fromOm (fetchUser 1) :: Strom {} _ User
358-- ✅ Correct - db provided
359✓ Strom.fromOm (fetchUser 1) :: Strom { db :: Database } _ User
361-- Errors are tracked
362riskyOp :: Om ctx (networkError :: String) Result
364-- Error appears in stream type
365riskyStream :: Strom ctx (networkError :: String) Result
366riskyStream = Strom.fromOm riskyOp
368-- Must handle or propagate
369handled = riskyStream # Strom.catchAll (\_ -> Strom.succeed defaultValue)
370```
372## Summary
374Strom provides a comprehensive, type-safe, and efficient streaming library for Om that:
376✅ **Scales**: Process millions of items without loading into memory
377✅ **Composes**: Rich combinator library for building pipelines
378✅ **Controls**: Pull-based model gives consumer control
379✅ **Parallels**: Easy concurrent processing with `mapParallel`
380✅ **Integrates**: Works seamlessly with Om's context and errors
381✅ **Performs**: Chunked processing for optimal throughput
383Perfect for: Data pipelines, API processing, event streams, file I/O, batch jobs, real-time feeds.