git.iko.soy/streamly
on master help sign in
Allow state passing through stream append operations
commite2c366c675d7b35e9c9fa4dfa2c2990d38dec520 authorHarendra Kumar <harendra.kumar@gmail.com> dateSat 18 Aug 2018, 02:38 (8 years ago) parent526f024
5 files changed+19−7
compare with master
unifiedsplit
Changelog.md+3 src/Streamly/Streams/Ahead.hs+2−1 src/Streamly/Streams/Async.hs+5−3 src/Streamly/Streams/Parallel.hs+3−1 src/Streamly/Streams/Serial.hs+6−2
Changelog.md+3
@@ -7,6 +7,9 @@
77 stream.
88* Leftover threads are now cleaned up as soon as the SVar is garbage
99 collected.
10+* Fix passing of state across parallel, async, wAsync, ahead, serial, wSerial
11+ combinators. Without this fix combinators that rely on state passing e.g.
12+ `maxThreads` and `maxBuffer` won't work across these combinators.
1013
1114### Enhancements
1215
src/Streamly/Streams/Ahead.hs+2−1
@@ -634,7 +634,8 @@ instance IsStream AheadT where
634634-- @since 0.3.0
635635{-# INLINE ahead #-}
636636ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
637−ahead m1 m2 = fromStream $ aheadS (toStream m1) (toStream m2)
637+ahead m1 m2 = fromStream $ Stream $ \st stp sng yld ->
638+ unStream (aheadS (toStream m1) (toStream m2)) st stp sng yld
638639
639640instance MonadAsync m => Semigroup (AheadT m a) where
640641 (<>) = ahead
src/Streamly/Streams/Async.hs+5−3
@@ -582,8 +582,9 @@ asyncS = joinStreamVarAsync AsyncVar
582582-- @since 0.2.0
583583{-# INLINE async #-}
584584async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
585−async m1 m2 = fromStream $
586− joinStreamVarAsync AsyncVar (toStream m1) (toStream m2)
585+async m1 m2 = fromStream $ Stream $ \st stp sng yld ->
586+ unStream (joinStreamVarAsync AsyncVar (toStream m1) (toStream m2))
587+ st stp sng yld
587588
588589-- | Same as 'async'.
589590--
@@ -729,7 +730,8 @@ consMWAsync m r = K.yieldM m `wAsyncS` r
729730-- @since 0.2.0
730731{-# INLINE wAsync #-}
731732wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
732−wAsync m1 m2 = fromStream $ wAsyncS (toStream m1) (toStream m2)
733+wAsync m1 m2 = fromStream $ Stream $ \st stp sng yld ->
734+ unStream (wAsyncS (toStream m1) (toStream m2)) st stp sng yld
733735
734736-- | Wide async composition or async composition with breadth first traversal.
735737-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed
src/Streamly/Streams/Parallel.hs+3−1
@@ -117,7 +117,9 @@ consMParallel m r = K.yieldM m `parallelStream` r
117117-- @since 0.2.0
118118{-# INLINE parallel #-}
119119parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
120−parallel m1 m2 = fromStream $ parallelStream (toStream m1) (toStream m2)
120+parallel m1 m2 = fromStream $ Stream $ \st stp sng yld -> do
121+ unStream (parallelStream (toStream m1) (toStream m2))
122+ st stp sng yld
121123
122124------------------------------------------------------------------------------
123125-- Convert a stream to parallel
src/Streamly/Streams/Serial.hs+6−2
@@ -166,7 +166,9 @@ instance IsStream SerialT where
166166-- @since 0.2.0
167167{-# INLINE serial #-}
168168serial :: IsStream t => t m a -> t m a -> t m a
169−serial m1 m2 = fromStream $ K.serial (toStream m1) (toStream m2)
169+serial m1 m2 = fromStream $ Stream $ \st stp sng yld ->
170+ unStream (K.serial (toStream m1) (toStream m2))
171+ (rstState st) stp sng yld
170172
171173------------------------------------------------------------------------------
172174-- Monad
@@ -298,7 +300,9 @@ interleave m1 m2 = Stream $ \st stp sng yld -> do
298300-- @since 0.2.0
299301{-# INLINE wSerial #-}
300302wSerial :: IsStream t => t m a -> t m a -> t m a
301−wSerial m1 m2 = fromStream $ interleave (toStream m1) (toStream m2)
303+wSerial m1 m2 = fromStream $ Stream $ \st stp sng yld ->
304+ unStream (interleave (toStream m1) (toStream m2))
305+ (rstState st) stp sng yld
302306
303307instance Semigroup (WSerialT m a) where
304308 (<>) = wSerial