@@ -74,16 +74,29 @@ module Yoga.Om.Strom
7474 , zipWithStrom
7575 , interleave
7676 , intersperse
77+ , HaltStrategy(..)
7778 , merge
7879 , mergeAll
80+ , mergeWith
81+ , mergeHaltEither
82+ , mergeHaltLeft
83+ , mergeHaltRight
7984 , mergeND
85+ , mergeNDWith
8086 , mergeAllND
87+ , mergeAllNDWith
8188 , race
8289 , raceAll
8390 -- Parallel Processing
8491 , mapPar
8592 , mapMPar
8693 , foreachPar
94+ , mapParUnordered
95+ , mapMParUnordered
96+ , foreachParUnordered
97+ -- Buffering
98+ , buffer
99+ , bufferChunks
87100 -- Error Handling
88101 , catchAll
89102 -- , catchSome -- TODO: Fix type constraints
@@ -115,7 +128,7 @@ import Data.Functor (map) as Functor
115128import Data.List as List
116129import Data.Maybe (Maybe(..))
117130import Data.Time.Duration (Milliseconds(..))
118−import Data.Traversable (traverse)
131+import Data.Traversable (traverse, traverse_)
119132import Data.Tuple (Tuple(..))
120133import Data.Variant (Variant)
121134import Effect.Aff (Aff, Error, delay, forkAff, joinFiber, killFiber)
@@ -123,7 +136,10 @@ import Effect.Aff.AVar as AVar
123136import Effect.Aff.Class (liftAff)
124137import Effect.Exception as Exception
125138import Effect.Now (now)
139+import Effect.Ref as Ref
126140import Effect.Class (liftEffect)
141+import Data.CatQueue as CatQueue
142+import Data.CatQueue (CatQueue)
127143import Data.DateTime.Instant (unInstant)
128144import Yoga.Om (Om)
129145import Yoga.Om as Om
@@ -175,6 +191,29 @@ data QueueItem a
175191
176192derive instance functorQueueItem :: Functor QueueItem
177193
194+-- | Messages for mergeAllND buffered queue
195+data MergeAllItem a
196+ = MergeChunk (Array a)
197+ | MergeDone
198+
199+derive instance functorMergeAllItem :: Functor MergeAllItem
200+
201+-- | Merge termination strategy (like ZStream HaltStrategy)
202+data HaltStrategy
203+ = HaltBoth
204+ | HaltEither
205+ | HaltLeft
206+ | HaltRight
207+
208+derive instance eqHaltStrategy :: Eq HaltStrategy
209+
210+-- | Messages for buffering queues
211+data BufferItem a
212+ = BufferChunk (Array a)
213+ | BufferDone
214+
215+derive instance functorBufferItem :: Functor BufferItem
216+
178217--------------------------------------------------------------------------------
179218-- Construction
180219--------------------------------------------------------------------------------
@@ -1097,41 +1136,187 @@ intersperse separator stream = intersperseHelper true stream
10971136 else Array.cons separator (Array.intercalate [ separator ] (Array.singleton <$> chunk))
10981137 pure $ Loop $ Tuple (Just withSeparators) (intersperseHelper false next)
10991138
1139+--------------------------------------------------------------------------------
1140+-- Buffering
1141+--------------------------------------------------------------------------------
1142+
1143+-- | Buffer stream chunks with backpressure (preserves chunking)
1144+bufferChunks :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a
1145+bufferChunks capacity stream
1146+ | capacity <= 0 = stream
1147+ | otherwise = mkStrom do
1148+ ctx <- Om.ask
1149+ liftAff do
1150+ bufferRef <- liftEffect $ Ref.new (CatQueue.empty :: CatQueue (BufferItem a))
1151+ sizeRef <- liftEffect $ Ref.new 0
1152+ mutex <- AVar.new unit
1153+ notEmpty <- AVar.empty
1154+ notFull <- AVar.new unit
1155+
1156+ let
1157+ enqueue item = do
1158+ AVar.take notFull
1159+ AVar.take mutex
1160+ size <- liftEffect $ Ref.read sizeRef
1161+ let wasEmpty = size == 0
1162+ buf <- liftEffect $ Ref.read bufferRef
1163+ liftEffect $ Ref.write (CatQueue.snoc buf item) bufferRef
1164+ liftEffect $ Ref.write (size + 1) sizeRef
1165+ AVar.put unit mutex
1166+ when wasEmpty $ AVar.put unit notEmpty
1167+ when (size + 1 < capacity) $ AVar.put unit notFull
1168+
1169+ dequeue = do
1170+ AVar.take notEmpty
1171+ AVar.take mutex
1172+ buf <- liftEffect $ Ref.read bufferRef
1173+ size <- liftEffect $ Ref.read sizeRef
1174+ case CatQueue.uncons buf of
1175+ Nothing -> do
1176+ AVar.put unit mutex
1177+ dequeue
1178+ Just (Tuple head tail) -> do
1179+ liftEffect $ Ref.write tail bufferRef
1180+ liftEffect $ Ref.write (size - 1) sizeRef
1181+ let hasMore = (size - 1) > 0
1182+ let wasFull = size == capacity
1183+ AVar.put unit mutex
1184+ when hasMore $ AVar.put unit notEmpty
1185+ when wasFull $ AVar.put unit notFull
1186+ pure head
1187+
1188+ producer s = do
1189+ stepResult <- Om.runReader ctx (runStrom s)
1190+ step <- either (\_ -> throwError (Exception.error "Stream error")) pure stepResult
1191+ case step of
1192+ Done Nothing -> enqueue BufferDone
1193+ Done (Just chunk) -> do
1194+ enqueue (BufferChunk chunk)
1195+ enqueue BufferDone
1196+ Loop (Tuple maybeChunk next) -> do
1197+ case maybeChunk of
1198+ Just chunk -> enqueue (BufferChunk chunk)
1199+ Nothing -> pure unit
1200+ producer next
1201+
1202+ fiber <- forkAff (producer stream)
1203+
1204+ let
1205+ consumer = mkStrom $ liftAff do
1206+ msg <- dequeue
1207+ case msg of
1208+ BufferDone -> do
1209+ killFiber (Exception.error "done") fiber
1210+ pure $ Done Nothing
1211+ BufferChunk chunk ->
1212+ pure $ Loop $ Tuple (Just chunk) consumer
1213+
1214+ stepResult <- Om.runReader ctx (runStrom consumer)
1215+ either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
1216+
1217+-- | Buffer stream elements with backpressure (destroys chunking)
1218+buffer :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a
1219+buffer capacity stream
1220+ | capacity <= 0 = stream
1221+ | otherwise = mkStrom do
1222+ ctx <- Om.ask
1223+ liftAff do
1224+ bufferRef <- liftEffect $ Ref.new (CatQueue.empty :: CatQueue (BufferItem a))
1225+ sizeRef <- liftEffect $ Ref.new 0
1226+ mutex <- AVar.new unit
1227+ notEmpty <- AVar.empty
1228+ notFull <- AVar.new unit
1229+
1230+ let
1231+ enqueue item = do
1232+ AVar.take notFull
1233+ AVar.take mutex
1234+ size <- liftEffect $ Ref.read sizeRef
1235+ let wasEmpty = size == 0
1236+ buf <- liftEffect $ Ref.read bufferRef
1237+ liftEffect $ Ref.write (CatQueue.snoc buf item) bufferRef
1238+ liftEffect $ Ref.write (size + 1) sizeRef
1239+ AVar.put unit mutex
1240+ when wasEmpty $ AVar.put unit notEmpty
1241+ when (size + 1 < capacity) $ AVar.put unit notFull
1242+
1243+ dequeue = do
1244+ AVar.take notEmpty
1245+ AVar.take mutex
1246+ buf <- liftEffect $ Ref.read bufferRef
1247+ size <- liftEffect $ Ref.read sizeRef
1248+ case CatQueue.uncons buf of
1249+ Nothing -> do
1250+ AVar.put unit mutex
1251+ dequeue
1252+ Just (Tuple head tail) -> do
1253+ liftEffect $ Ref.write tail bufferRef
1254+ liftEffect $ Ref.write (size - 1) sizeRef
1255+ let hasMore = (size - 1) > 0
1256+ let wasFull = size == capacity
1257+ AVar.put unit mutex
1258+ when hasMore $ AVar.put unit notEmpty
1259+ when wasFull $ AVar.put unit notFull
1260+ pure head
1261+
1262+ producer s = do
1263+ stepResult <- Om.runReader ctx (runStrom s)
1264+ step <- either (\_ -> throwError (Exception.error "Stream error")) pure stepResult
1265+ case step of
1266+ Done Nothing -> enqueue BufferDone
1267+ Done (Just chunk) -> do
1268+ traverse_ (\a -> enqueue (BufferChunk [ a ])) chunk
1269+ enqueue BufferDone
1270+ Loop (Tuple maybeChunk next) -> do
1271+ case maybeChunk of
1272+ Just chunk -> traverse_ (\a -> enqueue (BufferChunk [ a ])) chunk
1273+ Nothing -> pure unit
1274+ producer next
1275+
1276+ fiber <- forkAff (producer stream)
1277+
1278+ let
1279+ consumer = mkStrom $ liftAff do
1280+ msg <- dequeue
1281+ case msg of
1282+ BufferDone -> do
1283+ killFiber (Exception.error "done") fiber
1284+ pure $ Done Nothing
1285+ BufferChunk chunk ->
1286+ pure $ Loop $ Tuple (Just chunk) consumer
1287+
1288+ stepResult <- Om.runReader ctx (runStrom consumer)
1289+ either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
1290+
11001291--------------------------------------------------------------------------------
11011292-- Concurrency
11021293--------------------------------------------------------------------------------
11031294
1104−-- | Merge two streams concurrently
1295+-- | Merge two streams concurrently (nondeterministic like ZStream merge)
11051296-- | Elements are emitted as soon as they're available from either stream
11061297merge :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1107−merge s1 s2 = mkStrom do
1108− -- Pull from both streams concurrently
1109− step1 <- runStrom s1
1110− step2 <- runStrom s2
1111− case step1, step2 of
1112− Done Nothing, Done Nothing -> pure $ Done Nothing
1113− Done Nothing, Done (Just chunk2) -> pure $ Done $ Just chunk2
1114− Done (Just chunk1), Done Nothing -> pure $ Done $ Just chunk1
1115− Done (Just chunk1), Done (Just chunk2) -> pure $ Done $ Just (chunk1 <> chunk2)
1116− Done Nothing, Loop (Tuple maybeChunk2 next2) ->
1117− pure $ Loop $ Tuple maybeChunk2 next2
1118− Done (Just chunk1), Loop (Tuple maybeChunk2 next2) ->
1119− case maybeChunk2 of
1120− Nothing -> pure $ Loop $ Tuple (Just chunk1) next2
1121− Just chunk2 -> pure $ Loop $ Tuple (Just $ chunk1 <> chunk2) next2
1122− Loop (Tuple maybeChunk1 next1), Done Nothing ->
1123− pure $ Loop $ Tuple maybeChunk1 next1
1124− Loop (Tuple maybeChunk1 next1), Done (Just chunk2) ->
1125− case maybeChunk1 of
1126− Nothing -> pure $ Loop $ Tuple (Just chunk2) next1
1127− Just chunk1 -> pure $ Loop $ Tuple (Just $ chunk1 <> chunk2) next1
1128− Loop (Tuple maybeChunk1 next1), Loop (Tuple maybeChunk2 next2) ->
1129− -- Both streams continue, merge chunks and continue with both
1130− case maybeChunk1, maybeChunk2 of
1131− Nothing, Nothing -> pure $ Loop $ Tuple Nothing (merge next1 next2)
1132− Just c1, Nothing -> pure $ Loop $ Tuple (Just c1) (merge next1 next2)
1133− Nothing, Just c2 -> pure $ Loop $ Tuple (Just c2) (merge next1 next2)
1134− Just c1, Just c2 -> pure $ Loop $ Tuple (Just $ c1 <> c2) (merge next1 next2)
1298+merge = mergeNDWith 1024
1299+
1300+-- | Merge with halt strategy and mapping functions (ZStream-like)
1301+mergeWith
1302+ :: forall ctx err a b c
1303+ . HaltStrategy
1304+ -> (a -> c)
1305+ -> (b -> c)
1306+ -> Strom ctx err a
1307+ -> Strom ctx err b
1308+ -> Strom ctx err c
1309+mergeWith strategy f g s1 s2 =
1310+ mergeNDWithStrategy 1024 strategy (mapStrom f s1) (mapStrom g s2)
1311+
1312+mergeHaltEither :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1313+mergeHaltEither = mergeNDWithStrategy 1024 HaltEither
1314+
1315+mergeHaltLeft :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1316+mergeHaltLeft = mergeNDWithStrategy 1024 HaltLeft
1317+
1318+mergeHaltRight :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1319+mergeHaltRight = mergeNDWithStrategy 1024 HaltRight
11351320
11361321-- | Merge multiple streams concurrently
11371322mergeAll :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
@@ -1140,27 +1325,83 @@ mergeAll streams = Array.foldl merge empty streams
11401325-- | Race two streams - returns elements from whichever produces first
11411326-- | Uses Om.race for true concurrent racing
11421327race :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1143−race s1 s2 = mkStrom do
1144− Om.race [ runStrom s1, runStrom s2 ]
1328+race = mergeHaltEither
11451329
11461330-- | Race multiple streams - returns elements from whichever produces first
11471331raceAll :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
1148−raceAll streams = mkStrom do
1149− Om.race (Functor.map runStrom streams)
1332+raceAll streams =
1333+ case Array.uncons streams of
1334+ Nothing -> empty
1335+ Just { head, tail } -> Array.foldl mergeHaltEither head tail
11501336
11511337-- | Non-deterministic merge: Merge two streams concurrently, emitting elements as soon as they're available
11521338-- | Unlike `merge`, this runs both streams in parallel and elements may arrive in any order
11531339-- | The stream that produces results faster will have its elements appear first
11541340-- | Optimized: Spawns producer fibers once and uses a queue for communication
11551341mergeND :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1156−mergeND stream1 stream2 = mkStrom do
1342+mergeND = mergeNDWith 1024
1343+
1344+-- | Non-deterministic merge with configurable buffer capacity (chunk queue)
1345+mergeNDWith :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1346+mergeNDWith bufferCapacity = mergeNDWithStrategy bufferCapacity HaltBoth
1347+
1348+mergeNDWithStrategy
1349+ :: forall ctx err a
1350+ . Int
1351+ -> HaltStrategy
1352+ -> Strom ctx err a
1353+ -> Strom ctx err a
1354+ -> Strom ctx err a
1355+mergeNDWithStrategy bufferCapacity strategy stream1 stream2 = mkStrom do
11571356 -- Capture context
11581357 ctx <- Om.ask
1159−
1358+
11601359 liftAff do
1161− -- Create communication queue
1162− queue <- AVar.empty
1163−
1360+ -- Create buffered queue with backpressure
1361+ bufferRef <- liftEffect $ Ref.new (CatQueue.empty :: CatQueue (QueueItem a))
1362+ sizeRef <- liftEffect $ Ref.new 0
1363+ mutex <- AVar.new unit
1364+ notEmpty <- AVar.empty
1365+ notFull <- AVar.new unit
1366+
1367+ let
1368+ shouldStop doneLeft doneRight =
1369+ case strategy of
1370+ HaltBoth -> doneLeft && doneRight
1371+ HaltEither -> doneLeft || doneRight
1372+ HaltLeft -> doneLeft
1373+ HaltRight -> doneRight
1374+ enqueue item = do
1375+ AVar.take notFull
1376+ AVar.take mutex
1377+ size <- liftEffect $ Ref.read sizeRef
1378+ let wasEmpty = size == 0
1379+ buf <- liftEffect $ Ref.read bufferRef
1380+ liftEffect $ Ref.write (CatQueue.snoc buf item) bufferRef
1381+ liftEffect $ Ref.write (size + 1) sizeRef
1382+ AVar.put unit mutex
1383+ when wasEmpty $ AVar.put unit notEmpty
1384+ when (size + 1 < bufferCapacity) $ AVar.put unit notFull
1385+
1386+ dequeue = do
1387+ AVar.take notEmpty
1388+ AVar.take mutex
1389+ buf <- liftEffect $ Ref.read bufferRef
1390+ size <- liftEffect $ Ref.read sizeRef
1391+ case CatQueue.uncons buf of
1392+ Nothing -> do
1393+ AVar.put unit mutex
1394+ dequeue
1395+ Just (Tuple head tail) -> do
1396+ liftEffect $ Ref.write tail bufferRef
1397+ liftEffect $ Ref.write (size - 1) sizeRef
1398+ let hasMore = (size - 1) > 0
1399+ let wasFull = size == bufferCapacity
1400+ AVar.put unit mutex
1401+ when hasMore $ AVar.put unit notEmpty
1402+ when wasFull $ AVar.put unit notFull
1403+ pure head
1404+
11641405 -- Spawn producer fiber for stream1
11651406 fiber1 <- forkAff do
11661407 let
@@ -1168,18 +1409,18 @@ mergeND stream1 stream2 = mkStrom do
11681409 stepResult <- Om.runReader ctx (runStrom s)
11691410 step <- either (\e -> throwError (Exception.error "Stream error")) pure stepResult
11701411 case step of
1171− Done Nothing ->
1172− AVar.put (Left StreamId1) queue -- Signal stream1 done
1412+ Done Nothing ->
1413+ enqueue (StreamDone StreamId1)
11731414 Done (Just chunk) -> do
1174− AVar.put (Right chunk) queue -- Emit final chunk
1175− AVar.put (Left StreamId1) queue -- Signal stream1 done
1415+ enqueue (DataChunk chunk StreamId1)
1416+ enqueue (StreamDone StreamId1)
11761417 Loop (Tuple maybeChunk next) -> do
11771418 case maybeChunk of
1178− Just chunk -> AVar.put (Right chunk) queue
1419+ Just chunk -> enqueue (DataChunk chunk StreamId1)
11791420 Nothing -> pure unit
11801421 producer next
11811422 producer stream1
1182−
1423+
11831424 -- Spawn producer fiber for stream2
11841425 fiber2 <- forkAff do
11851426 let
@@ -1187,44 +1428,126 @@ mergeND stream1 stream2 = mkStrom do
11871428 stepResult <- Om.runReader ctx (runStrom s)
11881429 step <- either (\e -> throwError (Exception.error "Stream error")) pure stepResult
11891430 case step of
1190− Done Nothing ->
1191− AVar.put (Left StreamId2) queue -- Signal stream2 done
1431+ Done Nothing ->
1432+ enqueue (StreamDone StreamId2)
11921433 Done (Just chunk) -> do
1193− AVar.put (Right chunk) queue -- Emit final chunk
1194− AVar.put (Left StreamId2) queue -- Signal stream2 done
1434+ enqueue (DataChunk chunk StreamId2)
1435+ enqueue (StreamDone StreamId2)
11951436 Loop (Tuple maybeChunk next) -> do
11961437 case maybeChunk of
1197− Just chunk -> AVar.put (Right chunk) queue
1438+ Just chunk -> enqueue (DataChunk chunk StreamId2)
11981439 Nothing -> pure unit
11991440 producer next
12001441 producer stream2
1201−
1442+
12021443 -- Consumer: pull from queue
12031444 let
1204− consumer doneCount = mkStrom $ liftAff do
1205− if doneCount >= 2 then do
1445+ consumer doneLeft doneRight = mkStrom $ liftAff do
1446+ if shouldStop doneLeft doneRight then do
12061447 -- Both streams done, cleanup
12071448 killFiber (Exception.error "done") fiber1
12081449 killFiber (Exception.error "done") fiber2
12091450 pure $ Done Nothing
12101451 else do
1211− msg <- AVar.take queue
1452+ msg <- dequeue
12121453 case msg of
1213− Left _streamId -> do
1454+ StreamDone streamId -> do
12141455 -- One stream finished, recurse
1215− stepResult <- Om.runReader ctx (runStrom $ consumer (doneCount + 1))
1456+ let
1457+ doneLeft' = doneLeft || streamId == StreamId1
1458+ doneRight' = doneRight || streamId == StreamId2
1459+ stepResult <- Om.runReader ctx (runStrom $ consumer doneLeft' doneRight')
12161460 either (\_ -> throwError (Exception.error "Consumer error")) pure stepResult
1217− Right chunk ->
1461+ DataChunk chunk _streamId ->
12181462 -- Got a chunk
1219− pure $ Loop $ Tuple (Just chunk) (consumer doneCount)
1220−
1221− stepResult <- Om.runReader ctx (runStrom $ consumer 0)
1463+ pure $ Loop $ Tuple (Just chunk) (consumer doneLeft doneRight)
1464+
1465+ stepResult <- Om.runReader ctx (runStrom $ consumer false false)
12221466 either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
12231467
12241468-- | Non-deterministic merge of multiple streams
12251469-- | All streams run concurrently and emit elements as soon as they're available
12261470mergeAllND :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
1227−mergeAllND streams = Array.foldl mergeND empty streams
1471+mergeAllND = mergeAllNDWith 1024
1472+
1473+-- | Non-deterministic merge of multiple streams with configurable buffer capacity
1474+mergeAllNDWith :: forall ctx err a. Int -> Array (Strom ctx err a) -> Strom ctx err a
1475+mergeAllNDWith bufferCapacity streams
1476+ | Array.null streams = empty
1477+ | otherwise = mkStrom do
1478+ ctx <- Om.ask
1479+ liftAff do
1480+ bufferRef <- liftEffect $ Ref.new (CatQueue.empty :: CatQueue (MergeAllItem a))
1481+ sizeRef <- liftEffect $ Ref.new 0
1482+ mutex <- AVar.new unit
1483+ notEmpty <- AVar.empty
1484+ notFull <- AVar.new unit
1485+
1486+ let
1487+ enqueue item = do
1488+ AVar.take notFull
1489+ AVar.take mutex
1490+ size <- liftEffect $ Ref.read sizeRef
1491+ let wasEmpty = size == 0
1492+ buf <- liftEffect $ Ref.read bufferRef
1493+ liftEffect $ Ref.write (CatQueue.snoc buf item) bufferRef
1494+ liftEffect $ Ref.write (size + 1) sizeRef
1495+ AVar.put unit mutex
1496+ when wasEmpty $ AVar.put unit notEmpty
1497+ when (size + 1 < bufferCapacity) $ AVar.put unit notFull
1498+
1499+ dequeue = do
1500+ AVar.take notEmpty
1501+ AVar.take mutex
1502+ buf <- liftEffect $ Ref.read bufferRef
1503+ size <- liftEffect $ Ref.read sizeRef
1504+ case CatQueue.uncons buf of
1505+ Nothing -> do
1506+ AVar.put unit mutex
1507+ dequeue
1508+ Just (Tuple head tail) -> do
1509+ liftEffect $ Ref.write tail bufferRef
1510+ liftEffect $ Ref.write (size - 1) sizeRef
1511+ let hasMore = (size - 1) > 0
1512+ let wasFull = size == bufferCapacity
1513+ AVar.put unit mutex
1514+ when hasMore $ AVar.put unit notEmpty
1515+ when wasFull $ AVar.put unit notFull
1516+ pure head
1517+
1518+ producer s = do
1519+ stepResult <- Om.runReader ctx (runStrom s)
1520+ step <- either (\_ -> throwError (Exception.error "Stream error")) pure stepResult
1521+ case step of
1522+ Done Nothing -> enqueue MergeDone
1523+ Done (Just chunk) -> do
1524+ enqueue (MergeChunk chunk)
1525+ enqueue MergeDone
1526+ Loop (Tuple maybeChunk next) -> do
1527+ case maybeChunk of
1528+ Just chunk -> enqueue (MergeChunk chunk)
1529+ Nothing -> pure unit
1530+ producer next
1531+
1532+ fibers <- traverse (\s -> forkAff (producer s)) streams
1533+
1534+ let
1535+ totalStreams = Array.length streams
1536+ consumer doneCount = mkStrom $ liftAff do
1537+ if doneCount >= totalStreams then do
1538+ traverse_ (\fiber -> killFiber (Exception.error "done") fiber) fibers
1539+ pure $ Done Nothing
1540+ else do
1541+ msg <- dequeue
1542+ case msg of
1543+ MergeDone -> do
1544+ stepResult <- Om.runReader ctx (runStrom $ consumer (doneCount + 1))
1545+ either (\_ -> throwError (Exception.error "Consumer error")) pure stepResult
1546+ MergeChunk chunk ->
1547+ pure $ Loop $ Tuple (Just chunk) (consumer doneCount)
1548+
1549+ stepResult <- Om.runReader ctx (runStrom $ consumer 0)
1550+ either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
12281551
12291552--------------------------------------------------------------------------------
12301553-- Parallel Processing
@@ -1264,14 +1587,26 @@ mapPar concurrency f stream
12641587 in
12651588 if Array.null group then [] else Array.cons group (chunksOf n rest)
12661589
1590+-- | Unordered parallel map (currently same as mapPar)
1591+mapParUnordered :: forall ctx err a b. Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
1592+mapParUnordered = mapPar
1593+
12671594-- | Alias for mapPar with explicit "M" naming
12681595mapMPar :: forall ctx err a b. Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
12691596mapMPar = mapPar
12701597
1598+-- | Alias for unordered mapPar with explicit "M" naming
1599+mapMParUnordered :: forall ctx err a b. Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
1600+mapMParUnordered = mapParUnordered
1601+
12711602-- | For each element, execute an effect in parallel with bounded concurrency
12721603foreachPar :: forall ctx err a. Int -> (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit
12731604foreachPar concurrency f stream = runDrain (mapPar concurrency f stream)
12741605
1606+-- | Unordered parallel foreach
1607+foreachParUnordered :: forall ctx err a. Int -> (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit
1608+foreachParUnordered concurrency f stream = runDrain (mapParUnordered concurrency f stream)
1609+
12751610--------------------------------------------------------------------------------
12761611-- Error Handling
12771612--------------------------------------------------------------------------------