@@ -508,6 +508,7 @@ tapOffsetEvery offset n f xs =
508-- parent stream, but we guarantee that before the parent stream stops the tap
508-- parent stream, but we guarantee that before the parent stream stops the tap
509-- finishes and all exceptions from it are drained.
509-- finishes and all exceptions from it are drained.
511+-- >>> tapAsync f = Stream.tapAsyncK (Stream.fold f . Stream.adapt)
512-- Compare with 'tap'.
513-- Compare with 'tap'.
@@ -516,37 +517,18 @@ tapOffsetEvery offset n f xs =
516tapAsync :: (IsStream t, MonadAsync m) => FL.Fold m a b -> t m a -> t m a
517tapAsync :: (IsStream t, MonadAsync m) => FL.Fold m a b -> t m a -> t m a
517tapAsync f xs = fromStreamD $ Par.tapAsyncF f (toStreamD xs)
518tapAsync f xs = fromStreamD $ Par.tapAsyncF f (toStreamD xs)
519−-- | Redirect a copy of the stream to a supplied fold and run it concurrently
520+-- XXX We should keep only tapAsyncK as tapAsync can be implemented in terms
520−-- in an independent thread. The fold may buffer some elements. The buffer size
521+-- of tapAsyncK and it has better performance. Rename this to tapAsync.
521−-- is determined by the prevailing 'Streamly.Prelude.maxBuffer' setting.
522+
522−--
523+-- | Like 'tapAsyncF' but uses a stream fold function instead of a 'Fold' type.
526−-- -----stream m a ---------------stream m a-----
531−-- > S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
536−-- Exceptions from the concurrently running fold are propagated to the current
537−-- computation. Note that, because of buffering in the fold, exceptions may be
538−-- delayed and may not correspond to the current element being processed in the
539−-- parent stream, but we guarantee that before the parent stream stops the tap
540−-- finishes and all exceptions from it are drained.
543−-- Compare with 'tap'.
545-- /Pre-release/
525-- /Pre-release/
546{-# INLINE tapAsyncK #-}
526{-# INLINE tapAsyncK #-}
547tapAsyncK :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> t m a
527tapAsyncK :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> t m a
548tapAsyncK f m = fromStream $ Par.tapAsyncK (f . fromStream) (toStream m)
528tapAsyncK f m = fromStream $ Par.tapAsyncK (f . fromStream) (toStream m)
530+-- XXX rename this to tapManyAsync_?
550-- | Concurrently distribute a stream to a collection of fold functions,
532-- | Concurrently distribute a stream to a collection of fold functions,
551-- discarding the outputs of the folds.
533-- discarding the outputs of the folds.