@@ -62,14 +62,9 @@ module Streamly.Internal.Data.Stream.Concurrent
6262
6363 -- ** Combine two
6464 -- | Use a channel for each pair.
65− -- combine/concur/conjoin
66− , append2
67− , interleave2
68− , ahead2
69− , parallel2
70− , parallelFst2
71− , parallelMin2
72− , combineWith
65+ , parTwo
66+ , parZipWithM
67+ , parZipWith
7368
7469 -- ** List of streams
7570 -- | Shares a single channel across many streams.
@@ -81,9 +76,6 @@ module Streamly.Internal.Data.Stream.Concurrent
8176 , parListEagerMin
8277 , parList
8378
84− , parZipWithM
85− , parZipWith
86−
8779 -- ** Stream of streams
8880 -- *** Apply
8981 , parApply
@@ -246,85 +238,37 @@ _appendWithChanK :: MonadAsync m =>
246238_appendWithChanK chan stream1 stream2 =
247239 K.before (toChannelK chan stream2) stream1
248240
249−
250−-- | Like 'append' but with a Config modifier.
251−{-# INLINE combineWith #-}
252−combineWith :: MonadAsync m =>
253− (Config -> Config) -> Stream m a -> Stream m a -> Stream m a
254−combineWith modifier stream1 stream2 =
255− Stream.fromStreamK
256− $ appendWithK
257− modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)
258−
259−-- | Binary operation to evaluate two streams concurrently prioritizing the
260−-- left stream.
241+-- | Binary operation to evaluate two streams concurrently using a channel.
261242--
262243-- If you want to combine more than two streams you almost always want the
263−-- 'concat' operation instead. The performance of this operation degrades
264−-- rapidly when more streams are combined as each operation adds one more
265−-- concurrent channel. On the other hand, 'concat' uses a single channel for
266−-- all streams. However, with this operation you can precisely control the
267−-- scheduling by creating arbitrary shape expression trees.
244+-- 'parList' or `parConcat` operation instead. The performance of this
245+-- operation degrades rapidly when more streams are combined as each operation
246+-- adds one more concurrent channel. On the other hand, 'parConcat' uses a
247+-- single channel for all streams. However, with this operation you can
248+-- precisely control the scheduling by creating arbitrary shape expression
249+-- trees.
250+--
251+-- Definition:
268252--
269−-- >>> append2 = Stream.combineWith id
253+-- >>> parTwo cfg x y = Stream.parList cfg [x, y]
270254--
271−-- The following code finishes in 4 seconds:
255+-- Example, the following code finishes in 4 seconds:
272256--
257+-- >>> async = Stream.parTwo id
273258-- >>> stream1 = Stream.fromEffect (delay 4)
274259-- >>> stream2 = Stream.fromEffect (delay 2)
275−-- >>> Stream.fold Fold.toList $ stream1 `Stream.append2` stream2
260+-- >>> Stream.fold Fold.toList $ stream1 `async` stream2
276261-- 2 sec
277262-- 4 sec
278263-- [2,4]
279264--
280−{-# INLINE append2 #-}
281−append2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
282−append2 = combineWith id
283−
284−-- | Like 'append' but interleaves the streams fairly instead of prioritizing
285−-- the left stream. This schedules all streams in a round robin fashion over
286−-- limited number of threads.
287−--
288−-- >>> interleave2 = Stream.combineWith (Stream.interleaved True)
289−--
290−{-# INLINE interleave2 #-}
291−interleave2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
292−interleave2 = combineWith (interleaved True)
293−
294−-- | Like 'append' but with 'ordered' on.
295−--
296−-- >>> ahead2 = Stream.combineWith (Stream.ordered True)
297−--
298−{-# INLINE ahead2 #-}
299−ahead2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
300−ahead2 = combineWith (ordered True)
301−
302−-- | Like 'append2' but with 'eager' on.
303−--
304−-- >>> parallel2 = Stream.combineWith (Stream.eager True)
305−--
306−{-# INLINE parallel2 #-}
307−parallel2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
308−parallel2 = combineWith (eager True)
309−
310−-- | Like 'parallel2' but stops the output as soon as the first stream stops.
311−--
312−-- >>> parallelFst2 = Stream.combineWith (Stream.eager True . Stream.stopWhen Stream.FirstStops)
313−--
314−-- /Pre-release/
315−{-# INLINE parallelFst2 #-}
316−parallelFst2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
317−parallelFst2 = combineWith (eager True . stopWhen FirstStops)
318−
319−-- | Like 'parallel2' but stops the output as soon as any of the two streams
320−-- stops.
321−--
322−-- >>> parallelMin2 = Stream.combineWith (Stream.eager True . Stream.stopWhen Stream.AnyStops)
323−--
324−-- /Pre-release/
325−{-# INLINE parallelMin2 #-}
326−parallelMin2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
327−parallelMin2 = combineWith (eager True . stopWhen AnyStops)
265+{-# INLINE parTwo #-}
266+parTwo :: MonadAsync m =>
267+ (Config -> Config) -> Stream m a -> Stream m a -> Stream m a
268+parTwo modifier stream1 stream2 =
269+ Stream.fromStreamK
270+ $ appendWithK
271+ modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)
328272
329273-------------------------------------------------------------------------------
330274-- concat streams