@@ -114,20 +114,20 @@ import Streamly.Streams
114114unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a
115115unfoldr step = fromStream . go
116116 where
117− go s = Stream $ \_ stp yld ->
117+ go s = Stream $ \_ stp _ yld ->
118118 case step s of
119119 Nothing -> stp
120− Just (a, b) -> yld a (Just (go b))
120+ Just (a, b) -> yld a (go b)
121121
122122-- | Build a Stream by unfolding monadic steps starting from a seed.
123123unfoldrM :: (IsStream t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a
124124unfoldrM step = fromStream . go
125125 where
126− go s = Stream $ \_ stp yld -> do
126+ go s = Stream $ \_ stp _ yld -> do
127127 mayb <- step s
128128 case mayb of
129129 Nothing -> stp
130− Just (a, b) -> yld a (Just (go b))
130+ Just (a, b) -> yld a (go b)
131131
132132-- | Construct a stream from a 'Foldable' container.
133133{-# INLINE fromFoldable #-}
@@ -144,27 +144,27 @@ each = fromFoldable
144144iterate :: IsStream t => (a -> a) -> a -> t m a
145145iterate step = fromStream . go
146146 where
147− go s = S.cons s (Just (go (step s)))
147+ go s = S.cons s (go (step s))
148148
149149-- | Iterate a monadic function from a seed value, streaming the results forever
150150iterateM :: (IsStream t, Monad m) => (a -> m a) -> a -> t m a
151151iterateM step = fromStream . go
152152 where
153− go s = Stream $ \_ _ yld -> do
153+ go s = Stream $ \_ _ _ yld -> do
154154 a <- step s
155− yld s (Just (go a))
155+ yld s (go a)
156156
157157-- | Read lines from an IO Handle into a stream of Strings.
158158fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String
159159fromHandle h = fromStream go
160160 where
161− go = Stream $ \_ stp yld -> do
161+ go = Stream $ \_ stp _ yld -> do
162162 eof <- liftIO $ IO.hIsEOF h
163163 if eof
164164 then stp
165165 else do
166166 str <- liftIO $ IO.hGetLine h
167− yld str (Just go)
167+ yld str go
168168
169169------------------------------------------------------------------------------
170170-- Elimination
@@ -181,9 +181,9 @@ foldr step acc m = go (toStream m)
181181 where
182182 go m1 =
183183 let stop = return acc
184− yield a Nothing = return (step a acc)
185− yield a (Just x) = go x >>= \b -> return (step a b)
186− in (S.runStream m1) Nothing stop yield
184+ single a = return (step a acc)
185+ yield a r = go r >>= \b -> return (step a b)
186+ in (S.runStream m1) Nothing stop single yield
187187
188188-- | Lazy right fold with a monadic step function. For example, to fold a
189189-- stream into a list:
@@ -198,9 +198,9 @@ foldrM step acc m = go (toStream m)
198198 where
199199 go m1 =
200200 let stop = return acc
201− yield a Nothing = step a acc
202− yield a (Just x) = (go x) >>= (step a)
203− in (S.runStream m1) Nothing stop yield
201+ single a = step a acc
202+ yield a r = go r >>= step a
203+ in (S.runStream m1) Nothing stop single yield
204204
205205-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
206206-- user supplied extraction function (the third argument) at each step. This is
@@ -210,13 +210,12 @@ foldrM step acc m = go (toStream m)
210210scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
211211scanx step begin done m = cons (done begin) $ fromStream $ go (toStream m) begin
212212 where
213− go m1 !acc = Stream $ \_ stp yld ->
214− let stop = stp
215− yield a Nothing = yld (done $ step acc a) Nothing
216− yield a (Just x) =
213+ go m1 !acc = Stream $ \_ stp sng yld ->
214+ let single a = sng (done $ step acc a)
215+ yield a r =
217216 let s = step acc a
218− in yld (done s) (Just (go x s))
219− in S.runStream m1 Nothing stop yield
217+ in yld (done s) (go r s)
218+ in S.runStream m1 Nothing stp single yield
220219
221220{-# DEPRECATED scan "Please use scanx instead." #-}
222221scan :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
@@ -240,22 +239,20 @@ foldx step begin done m = get $ go (toStream m) begin
240239 where
241240 {-# NOINLINE get #-}
242241 get m1 =
243− let yield a Nothing = return $ done a
244− yield _ _ = undefined
245− in (S.runStream m1) Nothing undefined yield
242+ let single = return . done
243+ in (S.runStream m1) Nothing undefined single undefined
246244
247245 -- Note, this can be implemented by making a recursive call to "go",
248246 -- however that is more expensive because of unnecessary recursion
249247 -- that cannot be tail call optimized. Unfolding recursion explicitly via
250248 -- continuations is much more efficient.
251− go m1 !acc = Stream $ \_ _ yld ->
252− let stop = yld acc Nothing
249+ go m1 !acc = Stream $ \_ _ sng yld ->
250+ let stop = sng acc
251+ single a = sng $ step acc a
253252 yield a r =
254− let s = step acc a
255− in case r of
256− Nothing -> yld s Nothing
257− Just x -> (S.runStream (go x s)) Nothing undefined yld
258− in (S.runStream m1) Nothing stop yield
253+ let stream = go r (step acc a)
254+ in (S.runStream stream) Nothing undefined sng yld
255+ in (S.runStream m1) Nothing stop single yield
259256
260257{-# DEPRECATED foldl "Please use foldx instead." #-}
261258foldl :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b
@@ -273,9 +270,9 @@ foldxM step begin done m = go begin (toStream m)
273270 where
274271 go !acc m1 =
275272 let stop = acc >>= done
276− yield a Nothing = acc >>= \b -> step b a >>= done
277− yield a (Just x) = acc >>= \b -> go (step b a) x
278− in (S.runStream m1) Nothing stop yield
273+ single a = acc >>= \b -> step b a >>= done
274+ yield a r = acc >>= \b -> go (step b a) r
275+ in (S.runStream m1) Nothing stop single yield
279276
280277{-# DEPRECATED foldlM "Please use foldxM instead." #-}
281278foldlM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b
@@ -291,9 +288,9 @@ foldlM' step begin m = foldxM step (return begin) return m
291288uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))
292289uncons m =
293290 let stop = return Nothing
294− yield a Nothing = return (Just (a, nil))
295− yield a (Just x) = return (Just (a, fromStream x))
296− in (S.runStream (toStream m)) Nothing stop yield
291+ single a = return (Just (a, nil))
292+ yield a r = return (Just (a, fromStream r))
293+ in (S.runStream (toStream m)) Nothing stop single yield
297294
298295-- | Write a stream of Strings to an IO Handle.
299296toHandle :: MonadIO m => IO.Handle -> SerialT m String -> m ()
@@ -301,9 +298,9 @@ toHandle h m = go (toStream m)
301298 where
302299 go m1 =
303300 let stop = return ()
304− yield a Nothing = liftIO (IO.hPutStrLn h a)
305− yield a (Just x) = liftIO (IO.hPutStrLn h a) >> go x
306− in (S.runStream m1) Nothing stop yield
301+ single a = liftIO (IO.hPutStrLn h a)
302+ yield a r = liftIO (IO.hPutStrLn h a) >> go r
303+ in (S.runStream m1) Nothing stop single yield
307304
308305------------------------------------------------------------------------------
309306-- Special folds
@@ -319,46 +316,45 @@ toList = foldrM (\a xs -> return (a : xs)) []
319316take :: IsStream t => Int -> t m a -> t m a
320317take n m = fromStream $ go n (toStream m)
321318 where
322− go n1 m1 = Stream $ \ctx stp yld ->
323− let yield a Nothing = yld a Nothing
324− yield a (Just x) = yld a (Just (go (n1 - 1) x))
325− in if n1 <= 0 then stp else (S.runStream m1) ctx stp yield
319+ go n1 m1 = Stream $ \ctx stp sng yld ->
320+ let yield a r = yld a (go (n1 - 1) r)
321+ in if n1 <= 0 then stp else (S.runStream m1) ctx stp sng yield
326322
327323-- | Include only those elements that pass a predicate.
328324{-# INLINE filter #-}
329325filter :: IsStream t => (a -> Bool) -> t m a -> t m a
330326filter p m = fromStream $ go (toStream m)
331327 where
332− go m1 = Stream $ \ctx stp yld ->
333− let yield a Nothing | p a = yld a Nothing
334− | otherwise = stp
335− yield a (Just x) | p a = yld a (Just (go x))
336− | otherwise = (S.runStream x) ctx stp yield
337− in (S.runStream m1) ctx stp yield
328+ go m1 = Stream $ \ctx stp sng yld ->
329+ let single a | p a = sng a
330+ | otherwise = stp
331+ yield a r | p a = yld a (go r)
332+ | otherwise = (S.runStream r) ctx stp single yield
333+ in (S.runStream m1) ctx stp single yield
338334
339335-- | End the stream as soon as the predicate fails on an element.
340336{-# INLINE takeWhile #-}
341337takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
342338takeWhile p m = fromStream $ go (toStream m)
343339 where
344− go m1 = Stream $ \ctx stp yld ->
345− let yield a Nothing | p a = yld a Nothing
346− | otherwise = stp
347− yield a (Just x) | p a = yld a (Just (go x))
348− | otherwise = stp
349− in (S.runStream m1) ctx stp yield
340+ go m1 = Stream $ \ctx stp sng yld ->
341+ let single a | p a = sng a
342+ | otherwise = stp
343+ yield a r | p a = yld a (go r)
344+ | otherwise = stp
345+ in (S.runStream m1) ctx stp single yield
350346
351347-- | Discard first 'n' elements from the stream and take the rest.
352348drop :: IsStream t => Int -> t m a -> t m a
353349drop n m = fromStream $ go n (toStream m)
354350 where
355− go n1 m1 = Stream $ \ctx stp yld ->
356− let yield _ Nothing = stp
357− yield _ (Just x) = (S.runStream $ go (n1 - 1) x) ctx stp yld
351+ go n1 m1 = Stream $ \ctx stp sng yld ->
352+ let single _ = stp
353+ yield _ r = (S.runStream $ go (n1 - 1) r) ctx stp sng yld
358354 -- Somehow "<=" check performs better than a ">"
359355 in if n1 <= 0
360− then (S.runStream m1) ctx stp yld
361− else (S.runStream m1) ctx stp yield
356+ then (S.runStream m1) ctx stp sng yld
357+ else (S.runStream m1) ctx stp single yield
362358
363359-- | Drop elements in the stream as long as the predicate succeeds and then
364360-- take the rest of the stream.
@@ -366,34 +362,34 @@ drop n m = fromStream $ go n (toStream m)
366362dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
367363dropWhile p m = fromStream $ go (toStream m)
368364 where
369− go m1 = Stream $ \ctx stp yld ->
370− let yield a Nothing | p a = stp
371− | otherwise = yld a Nothing
372− yield a (Just x) | p a = (S.runStream x) ctx stp yield
373− | otherwise = yld a (Just x)
374− in (S.runStream m1) ctx stp yield
365+ go m1 = Stream $ \ctx stp sng yld ->
366+ let single a | p a = stp
367+ | otherwise = sng a
368+ yield a r | p a = (S.runStream r) ctx stp single yield
369+ | otherwise = yld a r
370+ in (S.runStream m1) ctx stp single yield
375371
376372-- | Determine whether all elements of a stream satisfy a predicate.
377373all :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
378374all p m = go (toStream m)
379375 where
380376 go m1 =
381− let yield a Nothing | p a = return True
382− | otherwise = return False
383− yield a (Just x) | p a = go x
384− | otherwise = return False
385− in (S.runStream m1) Nothing (return True) yield
377+ let single a | p a = return True
378+ | otherwise = return False
379+ yield a r | p a = go r
380+ | otherwise = return False
381+ in (S.runStream m1) Nothing (return True) single yield
386382
387383-- | Determine whether any of the elements of a stream satisfy a predicate.
388384any :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
389385any p m = go (toStream m)
390386 where
391387 go m1 =
392− let yield a Nothing | p a = return True
393− | otherwise = return False
394− yield a (Just x) | p a = return True
395− | otherwise = go x
396− in (S.runStream m1) Nothing (return False) yield
388+ let single a | p a = return True
389+ | otherwise = return False
390+ yield a r | p a = return True
391+ | otherwise = go r
392+ in (S.runStream m1) Nothing (return False) single yield
397393
398394-- | Determine the sum of all elements of a stream of numbers
399395sum :: (Monad m, Num a) => SerialT m a -> m a
@@ -407,16 +403,17 @@ product = foldl (*) 1 id
407403head :: Monad m => SerialT m a -> m (Maybe a)
408404head m =
409405 let stop = return Nothing
406+ single a = return (Just a)
410407 yield a _ = return (Just a)
411− in (S.runStream (toStream m)) Nothing stop yield
408+ in (S.runStream (toStream m)) Nothing stop single yield
412409
413410-- | Extract all but the first element of the stream, if any.
414411tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
415412tail m =
416− let stop = return Nothing
417− yield _ Nothing = return $ Just nil
418− yield _ (Just t) = return $ Just $ fromStream t
419− in (S.runStream (toStream m)) Nothing stop yield
413+ let stop = return Nothing
414+ single _ = return $ Just nil
415+ yield _ r = return $ Just $ fromStream r
416+ in (S.runStream (toStream m)) Nothing stop single yield
420417
421418-- | Extract the last element of the stream, if any.
422419{-# INLINE last #-}
@@ -427,28 +424,29 @@ last = foldl (\_ y -> Just y) Nothing id
427424null :: Monad m => SerialT m a -> m Bool
428425null m =
429426 let stop = return True
427+ single _ = return False
430428 yield _ _ = return False
431− in (S.runStream (toStream m)) Nothing stop yield
429+ in (S.runStream (toStream m)) Nothing stop single yield
432430
433431-- | Determine whether an element is present in the stream.
434432elem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
435433elem e m = go (toStream m)
436434 where
437435 go m1 =
438− let stop = return False
439− yield a Nothing = return (a == e)
440− yield a (Just x) = if a == e then return True else go x
441− in (S.runStream m1) Nothing stop yield
436+ let stop = return False
437+ single a = return (a == e)
438+ yield a r = if a == e then return True else go r
439+ in (S.runStream m1) Nothing stop single yield
442440
443441-- | Determine whether an element is not present in the stream.
444442notElem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
445443notElem e m = go (toStream m)
446444 where
447445 go m1 =
448− let stop = return True
449− yield a Nothing = return (a /= e)
450− yield a (Just x) = if a == e then return False else go x
451− in (S.runStream m1) Nothing stop yield
446+ let stop = return True
447+ single a = return (a /= e)
448+ yield a r = if a == e then return False else go r
449+ in (S.runStream m1) Nothing stop single yield
452450
453451-- | Determine the length of the stream.
454452length :: Monad m => SerialT m a -> m Int
@@ -457,28 +455,27 @@ length = foldl (\n _ -> n + 1) 0 id
457455-- | Returns the elements of the stream in reverse order.
458456-- The stream must be finite.
459457reverse :: (IsStream t) => t m a -> t m a
460−reverse m = fromStream $ go Nothing (toStream m)
458+reverse m = fromStream $ go S.nil (toStream m)
461459 where
462− go rev rest = Stream $ \svr stp yld ->
463− let stop = case rev of
464− Nothing -> stp
465− Just str -> S.runStream str svr stp yld
466− yield a Nothing = S.runStream (a `S.cons` rev) svr stp yld
467− yield a (Just x) = S.runStream (go (Just $ a `S.cons` rev) x) svr stp yld
468− in S.runStream rest svr stop yield
460+ go rev rest = Stream $ \svr stp sng yld ->
461+ let run x = S.runStream x svr stp sng yld
462+ stop = run rev
463+ single a = run $ a `S.cons` rev
464+ yield a r = run $ go (a `S.cons` rev) r
465+ in S.runStream rest svr stop single yield
469466
470467-- XXX replace the recursive "go" with continuation
471468-- | Determine the minimum element in a stream.
472469minimum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
473470minimum m = go Nothing (toStream m)
474471 where
475− go r m1 =
476− let stop = return r
477− yield a Nothing = return $ min_ a r
478− yield a (Just x) = go (min_ a r) x
479− in (S.runStream m1) Nothing stop yield
472+ go res m1 =
473+ let stop = return res
474+ single a = return $ min_ a res
475+ yield a r = go (min_ a res) r
476+ in (S.runStream m1) Nothing stop single yield
480477
481− min_ a r = case r of
478+ min_ a res = case res of
482479 Nothing -> Just a
483480 Just e -> Just $ min a e
484481
@@ -487,13 +484,13 @@ minimum m = go Nothing (toStream m)
487484maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
488485maximum m = go Nothing (toStream m)
489486 where
490− go r m1 =
491− let stop = return r
492− yield a Nothing = return $ max_ a r
493− yield a (Just x) = go (max_ a r) x
494− in (S.runStream m1) Nothing stop yield
487+ go res m1 =
488+ let stop = return res
489+ single a = return $ max_ a res
490+ yield a r = go (max_ a res) r
491+ in (S.runStream m1) Nothing stop single yield
495492
496− max_ a r = case r of
493+ max_ a res = case res of
497494 Nothing -> Just a
498495 Just e -> Just $ max a e
499496
@@ -509,11 +506,10 @@ maximum m = go Nothing (toStream m)
509506mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b
510507mapM f m = fromStream $ go (toStream m)
511508 where
512− go m1 = Stream $ \_ stp yld ->
513− let stop = stp
514− yield a Nothing = f a >>= \b -> yld b Nothing
515− yield a (Just x) = f a >>= \b -> yld b (Just (go x))
516− in (S.runStream m1) Nothing stop yield
509+ go m1 = Stream $ \_ stp sng yld ->
510+ let single a = f a >>= sng
511+ yield a r = f a >>= \b -> yld b (go r)
512+ in (S.runStream m1) Nothing stp single yield
517513
518514-- | Apply a monadic action to each element of the stream and discard the
519515-- output of the action.
@@ -522,29 +518,28 @@ mapM_ f m = go (toStream m)
522518 where
523519 go m1 =
524520 let stop = return ()
525− yield a Nothing = void (f a)
526− yield a (Just x) = f a >> go x
527− in (S.runStream m1) Nothing stop yield
521+ single a = void (f a)
522+ yield a r = f a >> go r
523+ in (S.runStream m1) Nothing stop single yield
528524
529525-- | Reduce a stream of monadic actions to a stream of the output of those
530526-- actions.
531527sequence :: (IsStream t, Monad m) => t m (m a) -> t m a
532528sequence m = fromStream $ go (toStream m)
533529 where
534− go m1 = Stream $ \_ stp yld ->
535− let stop = stp
536− yield a Nothing = a >>= \b -> yld b Nothing
537− yield a (Just x) = a >>= \b -> yld b (Just (go x))
538− in (S.runStream m1) Nothing stop yield
530+ go m1 = Stream $ \_ stp sng yld ->
531+ let single ma = ma >>= sng
532+ yield ma r = ma >>= \b -> yld b (go r)
533+ in (S.runStream m1) Nothing stp single yield
539534
540535-- | Generate a stream by performing an action @n@ times.
541536replicateM :: (IsStream t, Monad m) => Int -> m a -> t m a
542537replicateM n m = fromStream $ go n
543538 where
544− go cnt = Stream $ \_ stp yld ->
539+ go cnt = Stream $ \_ stp _ yld ->
545540 if cnt <= 0
546541 then stp
547− else m >>= \a -> yld a (Just $ go (cnt - 1))
542+ else m >>= \a -> yld a (go (cnt - 1))
548543
549544------------------------------------------------------------------------------
550545-- Serially Zipping Streams
@@ -554,16 +549,15 @@ replicateM n m = fromStream $ go n
554549zipWithM :: IsStream t => (a -> b -> t m c) -> t m a -> t m b -> t m c
555550zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
556551 where
557− go mx my = Stream $ \_ stp yld -> do
552+ go mx my = Stream $ \_ stp sng yld -> do
558553 let merge a ra =
559− let yield2 b Nothing = (S.runStream (g a b)) Nothing stp yld
560− yield2 b (Just rb) =
561− (S.runStream (g a b <> go ra rb)) Nothing stp yld
562− in (S.runStream my) Nothing stp yield2
563− let yield1 a Nothing = merge a S.nil
564− yield1 a (Just ra) = merge a ra
565− (S.runStream mx) Nothing stp yield1
566− g a b = toStream $ f a b
554+ let run x = S.runStream x Nothing stp sng yld
555+ single2 b = run $ toStream (f a b)
556+ yield2 b rb = run $ toStream (f a b) <> go ra rb
557+ in (S.runStream my) Nothing stp single2 yield2
558+ let single1 a = merge a S.nil
559+ yield1 a ra = merge a ra
560+ (S.runStream mx) Nothing stp single1 yield1
567561
568562------------------------------------------------------------------------------
569563-- Parallely Zipping Streams
@@ -573,10 +567,10 @@ zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
573567-- concurrently) using a monadic zipping function.
574568zipParallelWithM :: (IsStream t, MonadParallel m)
575569 => (a -> b -> t m c) -> t m a -> t m b -> t m c
576−zipParallelWithM f m1 m2 = fromStream $ Stream $ \_ stp yld -> do
570+zipParallelWithM f m1 m2 = fromStream $ Stream $ \_ stp sng yld -> do
577571 ma <- async m1
578572 mb <- async m2
579− (S.runStream (toStream (zipWithM f ma mb))) Nothing stp yld
573+ (S.runStream (toStream (zipWithM f ma mb))) Nothing stp sng yld
580574
581575{-# DEPRECATED zipAsyncWithM "Please use zipParallelWithM instead." #-}
582576zipAsyncWithM :: (IsStream t, MonadParallel m)