git.iko.soy/rowtype-yoga/purescript-yoga-om-strom
on v0.1.1 help sign in
rowtype-yoga/purescript-yoga-om-strom/INTEGRATION_GUIDE.md
334 lines 7.9 kB history raw
1# Strom Integration Guide: Merging & Kafka
3Quick visual guide to merging streams and integrating with Kafka.
5## 🔀 Merging Strategies
7### 1. `merge` - Non-Deterministic (Race-Based)
9```purescript
10result = Strom.merge stream1 stream2
11```
13```
14Stream 1: [1, 2, 3] ──┐
15 ├─→ [?, ?, ?, ?, ?, ?] (non-deterministic order)
16Stream 2: [a, b, c] ──┘
17```
19**Use for**: Event streams, multiple sources, max throughput
21---
23### 2. `append` / `concat` - Sequential
25```purescript
26result = stream1 `Strom.append` stream2
27-- or
28result = Strom.concat [stream1, stream2, stream3]
29```
31```
32Stream 1: [1, 2, 3] ──┐
33 ├─→ [1, 2, 3, a, b, c] (deterministic)
34Stream 2: [a, b, c] ──┘
35 (waits for stream1)
36```
38**Use for**: Pagination, stages, fallbacks
40---
42### 3. `interleave` - Deterministic Alternating
44```purescript
45result = Strom.interleave stream1 stream2
46```
48```
49Stream 1: [1, 2, 3] ──┐
50 ├─→ [1, a, 2, b, 3, c] (alternates)
51Stream 2: [a, b, c] ──┘
52```
54**Use for**: Fair scheduling, round-robin, load balancing
56---
58### 4. `zip` - Pair Elements
60```purescript
61result = Strom.zip stream1 stream2
62```
64```
65Stream 1: [1, 2, 3] ──┐
66 ├─→ [(1,a), (2,b), (3,c)]
67Stream 2: [a, b, c] ──┘
68```
70**Use for**: Joining related data, adding indices
72---
74### 5. `race` - First Wins
76```purescript
77result = Strom.race [stream1, stream2, stream3]
78```
80```
81Stream 1: [slow...] ──┐
82 │
83Stream 2: [FAST!] ────├─→ [FAST!] (first to emit wins)
84 │
85Stream 3: [slower] ────┘
86```
88**Use for**: Timeouts, fastest response, circuit breakers
90---
92### 6. Do-Notation - Cartesian Product
94```purescript
95result = do
96 x <- stream1
97 y <- stream2
98 pure (Tuple x y)
99```
101```
102Stream 1: [1, 2] ──┐
103 ├─→ [(1,a), (1,b), (2,a), (2,b)]
104Stream 2: [a, b] ──┘ (all combinations)
105```
107**Use for**: All combinations, test cases, grids
109---
111## 🎯 Quick Decision Tree
113```
114Need to merge streams?
115│
116├─ Order matters?
117│ ├─ YES → Use `append`/`concat`
118│ └─ NO → Continue...
119│
120├─ Need all combinations?
121│ ├─ YES → Use do-notation
122│ └─ NO → Continue...
123│
124├─ Need pairing?
125│ ├─ YES → Use `zip`
126│ └─ NO → Continue...
127│
128├─ Want fastest?
129│ ├─ YES → Use `race`
130│ └─ NO → Continue...
131│
132├─ Want fair scheduling?
133│ ├─ YES → Use `interleave`
134│ └─ NO → Use `merge`
135```
137---
139## ☕ Kafka Integration Patterns
141### Pattern 1: Simple Consumer
143```purescript
144kafkaStream config "my-topic"
145 # Strom.mapM processMessage
146 # Strom.traverse_ handleResult
147```
149### Pattern 2: Infinite Consumer with Batching
151```purescript
152kafkaInfinite config "my-topic"
153 # Strom.grouped 100 -- Batch 100
154 # Strom.traverse_ processBatch
155```
157### Pattern 3: Consumer → Producer Pipeline
159```purescript
160producer <- createProducer config
162kafkaStream config "input"
163 # Strom.mapParallel 10 transform
164 # Strom.grouped 50
165 # Strom.traverse_ (\batch ->
166 producer.sendBatch "output" batch
167 )
168```
170### Pattern 4: Multi-Topic Consumer
172```purescript
173-- Option A: Sequential
174Strom.concat
175 [ kafkaStream config "topic1"
176 , kafkaStream config "topic2"
177 , kafkaStream config "topic3"
178 ]
179 # Strom.traverse_ process
181-- Option B: With tagging
182Strom.concat
183 [ kafkaStream config "t1" # Strom.map (\m -> {topic: "t1", msg: m})
184 , kafkaStream config "t2" # Strom.map (\m -> {topic: "t2", msg: m})
185 ]
186 # Strom.traverse_ routeByTopic
187```
189### Pattern 5: Error Handling with DLQ
191```purescript
192producer <- createProducer config
194source
195 # Strom.mapM (\msg ->
196 processMessage msg
197 # Om.catchAll (\err -> do
198 producer.send "dlq-topic" msg
199 pure Nothing
200 )
201 )
202 # Strom.collect identity
203 # Strom.runDrain
204```
206### Pattern 6: Exactly-Once Semantics
208```purescript
209consumer <- createConsumer config
210producer <- createProducer config
212Strom.repeatOm do
213 batch <- consumer.consume 5000
214 processed <- Array.traverse transform batch
215 producer.sendBatch "output" processed
216 consumer.commit -- After successful send
217 pure batch
218>>= Strom.fromArray
219# Strom.runDrain
220```
222---
224## 🚀 Real-World: Complete ETL Pipeline
226```purescript
227-- Kafka → Transform → Enrich → Validate → Kafka
228kafkaETL :: Om Context () Unit
229kafkaETL = do
230 producer <- createProducer producerConfig
232 kafkaInfinite consumerConfig "raw-events"
233 # Strom.mapM parseEvent -- Parse JSON
234 # Strom.collect identity -- Filter failures
235 # Strom.filter isValidEvent -- Business validation
236 # Strom.mapParallel 10 enrichWithUserData -- Parallel enrichment
237 # Strom.filter (\e -> e.user.active) -- Filter inactive
238 # Strom.grouped 100 -- Batch
239 # Strom.tapM logBatchStats -- Monitor
240 # Strom.traverse_ (\batch -> do -- Send & commit
241 producer.sendBatch "enriched-events" batch
242 commitOffsets batch
243 )
244```
246**With Do-Notation for Multiple Topics:**
248```purescript
249import Yoga.Om.Strom.Do (guard)
251multiTopicETL :: Om Context () Unit
252multiTopicETL = do
253 producer <- createProducer config
255 event <- do
256 -- Merge 3 topics with filtering
257 msg <- Strom.concat
258 [ kafkaStream config "events-v1"
259 , kafkaStream config "events-v2"
260 , kafkaStream config "events-v3"
261 ]
263 -- Parse based on topic
264 parsed <- Strom.fromOm $ parseByTopic msg
265 guard (isValid parsed)
266 pure parsed
268 -- Enrich
269 enriched <- Strom.fromOm $ enrichEvent event
270 guard enriched.isComplete
272 -- Send to output
273 _ <- Strom.fromOm $
274 Om.fromAff $ producer.send "output" (serialize enriched)
276 pure unit
277```
279---
281## 📊 Comparison: Merge Methods
283| Method | Order | Speed | Memory | Best For |
284|--------|-------|-------|--------|----------|
285| `merge` | ❌ | ⚡⚡⚡ | 💾 | High throughput |
286| `append` | ✅ | ⚡ | 💾 | Stages |
287| `concat` | ✅ | ⚡ | 💾 | Multiple stages |
288| `interleave` | ✅ | ⚡⚡ | 💾💾 | Fair scheduling |
289| `zip` | ✅ | ⚡⚡ | 💾💾 | Pairing |
290| `race` | ❌ | ⚡⚡⚡ | 💾 | Fastest wins |
291| Do-notation | ✅ | ⚡ | 💾💾💾 | All combos |
293---
295## 🎓 Key Takeaways
297### Merging Streams
2991. **5 merge strategies** - each with different semantics
3002. **Choose based on needs** - order, speed, semantics
3013. **Composable** - can combine strategies
3024. **Type-safe** - errors tracked through merges
304### Kafka Integration
3061. **Simple pattern** - `unfoldOm` for polling
3072. **Natural backpressure** - pull-based Strom throttles automatically
3083. **Batching** - use `grouped` for efficiency
3094. **Parallel** - use `mapParallel` for throughput
3105. **Error handling** - use `catchAll` and DLQ patterns
3116. **Exactly-once** - read → process → send → commit
313### Best Practices
315✅ **Batch messages** - Group for efficiency
316✅ **Use parallel processing** - `mapParallel` for I/O
317✅ **Handle errors gracefully** - DLQ pattern
318✅ **Commit after processing** - Exactly-once semantics
319✅ **Monitor throughput** - Log batch sizes
320✅ **Test with mocks** - Replace Kafka with `fromArray`
322---
324## 📚 See Also
326- **[MERGING.md](./MERGING.md)** - Detailed merge strategies
327- **[KAFKA_INTEGRATION.md](./KAFKA_INTEGRATION.md)** - Complete Kafka guide
328- **[examples/MergingShowcase.purs](./examples/MergingShowcase.purs)** - All patterns
329- **[examples/KafkaIntegration.purs](./examples/KafkaIntegration.purs)** - Kafka examples
330- **[examples/KafkaIntegration.js](./examples/KafkaIntegration.js)** - FFI implementation
332---
334**Strom makes stream merging and Kafka integration simple and type-safe!** 🌊