conduit 1.2.1 → 1.2.2

raw patch · 9 files changed

+1401/−339 lines, 9 filesdep +safePVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: safe

API changes (from Hackage documentation)

Files

• Data/Conduit/Internal/Conduit.hs +10/−10
• Data/Conduit/Internal/Fusion.hs +49/−32
• Data/Conduit/Internal/List/Stream.hs +475/−0
• Data/Conduit/Internal/Pipe.hs +5/−5
• Data/Conduit/List.hs +233/−289
• conduit.cabal +8/−3
• fusion-macros.h +23/−0
• test/Data/Conduit/StreamSpec.hs +596/−0
• test/main.hs +2/−0

@@ -735,7 +735,7 @@ ($=) :: Monad m => Conduit a m b -> ConduitM b c m r -> ConduitM a c m r ($=) = (=$=) {-# INLINE [0] ($=) #-}-{-# RULES "$= is =$=" ($=) = (=$=) #-}+{-# RULES "conduit: $= is =$=" ($=) = (=$=) #-}  -- | Right fuse, combining a conduit and a sink together into a new sink. --@@ -751,7 +751,7 @@ (=$) :: Monad m => Conduit a m b -> ConduitM b c m r -> ConduitM a c m r (=$) = (=$=) {-# INLINE [0] (=$) #-}-{-# RULES "=$ is =$=" (=$) = (=$=) #-}+{-# RULES "conduit: =$ is =$=" (=$) = (=$=) #-}  -- | Fusion operator, combining two @Conduit@s together into a new @Conduit@. --@@ -801,7 +801,7 @@     (\i -> unConduitM (g i) rest)     (const $ unConduitM f rest) {-# INLINE await' #-}-{-# RULES "await >>= maybe" forall x y. await >>= maybe x y = await' x y #-}+{-# RULES "conduit: await >>= maybe" forall x y. await >>= maybe x y = await' x y #-}  -- | Send a value downstream to the next component to consume. If the -- downstream component terminates, this call will never return control. If you@@ -1179,14 +1179,14 @@ -- Rewrite rules  {- FIXME-{-# RULES "ConduitM: lift x >>= f" forall m f. lift m >>= f = ConduitM (PipeM (liftM (unConduitM . f) m)) #-}-{-# RULES "ConduitM: lift x >> f" forall m f. lift m >> f = ConduitM (PipeM (liftM (\_ -> unConduitM f) m)) #-}+{-# RULES "conduit: ConduitM: lift x >>= f" forall m f. lift m >>= f = ConduitM (PipeM (liftM (unConduitM . f) m)) #-}+{-# RULES "conduit: ConduitM: lift x >> f" forall m f. lift m >> f = ConduitM (PipeM (liftM (\_ -> unConduitM f) m)) #-} -{-# RULES "ConduitM: liftIO x >>= f" forall m (f :: MonadIO m => a -> ConduitM i o m r). liftIO m >>= f = ConduitM (PipeM (liftM (unConduitM . f) (liftIO m))) #-}-{-# RULES "ConduitM: liftIO x >> f" forall m (f :: MonadIO m => ConduitM i o m r). liftIO m >> f = ConduitM (PipeM (liftM (\_ -> unConduitM f) (liftIO m))) #-}+{-# RULES "conduit: ConduitM: liftIO x >>= f" forall m (f :: MonadIO m => a -> ConduitM i o m r). liftIO m >>= f = ConduitM (PipeM (liftM (unConduitM . f) (liftIO m))) #-}+{-# RULES "conduit: ConduitM: liftIO x >> f" forall m (f :: MonadIO m => ConduitM i o m r). liftIO m >> f = ConduitM (PipeM (liftM (\_ -> unConduitM f) (liftIO m))) #-} -{-# RULES "ConduitM: liftBase x >>= f" forall m (f :: MonadBase b m => a -> ConduitM i o m r). liftBase m >>= f = ConduitM (PipeM (liftM (unConduitM . f) (liftBase m))) #-}-{-# RULES "ConduitM: liftBase x >> f" forall m (f :: MonadBase b m => ConduitM i o m r). liftBase m >> f = ConduitM (PipeM (liftM (\_ -> unConduitM f) (liftBase m))) #-}+{-# RULES "conduit: ConduitM: liftBase x >>= f" forall m (f :: MonadBase b m => a -> ConduitM i o m r). liftBase m >>= f = ConduitM (PipeM (liftM (unConduitM . f) (liftBase m))) #-}+{-# RULES "conduit: ConduitM: liftBase x >> f" forall m (f :: MonadBase b m => ConduitM i o m r). liftBase m >> f = ConduitM (PipeM (liftM (\_ -> unConduitM f) (liftBase m))) #-}  {-# RULES     "yield o >> p" forall o (p :: ConduitM i o m r). yield o >> p = ConduitM (HaveOutput (unConduitM p) (return ()) o)@@ -1198,6 +1198,6 @@         if b then p else ConduitM (HaveOutput (unConduitM p) (return ()) o)   ; "lift m >>= yield" forall m. lift m >>= yield = yieldM m    #-}-{-# RULES "leftover l >> p" forall l (p :: ConduitM i o m r). leftover l >> p =+{-# RULES "conduit: leftover l >> p" forall l (p :: ConduitM i o m r). leftover l >> p =     ConduitM (Leftover (unConduitM p) l) #-}     -}

@@ -1,11 +1,18 @@ {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE BangPatterns #-}+{-# LANGUAGE DeriveFunctor #-} module Data.Conduit.Internal.Fusion     ( -- ** Types       Step (..)     , Stream (..)+    , ConduitWithStream+    , StreamConduitM     , StreamConduit+    , StreamSource+    , StreamProducer+    , StreamSink+    , StreamConsumer       -- ** Functions     , streamConduit     , streamSource@@ -16,9 +23,6 @@ import Data.Conduit.Internal.Conduit import Data.Conduit.Internal.Pipe (Pipe (..)) import Data.Functor.Identity (Identity (runIdentity))-import Control.Monad.Trans.Identity (IdentityT, runIdentityT)-import Control.Monad.Trans.Class (MonadTrans, lift)-import Control.Monad (liftM) import Data.Void (Void, absurd)  -- | This is the same as stream fusion\'s Step. Constructors are renamed to@@ -27,34 +31,47 @@     = Emit s o     | Skip s     | Stop r+    deriving Functor  data Stream m o r = forall s. Stream     (s -> m (Step s o r))     (m s) -data StreamConduit i o m r = StreamConduit+data ConduitWithStream i o m r = ConduitWithStream     (ConduitM i o m r)-    (Stream m i () -> Stream m o r)+    (StreamConduitM i o m r) -unstream :: StreamConduit i o m r -> ConduitM i o m r-unstream (StreamConduit c _) = c+type StreamConduitM i o m r = Stream m i () -> Stream m o r++type StreamConduit i m o = StreamConduitM i o m ()++type StreamSource m o = StreamConduitM () o m ()++type StreamProducer m o = forall i. StreamConduitM i o m ()++type StreamSink i m r = StreamConduitM i Void m r++type StreamConsumer i m r = forall o. StreamConduitM i o m r++unstream :: ConduitWithStream i o m r -> ConduitM i o m r+unstream (ConduitWithStream c _) = c {-# INLINE [0] unstream #-}  fuseStream :: Monad m-           => StreamConduit a b m ()-           -> StreamConduit b c m r-           -> StreamConduit a c m r-fuseStream (StreamConduit a x) (StreamConduit b y) = StreamConduit (a =$= b) (y . x)+           => ConduitWithStream a b m ()+           -> ConduitWithStream b c m r+           -> ConduitWithStream a c m r+fuseStream (ConduitWithStream a x) (ConduitWithStream b y) = ConduitWithStream (a =$= b) (y . x) {-# INLINE fuseStream #-} -{-# RULES "fuseStream" forall left right.+{-# RULES "conduit: fuseStream" forall left right.         unstream left =$= unstream right = unstream (fuseStream left right)   #-}  runStream :: Monad m-          => StreamConduit () Void m r+          => ConduitWithStream () Void m r           -> m r-runStream (StreamConduit _ f) =+runStream (ConduitWithStream _ f) =     run $ f $ Stream emptyStep (return ())   where     emptyStep _ = return $ Stop ()@@ -69,15 +86,15 @@                 Emit _ o -> absurd o {-# INLINE runStream #-} -{-# RULES "runStream" forall stream.+{-# RULES "conduit: runStream" forall stream.         runConduit (unstream stream) = runStream stream   #-}  connectStream :: Monad m-              => StreamConduit () i    m ()-              -> StreamConduit i  Void m r+              => ConduitWithStream () i    m ()+              -> ConduitWithStream i  Void m r               -> m r-connectStream (StreamConduit _ stream) (StreamConduit _ f) =+connectStream (ConduitWithStream _ stream) (ConduitWithStream _ f) =     run $ f $ stream $ Stream emptyStep (return ())   where     emptyStep _ = return $ Stop ()@@ -92,15 +109,15 @@                 Emit _ o -> absurd o {-# INLINE connectStream #-} -{-# RULES "connectStream" forall left right.+{-# RULES "conduit: connectStream" forall left right.         unstream left $$ unstream right = connectStream left right   #-}  connectStream1 :: Monad m-               => StreamConduit () i    m ()-               -> ConduitM      i  Void m r+               => ConduitWithStream () i    m ()+               -> ConduitM          i  Void m r                -> m r-connectStream1 (StreamConduit _ fstream) (ConduitM sink0) =+connectStream1 (ConduitWithStream _ fstream) (ConduitM sink0) =     case fstream $ Stream (const $ return $ Stop ()) (return ()) of         Stream step ms0 ->             let loop _ (Done r) _ = return r@@ -117,7 +134,7 @@              in ms0 >>= loop [] (sink0 Done) {-# INLINE connectStream1 #-} -{-# RULES "connectStream1" forall left right.+{-# RULES "conduit: connectStream1" forall left right.         unstream left $$ right = connectStream1 left right   #-} @@ -129,9 +146,9 @@  connectStream2 :: Monad m                => ConduitM      () i    m ()-               -> StreamConduit i  Void m r+               -> ConduitWithStream i  Void m r                -> m r-connectStream2 (ConduitM src0) (StreamConduit _ fstream) =+connectStream2 (ConduitM src0) (ConduitWithStream _ fstream) =     run $ fstream $ Stream step' $ return (return (), src0 Done)   where     step' (_, Done ()) = return $ Stop ()@@ -148,23 +165,23 @@                 Skip s' -> loop s' {-# INLINE connectStream2 #-} -{-# RULES "connectStream2" forall left right.+{-# RULES "conduit: connectStream2" forall left right.         left $$ unstream right = connectStream2 left right   #-} -}  streamConduit :: ConduitM i o m r               -> (Stream m i () -> Stream m o r)-              -> StreamConduit i o m r-streamConduit = StreamConduit+              -> ConduitWithStream i o m r+streamConduit = ConduitWithStream {-# INLINE CONLIKE streamConduit #-}  streamSource     :: Monad m     => Stream m o ()-    -> StreamConduit i o m ()+    -> ConduitWithStream i o m () streamSource str@(Stream step ms0) =-    StreamConduit con (const str)+    ConduitWithStream con (const str)   where     con = ConduitM $ \rest -> PipeM $ do         s0 <- ms0@@ -180,9 +197,9 @@ streamSourcePure     :: Monad m     => Stream Identity o ()-    -> StreamConduit i o m ()+    -> ConduitWithStream i o m () streamSourcePure (Stream step ms0) =-    StreamConduit con (const $ Stream (return . runIdentity . step) (return s0))+    ConduitWithStream con (const $ Stream (return . runIdentity . step) (return s0))   where     s0 = runIdentity ms0     con = ConduitM $ \rest ->

@@ -0,0 +1,475 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE RankNTypes #-}+module Data.Conduit.Internal.List.Stream where++import           Control.Monad (liftM)+import           Data.Conduit.Internal.Fusion+import qualified Data.Foldable as F++--FIXME: Should streamSource / streamSourcePure be used for sources?++unfoldS :: Monad m+        => (b -> Maybe (a, b))+        -> b+        -> StreamProducer m a+unfoldS f s0 _ =+    Stream step (return s0)+  where+    step s = return $+        case f s of+            Nothing -> Stop ()+            Just (x, s') -> Emit s' x+{-# INLINE unfoldS #-}++unfoldMS :: Monad m+         => (b -> m (Maybe (a, b)))+         -> b+         -> StreamProducer m a+unfoldMS f s0 _ =+    Stream step (return s0)+  where+    step s = do+        ms' <- f s+        return $ case ms' of+            Nothing -> Stop ()+            Just (x, s') -> Emit s' x+{-# INLINE unfoldMS #-}++sourceListS :: Monad m => [a] -> StreamProducer m a+sourceListS xs0 _ =+    Stream (return . step) (return xs0)+  where+    step [] = Stop ()+    step (x:xs) = Emit xs x+{-# INLINE sourceListS #-}++enumFromToS :: (Enum a, Prelude.Ord a, Monad m)+            => a+            -> a+            -> StreamProducer m a+enumFromToS x0 y _ =+    Stream step (return x0)+  where+    step x = return $ if x Prelude.> y+        then Stop ()+        else Emit (Prelude.succ x) x+{-# INLINE [0] enumFromToS #-}++enumFromToS_int :: (Prelude.Integral a, Monad m)+                => a+                -> a+                -> StreamProducer m a+enumFromToS_int x0 y _ = x0 `seq` y `seq` Stream step (return x0)+  where+    step x | x <= y    = return $ Emit (x Prelude.+ 1) x+           | otherwise = return $ Stop ()+{-# INLINE enumFromToS_int #-}++{-# RULES "conduit: enumFromTo<Int>" forall f t.+      enumFromToS f t = enumFromToS_int f t :: Monad m => StreamProducer m Int+  #-}++iterateS :: Monad m => (a -> a) -> a -> StreamProducer m a+iterateS f x0 _ =+    Stream (return . step) (return x0)+  where+    step x = Emit x' x+      where+        x' = f x+{-# INLINE iterateS #-}++replicateS :: Monad m => Int -> a -> StreamProducer m a+replicateS cnt0 a _ =+    Stream step (return cnt0)+  where+    step cnt+        | cnt <= 0  = return $ Stop ()+        | otherwise = return $ Emit (cnt - 1) a+{-# INLINE replicateS #-}++replicateMS :: Monad m => Int -> m a -> StreamProducer m a+replicateMS cnt0 ma _ =+    Stream step (return cnt0)+  where+    step cnt+        | cnt <= 0  = return $ Stop ()+        | otherwise = Emit (cnt - 1) `liftM` ma+{-# INLINE replicateMS #-}++foldS :: Monad m => (b -> a -> b) -> b -> StreamConsumer a m b+foldS f b0 (Stream step ms0) =+    Stream step' (liftM (b0, ) ms0)+  where+    step' (!b, s) = do+        res <- step s+        return $ case res of+            Stop () -> Stop b+            Skip s' -> Skip (b, s')+            Emit s' a -> Skip (f b a, s')+{-# INLINE foldS #-}++foldMS :: Monad m => (b -> a -> m b) -> b -> StreamConsumer a m b+foldMS f b0 (Stream step ms0) =+    Stream step' (liftM (b0, ) ms0)+  where+    step' (!b, s) = do+        res <- step s+        case res of+            Stop () -> return $ Stop b+            Skip s' -> return $ Skip (b, s')+            Emit s' a -> do+                b' <- f b a+                return $ Skip (b', s')+{-# INLINE foldMS #-}++mapM_S :: Monad m+       => (a -> m ())+       -> StreamConsumer a m ()+mapM_S f (Stream step ms0) =+    Stream step' ms0+  where+    step' s = do+        res <- step s+        case res of+          Stop () -> return $ Stop ()+          Skip s' -> return $ Skip s'+          Emit s' x -> f x >> return (Skip s')+{-# INLINE [1] mapM_S #-}++dropS :: Monad m+      => Int+      -> StreamConsumer a m ()+dropS n0 (Stream step ms0) =+    Stream step' (liftM (, n0) ms0)+  where+    step' (_, n) | n <= 0 = return $ Stop ()+    step' (s, n) = do+        res <- step s+        return $ case res of+            Stop () -> Stop ()+            Skip s' -> Skip (s', n)+            Emit s' _ -> Skip (s', n - 1)+{-# INLINE dropS #-}++takeS :: Monad m+      => Int+      -> StreamConsumer a m [a]+takeS n0 (Stream step s0) =+    Stream step' (liftM (id, n0,) s0)+  where+    step' (output, n, _) | n <= 0 = return $ Stop (output [])+    step' (output, n, s) = do+        res <- step s+        return $ case res of+            Stop () -> Stop (output [])+            Skip s' -> Skip (output, n, s')+            Emit s' x -> Skip (output . (x:), n - 1, s')+{-# INLINE takeS #-}++headS :: Monad m => StreamConsumer a m (Maybe a)+headS (Stream step s0) =+    Stream step' s0+  where+    step' s = do+        res <- step s+        return $ case res of+            Stop () -> Stop Nothing+            Skip s' -> Skip s'+            Emit _ x -> Stop (Just x)+{-# INLINE headS #-}++mapS :: Monad m => (a -> b) -> StreamConduit a m b+mapS f (Stream step ms0) =+    Stream step' ms0+  where+    step' s = do+        res <- step s+        return $ case res of+            Stop r -> Stop r+            Emit s' a -> Emit s' (f a)+            Skip s' -> Skip s'+{-# INLINE mapS #-}++mapMS :: Monad m => (a -> m b) -> StreamConduit a m b+mapMS f (Stream step ms0) =+    Stream step' ms0+  where+    step' s = do+        res <- step s+        case res of+            Stop r -> return $ Stop r+            Emit s' a -> Emit s' `liftM` f a+            Skip s' -> return $ Skip s'+{-# INLINE mapMS #-}++iterMS :: Monad m => (a -> m ()) -> StreamConduit a m a+iterMS f (Stream step ms0) =+    Stream step' ms0+  where+    step' s = do+        res <- step s+        case res of+            Stop () -> return $ Stop ()+            Skip s' -> return $ Skip s'+            Emit s' x -> f x >> return (Emit s' x)+{-# INLINE iterMS #-}++mapMaybeS :: Monad m => (a -> Maybe b) -> StreamConduit a m b+mapMaybeS f (Stream step ms0) =+    Stream step' ms0+  where+    step' s = do+        res <- step s+        return $ case res of+            Stop () -> Stop ()+            Skip s' -> Skip s'+            Emit s' x ->+                case f x of+                    Just y -> Emit s' y+                    Nothing -> Skip s'+{-# INLINE mapMaybeS #-}++mapMaybeMS :: Monad m => (a -> m (Maybe b)) -> StreamConduit a m b+mapMaybeMS f (Stream step ms0) =+    Stream step' ms0+  where+    step' s = do+        res <- step s+        case res of+            Stop () -> return $ Stop ()+            Skip s' -> return $ Skip s'+            Emit s' x -> do+                my <- f x+                case my of+                    Just y -> return $ Emit s' y+                    Nothing -> return $ Skip s'+{-# INLINE mapMaybeMS #-}++catMaybesS :: Monad m => StreamConduit (Maybe a) m a+catMaybesS (Stream step ms0) =+    Stream step' ms0+  where+    step' s = do+        res <- step s+        return $ case res of+            Stop () -> Stop ()+            Skip s' -> Skip s'+            Emit s' Nothing -> Skip s'+            Emit s' (Just x) -> Emit s' x+{-# INLINE catMaybesS #-}++concatS :: (Monad m, F.Foldable f) => StreamConduit (f a) m a+concatS (Stream step ms0) =+    Stream step' (liftM ([], ) ms0)+  where+    step' ([], s) = do+        res <- step s+        return $ case res of+            Stop () -> Stop ()+            Skip s' -> Skip ([], s')+            Emit s' x -> Skip (F.toList x, s')+    step' ((x:xs), s) = return (Emit (xs, s) x)+{-# INLINE concatS #-}++concatMapS :: Monad m => (a -> [b]) -> StreamConduit a m b+concatMapS f (Stream step ms0) =+    Stream step' (liftM ([], ) ms0)+  where+    step' ([], s) = do+        res <- step s+        return $ case res of+            Stop () -> Stop ()+            Skip s' -> Skip ([], s')+            Emit s' x -> Skip (f x, s')+    step' ((x:xs), s) = return (Emit (xs, s) x)+{-# INLINE concatMapS #-}++concatMapMS :: Monad m => (a -> m [b]) -> StreamConduit a m b+concatMapMS f (Stream step ms0) =+    Stream step' (liftM ([], ) ms0)+  where+    step' ([], s) = do+        res <- step s+        case res of+            Stop () -> return $ Stop ()+            Skip s' -> return $ Skip ([], s')+            Emit s' x -> do+                xs <- f x+                return $ Skip (xs, s')+    step' ((x:xs), s) = return (Emit (xs, s) x)+{-# INLINE concatMapMS #-}++concatMapAccumS :: Monad m => (a -> accum -> (accum, [b])) -> accum -> StreamConduit a m b+concatMapAccumS f  initial (Stream step ms0) =+    Stream step' (liftM (initial, [], ) ms0)+  where+    step' (accum, [], s) = do+        res <- step s+        return $ case res of+            Stop () -> Stop ()+            Skip s' -> Skip (accum, [], s')+            Emit s' x ->+                let (accum', xs) = f x accum+                in Skip (accum', xs, s')+    step' (accum, (x:xs), s) = return (Emit (accum, xs, s) x)+{-# INLINE concatMapAccumS #-}++mapAccumS :: Monad m => (a -> s -> (s, b)) -> s -> StreamConduitM a b m s+mapAccumS f initial (Stream step ms0) =+    Stream step' (liftM (initial, ) ms0)+  where+    step' (accum, s) = do+        res <- step s+        return $ case res of+            Stop () -> Stop accum+            Skip s' -> Skip (accum, s')+            Emit s' x ->+                let (accum', r) = f x accum+                in Emit (accum', s') r+{-# INLINE mapAccumS #-}++mapAccumMS :: Monad m => (a -> s -> m (s, b)) -> s -> StreamConduitM a b m s+mapAccumMS f initial (Stream step ms0) =+    Stream step' (liftM (initial, ) ms0)+  where+    step' (accum, s) = do+        res <- step s+        case res of+            Stop () -> return $ Stop accum+            Skip s' -> return $ Skip (accum, s')+            Emit s' x -> do+                (accum', r) <- f x accum+                return $ Emit (accum', s') r+{-# INLINE mapAccumMS #-}++concatMapAccumMS :: Monad m => (a -> accum -> m (accum, [b])) -> accum -> StreamConduit a m b+concatMapAccumMS f  initial (Stream step ms0) =+    Stream step' (liftM (initial, [], ) ms0)+  where+    step' (accum, [], s) = do+        res <- step s+        case res of+            Stop () -> return $ Stop ()+            Skip s' -> return $ Skip (accum, [], s')+            Emit s' x -> do+                (accum', xs) <- f x accum+                return $ Skip (accum', xs, s')+    step' (accum, (x:xs), s) = return (Emit (accum, xs, s) x)+{-# INLINE concatMapAccumMS #-}++mapFoldableS :: (Monad m, F.Foldable f) => (a -> f b) -> StreamConduit a m b+mapFoldableS f (Stream step ms0) =+    Stream step' (liftM ([], ) ms0)+  where+    step' ([], s) = do+        res <- step s+        return $ case res of+            Stop () -> Stop ()+            Skip s' -> Skip ([], s')+            Emit s' x -> Skip (F.toList (f x), s')+    step' ((x:xs), s) = return (Emit (xs, s) x)+{-# INLINE mapFoldableS #-}++mapFoldableMS :: (Monad m, F.Foldable f) => (a -> m (f b)) -> StreamConduit a m b+mapFoldableMS f (Stream step ms0) =+    Stream step' (liftM ([], ) ms0)+  where+    step' ([], s) = do+        res <- step s+        case res of+            Stop () -> return $ Stop ()+            Skip s' -> return $ Skip ([], s')+            Emit s' x -> do+                y <- f x+                return $ Skip (F.toList y, s')+    step' ((x:xs), s) = return (Emit (xs, s) x)+{-# INLINE mapFoldableMS #-}++consumeS :: Monad m => StreamConsumer a m [a]+consumeS (Stream step ms0) =+    Stream step' (liftM (id,) ms0)+  where+    step' (front, s) = do+        res <- step s+        return $ case res of+            Stop () -> Stop (front [])+            Skip s' -> Skip (front, s')+            Emit s' a -> Skip (front . (a:), s')+{-# INLINE consumeS #-}++groupByS :: Monad m => (a -> a -> Bool) -> StreamConduit a m [a]+groupByS f = mapS (Prelude.uncurry (:)) . groupBy1S id f+{-# INLINE groupByS #-}++groupOn1S :: (Monad m, Eq b) => (a -> b) -> StreamConduit a m (a, [a])+groupOn1S f = groupBy1S f (==)+{-# INLINE groupOn1S #-}++data GroupByState a b s+     = GBStart s+     | GBLoop ([a] -> [a]) a b s+     | GBDone++groupBy1S :: Monad m => (a -> b) -> (b -> b -> Bool) -> StreamConduit a m (a, [a])+groupBy1S f eq (Stream step ms0) =+    Stream step' (liftM GBStart ms0)+  where+    step' (GBStart s) = do+        res <- step s+        return $ case res of+            Stop () -> Stop ()+            Skip s' -> Skip (GBStart s')+            Emit s' x0 -> Skip (GBLoop id x0 (f x0) s')+    step' (GBLoop rest x0 fx0 s) = do+        res <- step s+        return $ case res of+            Stop () -> Emit GBDone (x0, rest [])+            Skip s' -> Skip (GBLoop rest x0 fx0 s')+            Emit s' x+                | fx0 `eq` f x -> Skip (GBLoop (rest . (x:)) x0 fx0 s')+                | otherwise -> Emit (GBLoop id x (f x) s') (x0, rest [])+    step' GBDone = return $ Stop ()+{-# INLINE groupBy1S #-}++isolateS :: Monad m => Int -> StreamConduit a m a+isolateS count (Stream step ms0) =+    Stream step' (liftM (count,) ms0)+  where+    step' (n, _) | n <= 0 = return $ Stop ()+    step' (n, s) = do+        res <- step s+        return $ case res of+            Stop () -> Stop ()+            Skip s' -> Skip (n, s')+            Emit s' x -> Emit (n - 1, s') x+{-# INLINE isolateS #-}++filterS :: Monad m => (a -> Bool) -> StreamConduit a m a+filterS f (Stream step ms0) =+    Stream step' ms0+  where+    step' s = do+        res <- step s+        return $ case res of+            Stop () -> Stop ()+            Skip s' -> Skip s'+            Emit s' x+                | f x -> Emit s' x+                | otherwise -> Skip s'++sinkNullS :: Monad m => StreamConsumer a m ()+sinkNullS (Stream step ms0) =+    Stream step' ms0+  where+    step' s = do+        res <- step s+        return $ case res of+            Stop () -> Stop ()+            Skip s' -> Skip s'+            Emit s' _ -> Skip s'+{-# INLINE sinkNullS #-}++sourceNullS :: Monad m => StreamProducer m a+sourceNullS _ = Stream (\_ -> return (Stop ())) (return ())+{-# INLINE sourceNullS #-}

@@ -220,7 +220,7 @@ -- Since 0.5.0 await :: Pipe l i o u m (Maybe i) await = NeedInput (Done . Just) (\_ -> Done Nothing)-{-# RULES "CI.await >>= maybe" forall x y. await >>= maybe x y = NeedInput y (const x) #-}+{-# RULES "conduit: CI.await >>= maybe" forall x y. await >>= maybe x y = NeedInput y (const x) #-} {-# INLINE [1] await #-}  -- | This is similar to @await@, but will return the upstream result value as@@ -229,7 +229,7 @@ -- Since 0.5.0 awaitE :: Pipe l i o u m (Either u i) awaitE = NeedInput (Done . Right) (Done . Left)-{-# RULES "awaitE >>= either" forall x y. awaitE >>= either x y = NeedInput y x #-}+{-# RULES "conduit: awaitE >>= either" forall x y. awaitE >>= either x y = NeedInput y x #-} {-# INLINE [1] awaitE #-}  -- | Wait for input forever, calling the given inner @Pipe@ for each piece of@@ -286,7 +286,7 @@ leftover :: l -> Pipe l i o u m () leftover = Leftover (Done ()) {-# INLINE [1] leftover #-}-{-# RULES "leftover l >> p" forall l (p :: Pipe l i o u m r). leftover l >> p = Leftover p l #-}+{-# RULES "conduit: leftover l >> p" forall l (p :: Pipe l i o u m r). leftover l >> p = Leftover p l #-}  -- | Perform some allocation and run an inner @Pipe@. Two guarantees are given -- about resource finalization:@@ -628,5 +628,5 @@     go (Leftover p l) = Leftover (go p) l {-# INLINE generalizeUpstream #-} -{-# RULES "Pipe: lift x >>= f" forall m f. lift m >>= f = PipeM (liftM f m) #-}-{-# RULES "Pipe: lift x >> f" forall m f. lift m >> f = PipeM (liftM (\_ -> f) m) #-}+{-# RULES "conduit: Pipe: lift x >>= f" forall m f. lift m >>= f = PipeM (liftM f m) #-}+{-# RULES "conduit: Pipe: lift x >> f" forall m f. lift m >> f = PipeM (liftM (\_ -> f) m) #-}

@@ -1,6 +1,6 @@ {-# LANGUAGE RankNTypes #-} {-# LANGUAGE BangPatterns #-}-{-# LANGUAGE TupleSections #-}+{-# LANGUAGE CPP #-} -- | Higher-level functions to interact with the elements of a stream. Most of -- these are based on list functions. --@@ -80,34 +80,44 @@ import Data.Monoid (Monoid, mempty, mappend) import qualified Data.Foldable as F import Data.Conduit-import qualified Data.Conduit.Internal as CI import Data.Conduit.Internal.Fusion+import Data.Conduit.Internal.List.Stream+import qualified Data.Conduit.Internal as CI import Control.Monad (when, (<=<), liftM, void) import Control.Monad.Trans.Class (lift) +-- Defines INLINE_RULE0, INLINE_RULE, STREAMING0, and STREAMING.+#include "fusion-macros.h"+ -- | Generate a source from a seed value. --+-- Subject to fusion+-- -- Since 0.4.2-unfold :: Monad m-       => (b -> Maybe (a, b))-       -> b-       -> Producer m a-unfold f =+unfold, unfoldC :: Monad m+                => (b -> Maybe (a, b))+                -> b+                -> Producer m a+unfoldC f =     go   where     go seed =         case f seed of             Just (a, seed') -> yield a >> go seed'             Nothing -> return ()+{-# INLINE unfoldC #-}+STREAMING(unfold, f x)  -- | A monadic unfold. --+-- Subject to fusion+-- -- Since 1.1.2-unfoldM :: Monad m-        => (b -> m (Maybe (a, b)))-        -> b-        -> Producer m a-unfoldM f =+unfoldM, unfoldMC :: Monad m+                  => (b -> m (Maybe (a, b)))+                  -> b+                  -> Producer m a+unfoldMC f =     go   where     go seed = do@@ -115,9 +125,15 @@         case mres of             Just (a, seed') -> yield a >> go seed'             Nothing -> return ()+STREAMING(unfoldM, f seed) -sourceList :: Monad m => [a] -> Producer m a-sourceList = Prelude.mapM_ yield+-- | Yield the values from the list.+--+-- Subject to fusion+sourceList, sourceListC :: Monad m => [a] -> Producer m a+sourceListC = Prelude.mapM_ yield+{-# INLINE sourceListC #-}+STREAMING(sourceList, xs)  -- | Enumerate from a value to a final value, inclusive, via 'succ'. --@@ -128,71 +144,37 @@ -- Subject to fusion -- -- Since 0.4.2-enumFromTo :: (Enum a, Prelude.Ord a, Monad m)-           => a-           -> a-           -> Producer m a-enumFromTo x y = unstream $ streamSource $ enumFromToS x y-{-# INLINE [0] enumFromTo #-}-{-# RULES "unstream enumFromTo" forall x y.-    enumFromTo x y = unstream (streamSourcePure $ enumFromToS x y)-  #-}--enumFromToC :: (Enum a, Prelude.Ord a, Monad m)-            => a-            -> a-            -> Producer m a+enumFromTo, enumFromToC :: (Enum a, Prelude.Ord a, Monad m)+                        => a+                        -> a+                        -> Producer m a enumFromToC x0 y =     loop x0   where     loop x         | x Prelude.> y = return ()         | otherwise = yield x >> loop (Prelude.succ x)-{-# INLINE [0] enumFromToC #-}--enumFromToS :: (Enum a, Prelude.Ord a, Monad m)-            => a-            -> a-            -> Stream m a ()-enumFromToS x0 y =-    Stream step (return x0)-  where-    step x = return $ if x Prelude.> y-        then Stop ()-        else Emit (Prelude.succ x) x-{-# INLINE [0] enumFromToS #-}--enumFromToS_int :: (Prelude.Integral a, Monad m) => a -> a -> Stream m a ()-enumFromToS_int x0 y = x0 `seq` y `seq` Stream step (return x0)-  where-    step x | x <= y    = return $ Emit (x Prelude.+ 1) x-           | otherwise = return $ Stop ()-{-# INLINE enumFromToS_int #-}--{-# RULES "enumFromTo<Int>"-      enumFromToS = enumFromToS_int :: Monad m => Int -> Int -> Stream m Int ()-  #-}+{-# INLINE enumFromToC #-}+STREAMING(enumFromTo, x0 y)  -- | Produces an infinite stream of repeated applications of f to x.-iterate :: Monad m => (a -> a) -> a -> Producer m a-iterate f =+--+-- Subject to fusion+--+iterate, iterateC :: Monad m => (a -> a) -> a -> Producer m a+iterateC f =     go   where     go a = yield a >> go (f a)+{-# INLINE iterateC #-}+STREAMING(iterate, f a)  -- | Replicate a single value the given number of times. -- -- Subject to fusion -- -- Since 1.2.0-replicate :: Monad m => Int -> a -> Producer m a-replicate = replicateC-{-# INLINE [0] replicate #-}-{-# RULES "unstream replicate" forall i a.-     replicate i a = unstream (streamConduit (replicateC i a) (\_ -> replicateS i a))-  #-}--replicateC :: Monad m => Int -> a -> Producer m a+replicate, replicateC :: Monad m => Int -> a -> Producer m a replicateC cnt0 a =     loop cnt0   where@@ -200,27 +182,14 @@         | i <= 0 = return ()         | otherwise = yield a >> loop (i - 1) {-# INLINE replicateC #-}--replicateS :: Monad m => Int -> a -> Stream m a ()-replicateS cnt0 a =-    Stream step (return cnt0)-  where-    step cnt-        | cnt <= 0  = return $ Stop ()-        | otherwise = return $ Emit (cnt - 1) a-{-# INLINE replicateS #-}+STREAMING(replicate, cnt0 a)  -- | Replicate a monadic value the given number of times. --+-- Subject to fusion+-- -- Since 1.2.0-replicateM :: Monad m => Int -> m a -> Producer m a-replicateM = replicateMC-{-# INLINE [0] replicateM #-}-{-# RULES "unstream replicateM" forall i a.-     replicateM i a = unstream (streamConduit (replicateMC i a) (\_ -> replicateMS i a))-  #-}--replicateMC :: Monad m => Int -> m a -> Producer m a+replicateM, replicateMC :: Monad m => Int -> m a -> Producer m a replicateMC cnt0 ma =     loop cnt0   where@@ -228,72 +197,33 @@         | i <= 0 = return ()         | otherwise = lift ma >>= yield >> loop (i - 1) {-# INLINE replicateMC #-}--replicateMS :: Monad m => Int -> m a -> Stream m a ()-replicateMS cnt0 ma =-    Stream step (return cnt0)-  where-    step cnt-        | cnt <= 0  = return $ Stop ()-        | otherwise = Emit (cnt - 1) `liftM` ma-{-# INLINE replicateMS #-}+STREAMING(replicateM, cnt0 ma)  -- | A strict left fold. -- -- Subject to fusion -- -- Since 0.3.0-fold :: Monad m-     => (b -> a -> b)-     -> b-     -> Consumer a m b-fold = foldC-{-# INLINE [0] fold #-}-{-# RULES "unstream fold" forall f b.-        fold f b = unstream (streamConduit (foldC f b) (foldS f b))-  #-}--foldC :: Monad m-      => (b -> a -> b)-      -> b-      -> Consumer a m b+fold, foldC :: Monad m+            => (b -> a -> b)+            -> b+            -> Consumer a m b foldC f =     loop   where     loop !accum = await >>= maybe (return accum) (loop . f accum) {-# INLINE foldC #-}--foldS :: Monad m => (b -> a -> b) -> b -> Stream m a () -> Stream m o b-foldS f b0 (Stream step ms0) =-    Stream step' (liftM (b0, ) ms0)-  where-    step' (!b, s) = do-        res <- step s-        return $ case res of-            Stop () -> Stop b-            Skip s' -> Skip (b, s')-            Emit s' a -> Skip (f b a, s')-{-# INLINE foldS #-}+STREAMING(fold, f accum)  -- | A monadic strict left fold. -- -- Subject to fusion -- -- Since 0.3.0-foldM :: Monad m-      => (b -> a -> m b)-      -> b-      -> Consumer a m b-foldM = foldMC-{-# INLINE [0] foldM #-}-{-# RULES "unstream foldM" forall f b.-        foldM f b = unstream (streamConduit (foldMC f b) (foldMS f b))-  #-}--foldMC :: Monad m-       => (b -> a -> m b)-       -> b-       -> Consumer a m b+foldM, foldMC :: Monad m+              => (b -> a -> m b)+              -> b+              -> Consumer a m b foldMC f =     loop   where@@ -304,20 +234,7 @@             accum' <- lift $ f accum a             accum' `seq` loop accum' {-# INLINE foldMC #-}--foldMS :: Monad m => (b -> a -> m b) -> b -> Stream m a () -> Stream m o b-foldMS f b0 (Stream step ms0) =-    Stream step' (liftM (b0, ) ms0)-  where-    step' (!b, s) = do-        res <- step s-        case res of-            Stop () -> return $ Stop b-            Skip s' -> return $ Skip (b, s')-            Emit s' a -> do-                b' <- f b a-                return $ Skip (b', s')-{-# INLINE foldMS #-}+STREAMING(foldM, f accum)  ----------------------------------------------------------------- -- These are for cases where- for whatever reason- stream fusion cannot be@@ -334,7 +251,7 @@         src <- msrc         go src b {-# INLINE connectFold #-}-{-# RULES "$$ fold" forall src f b. src $$ fold f b = connectFold src f b #-}+{-# RULES "conduit: $$ fold" forall src f b. src $$ fold f b = connectFold src f b #-}  connectFoldM :: Monad m => Source m a -> (b -> a -> m b) -> b -> m b connectFoldM (CI.ConduitM src0) f =@@ -350,19 +267,18 @@         src <- msrc         go src b {-# INLINE connectFoldM #-}-{-# RULES "$$ foldM" forall src f b. src $$ foldM f b = connectFoldM src f b #-}+{-# RULES "conduit: $$ foldM" forall src f b. src $$ foldM f b = connectFoldM src f b #-} -----------------------------------------------------------------  -- | A monoidal strict left fold. --+-- Subject to fusion+-- -- Since 0.5.3 foldMap :: (Monad m, Monoid b)         => (a -> b)         -> Consumer a m b-foldMap f =-    fold combiner mempty-  where-    combiner accum = mappend accum . f+INLINE_RULE(foldMap, f, let combiner accum = mappend accum . f in fold combiner mempty)  -- | A monoidal strict left fold in a Monad. --@@ -370,19 +286,19 @@ foldMapM :: (Monad m, Monoid b)         => (a -> m b)         -> Consumer a m b-foldMapM f =-    foldM combiner mempty-  where-    combiner accum = liftM (mappend accum) . f+INLINE_RULE(foldMapM, f, let combiner accum = liftM (mappend accum) . f in foldM combiner mempty)  -- | Apply the action to all values in the stream. --+-- Subject to fusion+-- -- Since 0.3.0-mapM_ :: Monad m-      => (a -> m ())-      -> Consumer a m ()-mapM_ f = awaitForever $ lift . f-{-# INLINE [1] mapM_ #-}+mapM_, mapM_C :: Monad m+              => (a -> m ())+              -> Consumer a m ()+mapM_C f = awaitForever $ lift . f+{-# INLINE mapM_C #-}+STREAMING(mapM_, f)  srcMapM_ :: Monad m => Source m a -> (a -> m ()) -> m () srcMapM_ (CI.ConduitM src) f =@@ -394,7 +310,7 @@     go (CI.HaveOutput p _ o) = f o >> go p     go (CI.NeedInput _ c) = go (c ()) {-# INLINE srcMapM_ #-}-{-# RULES "connect to mapM_" forall f src. src $$ mapM_ f = srcMapM_ src f #-}+{-# RULES "conduit: connect to mapM_" [2] forall f src. src $$ mapM_ f = srcMapM_ src f #-}  -- | Ignore a certain number of values in the stream. This function is -- semantically equivalent to:@@ -404,15 +320,19 @@ -- However, @drop@ is more efficient as it does not need to hold values in -- memory. --+-- Subject to fusion+-- -- Since 0.3.0-drop :: Monad m-     => Int-     -> Consumer a m ()-drop =+drop, dropC :: Monad m+            => Int+            -> Consumer a m ()+dropC =     loop   where     loop i | i <= 0 = return ()     loop count = await >>= maybe (return ()) (\_ -> loop (count - 1))+{-# INLINE dropC #-}+STREAMING(drop, i)  -- | Take some values from the stream and return as a list. If you want to -- instead create a conduit that pipes data to another sink, see 'isolate'.@@ -420,23 +340,31 @@ -- -- > take i = isolate i =$ consume --+-- Subject to fusion+-- -- Since 0.3.0-take :: Monad m-     => Int-     -> Consumer a m [a]-take =+take, takeC :: Monad m+            => Int+            -> Consumer a m [a]+takeC =     loop id   where     loop front count | count <= 0 = return $ front []     loop front count = await >>= maybe         (return $ front [])-        (\x -> loop (front .(x:)) (count - 1))+        (\x -> loop (front . (x:)) (count - 1))+{-# INLINE takeC #-}+STREAMING(take, i)  -- | Take a single value from the stream, if available. --+-- Subject to fusion+-- -- Since 0.3.0-head :: Monad m => Consumer a m (Maybe a)-head = await+head, headC :: Monad m => Consumer a m (Maybe a)+headC = await+{-# INLINE headC #-}+STREAMING0(head)  -- | Look at the next value in the stream, if available. This function will not -- change the state of the stream.@@ -450,32 +378,14 @@ -- Subject to fusion -- -- Since 0.3.0-map :: Monad m => (a -> b) -> Conduit a m b-map = mapC-{-# INLINE [0] map #-}-{-# RULES "unstream map" forall f.-    map f = unstream (streamConduit (mapC f) (mapS f))-  #-}--mapC :: Monad m => (a -> b) -> Conduit a m b+map, mapC :: Monad m => (a -> b) -> Conduit a m b mapC f = awaitForever $ yield . f {-# INLINE mapC #-}--mapS :: Monad m => (a -> b) -> Stream m a r -> Stream m b r-mapS f (Stream step ms0) =-    Stream step' ms0-  where-    step' s = do-        res <- step s-        return $ case res of-            Stop r -> Stop r-            Emit s' a -> Emit s' (f a)-            Skip s' -> Skip s'-{-# INLINE mapS #-}+STREAMING(map, f)  -- Since a Source never has any leftovers, fusion rules on it are safe. {--{-# RULES "source/map fusion =$=" forall f src. src =$= map f = mapFuseRight src f #-}+{-# RULES "conduit: source/map fusion =$=" forall f src. src =$= map f = mapFuseRight src f #-}  mapFuseRight :: Monad m => Source m a -> (a -> b) -> Source m b mapFuseRight src f = CIC.mapOutput f src@@ -487,18 +397,18 @@ It might be nice to include these rewrite rules, but they may have subtle differences based on leftovers. -{-# RULES "map-to-mapOutput pipeL" forall f src. pipeL src (map f) = mapOutput f src #-}-{-# RULES "map-to-mapOutput $=" forall f src. src $= (map f) = mapOutput f src #-}-{-# RULES "map-to-mapOutput pipe" forall f src. pipe src (map f) = mapOutput f src #-}-{-# RULES "map-to-mapOutput >+>" forall f src. src >+> (map f) = mapOutput f src #-}+{-# RULES "conduit: map-to-mapOutput pipeL" forall f src. pipeL src (map f) = mapOutput f src #-}+{-# RULES "conduit: map-to-mapOutput $=" forall f src. src $= (map f) = mapOutput f src #-}+{-# RULES "conduit: map-to-mapOutput pipe" forall f src. pipe src (map f) = mapOutput f src #-}+{-# RULES "conduit: map-to-mapOutput >+>" forall f src. src >+> (map f) = mapOutput f src #-} -{-# RULES "map-to-mapInput pipeL" forall f sink. pipeL (map f) sink = mapInput f (Prelude.const Prelude.Nothing) sink #-}-{-# RULES "map-to-mapInput =$" forall f sink. map f =$ sink = mapInput f (Prelude.const Prelude.Nothing) sink #-}-{-# RULES "map-to-mapInput pipe" forall f sink. pipe (map f) sink = mapInput f (Prelude.const Prelude.Nothing) sink #-}-{-# RULES "map-to-mapInput >+>" forall f sink. map f >+> sink = mapInput f (Prelude.const Prelude.Nothing) sink #-}+{-# RULES "conduit: map-to-mapInput pipeL" forall f sink. pipeL (map f) sink = mapInput f (Prelude.const Prelude.Nothing) sink #-}+{-# RULES "conduit: map-to-mapInput =$" forall f sink. map f =$ sink = mapInput f (Prelude.const Prelude.Nothing) sink #-}+{-# RULES "conduit: map-to-mapInput pipe" forall f sink. pipe (map f) sink = mapInput f (Prelude.const Prelude.Nothing) sink #-}+{-# RULES "conduit: map-to-mapInput >+>" forall f sink. map f >+> sink = mapInput f (Prelude.const Prelude.Nothing) sink #-} -{-# RULES "map-to-mapOutput =$=" forall f con. con =$= map f = mapOutput f con #-}-{-# RULES "map-to-mapInput =$=" forall f con. map f =$= con = mapInput f (Prelude.const Prelude.Nothing) con #-}+{-# RULES "conduit: map-to-mapOutput =$=" forall f con. con =$= map f = mapOutput f con #-}+{-# RULES "conduit: map-to-mapInput =$=" forall f con. map f =$= con = mapInput f (Prelude.const Prelude.Nothing) con #-}  {-# INLINE [1] map #-} @@ -512,28 +422,10 @@ -- Subject to fusion -- -- Since 0.3.0-mapM :: Monad m => (a -> m b) -> Conduit a m b-mapM = mapMC-{-# INLINE [0] mapM #-}-{-# RULES "unstream mapM" forall f.-    mapM f = unstream (streamConduit (mapMC f) (mapMS f))-  #-}--mapMC :: Monad m => (a -> m b) -> Conduit a m b+mapM, mapMC :: Monad m => (a -> m b) -> Conduit a m b mapMC f = awaitForever $ \a -> lift (f a) >>= yield {-# INLINE mapMC #-}--mapMS :: Monad m => (a -> m b) -> Stream m a r -> Stream m b r-mapMS f (Stream step ms0) =-    Stream step' ms0-  where-    step' s = do-        res <- step s-        case res of-            Stop r -> return $ Stop r-            Emit s' a -> Emit s' `liftM` f a-            Skip s' -> return $ Skip s'-{-# INLINE mapMS #-}+STREAMING(mapM, f)  -- | Apply a monadic action on all values in a stream. --@@ -542,56 +434,88 @@ -- -- > iterM f = mapM (\a -> f a >>= \() -> return a) --+-- Subject to fusion+-- -- Since 0.5.6-iterM :: Monad m => (a -> m ()) -> Conduit a m a-iterM f = awaitForever $ \a -> lift (f a) >> yield a+iterM, iterMC :: Monad m => (a -> m ()) -> Conduit a m a+iterMC f = awaitForever $ \a -> lift (f a) >> yield a+{-# INLINE iterMC #-}+STREAMING(iterM, f)  -- | Apply a transformation that may fail to all values in a stream, discarding -- the failures. --+-- Subject to fusion+-- -- Since 0.5.1-mapMaybe :: Monad m => (a -> Maybe b) -> Conduit a m b-mapMaybe f = awaitForever $ maybe (return ()) yield . f+mapMaybe, mapMaybeC :: Monad m => (a -> Maybe b) -> Conduit a m b+mapMaybeC f = awaitForever $ maybe (return ()) yield . f+{-# INLINE mapMaybeC #-}+STREAMING(mapMaybe, f)  -- | Apply a monadic transformation that may fail to all values in a stream, -- discarding the failures. --+-- Subject to fusion+-- -- Since 0.5.1-mapMaybeM :: Monad m => (a -> m (Maybe b)) -> Conduit a m b-mapMaybeM f = awaitForever $ maybe (return ()) yield <=< lift . f+mapMaybeM, mapMaybeMC :: Monad m => (a -> m (Maybe b)) -> Conduit a m b+mapMaybeMC f = awaitForever $ maybe (return ()) yield <=< lift . f+{-# INLINE mapMaybeMC #-}+STREAMING(mapMaybeM, f)  -- | Filter the @Just@ values from a stream, discarding the @Nothing@  values. --+-- Subject to fusion+-- -- Since 0.5.1-catMaybes :: Monad m => Conduit (Maybe a) m a-catMaybes = awaitForever $ maybe (return ()) yield+catMaybes, catMaybesC :: Monad m => Conduit (Maybe a) m a+catMaybesC = awaitForever $ maybe (return ()) yield+{-# INLINE catMaybesC #-}+STREAMING0(catMaybes)  -- | Generalization of 'catMaybes'. It puts all values from --   'F.Foldable' into stream. --+-- Subject to fusion+-- -- Since 1.0.6-concat :: (Monad m, F.Foldable f) => Conduit (f a) m a-concat = awaitForever $ F.mapM_ yield+concat, concatC :: (Monad m, F.Foldable f) => Conduit (f a) m a+concatC = awaitForever $ F.mapM_ yield+{-# INLINE concatC #-}+STREAMING0(concat)  -- | Apply a transformation to all values in a stream, concatenating the output -- values. --+-- Subject to fusion+-- -- Since 0.3.0-concatMap :: Monad m => (a -> [b]) -> Conduit a m b-concatMap f = awaitForever $ sourceList . f+concatMap, concatMapC :: Monad m => (a -> [b]) -> Conduit a m b+concatMapC f = awaitForever $ sourceList . f+{-# INLINE concatMapC #-}+STREAMING(concatMap, f)  -- | Apply a monadic transformation to all values in a stream, concatenating -- the output values. --+-- Subject to fusion+-- -- Since 0.3.0-concatMapM :: Monad m => (a -> m [b]) -> Conduit a m b-concatMapM f = awaitForever $ sourceList <=< lift . f+concatMapM, concatMapMC :: Monad m => (a -> m [b]) -> Conduit a m b+concatMapMC f = awaitForever $ sourceList <=< lift . f+{-# INLINE concatMapMC #-}+STREAMING(concatMapM, f)  -- | 'concatMap' with an accumulator. --+-- Subject to fusion+-- -- Since 0.3.0-concatMapAccum :: Monad m => (a -> accum -> (accum, [b])) -> accum -> Conduit a m b-concatMapAccum f x0 = void (mapAccum f x0) =$= concat+concatMapAccum, concatMapAccumC :: Monad m => (a -> accum -> (accum, [b])) -> accum -> Conduit a m b+concatMapAccumC f x0 = void (mapAccum f x0) =$= concat+{-# INLINE concatMapAccumC #-}+STREAMING(concatMapAccum, f x0)  -- | Deprecated synonym for @mapAccum@ --@@ -609,21 +533,26 @@  -- | Analog of @mapAccumL@ for lists. --+-- Subject to fusion+-- -- Since 1.1.1-mapAccum :: Monad m => (a -> s -> (s, b)) -> s -> ConduitM a b m s-mapAccum f =+mapAccum, mapAccumC :: Monad m => (a -> s -> (s, b)) -> s -> ConduitM a b m s+mapAccumC f =     loop   where     loop s = await >>= maybe (return s) go       where         go a = case f a s of                  (s', b) -> yield b >> loop s'+STREAMING(mapAccum, f s)  -- | Monadic `mapAccum`. --+-- Subject to fusion+-- -- Since 1.1.1-mapAccumM :: Monad m => (a -> s -> m (s, b)) -> s -> ConduitM a b m s-mapAccumM f =+mapAccumM, mapAccumMC :: Monad m => (a -> s -> m (s, b)) -> s -> ConduitM a b m s+mapAccumMC f =     loop   where     loop s = await >>= maybe (return s) go@@ -631,43 +560,56 @@         go a = do (s', b) <- lift $ f a s                   yield b                   loop s'+{-# INLINE mapAccumMC #-}+STREAMING(mapAccumM, f s)  -- | Analog of 'Prelude.scanl' for lists. --+-- Subject to fusion+-- -- Since 1.1.1 scan :: Monad m => (a -> b -> b) -> b -> ConduitM a b m b-scan f =-    mapAccum $ \a b -> let b' = f a b in (b', b')+INLINE_RULE(scan, f, mapAccum (\a b -> let r = f a b in (r, r)))  -- | Monadic @scanl@. --+-- Subject to fusion+-- -- Since 1.1.1 scanM :: Monad m => (a -> b -> m b) -> b -> ConduitM a b m b-scanM f =-    mapAccumM $ \a b -> do b' <- f a b-                           return (b', b')+INLINE_RULE(scanM, f, mapAccumM (\a b -> f a b >>= \r -> return (r, r)))  -- | 'concatMapM' with an accumulator. --+-- Subject to fusion+-- -- Since 0.3.0-concatMapAccumM :: Monad m => (a -> accum -> m (accum, [b])) -> accum -> Conduit a m b-concatMapAccumM f x0 = void (mapAccumM f x0) =$= concat-+concatMapAccumM, concatMapAccumMC :: Monad m => (a -> accum -> m (accum, [b])) -> accum -> Conduit a m b+concatMapAccumMC f x0 = void (mapAccumM f x0) =$= concat+{-# INLINE concatMapAccumMC #-}+STREAMING(concatMapAccumM, f x0)  -- | Generalization of 'mapMaybe' and 'concatMap'. It applies function -- to all values in a stream and send values inside resulting -- 'Foldable' downstream. --+-- Subject to fusion+-- -- Since 1.0.6-mapFoldable :: (Monad m, F.Foldable f) => (a -> f b) -> Conduit a m b-mapFoldable f = awaitForever $ F.mapM_ yield . f+mapFoldable, mapFoldableC :: (Monad m, F.Foldable f) => (a -> f b) -> Conduit a m b+mapFoldableC f = awaitForever $ F.mapM_ yield . f+{-# INLINE mapFoldableC #-}+STREAMING(mapFoldable, f)  -- | Monadic variant of 'mapFoldable'. --+-- Subject to fusion+-- -- Since 1.0.6-mapFoldableM :: (Monad m, F.Foldable f) => (a -> m (f b)) -> Conduit a m b-mapFoldableM f = awaitForever $ F.mapM_ yield <=< lift . f-+mapFoldableM, mapFoldableMC :: (Monad m, F.Foldable f) => (a -> m (f b)) -> Conduit a m b+mapFoldableMC f = awaitForever $ F.mapM_ yield <=< lift . f+{-# INLINE mapFoldableMC #-}+STREAMING(mapFoldableM, f)  -- | Consume all values from the stream and return as a list. Note that this -- will pull all values into memory. For a lazy variant, see@@ -676,35 +618,21 @@ -- Subject to fusion -- -- Since 0.3.0-consume :: Monad m => Consumer a m [a]-consume = consumeC-{-# INLINE [0] consume #-}-{-# RULES "unstream consume" consume = unstream (streamConduit consumeC consumeS) #-}--consumeC :: Monad m => Consumer a m [a]+consume, consumeC :: Monad m => Consumer a m [a] consumeC =     loop id   where     loop front = await >>= maybe (return $ front []) (\x -> loop $ front . (x:)) {-# INLINE consumeC #-}--consumeS :: Monad m => Stream m a () -> Stream m o [a]-consumeS (Stream step ms0) =-    Stream step' (liftM (id,) ms0)-  where-    step' (front, s) = do-        res <- step s-        return $ case res of-            Stop () -> Stop (front [])-            Skip s' -> Skip (front, s')-            Emit s' a -> Skip (front . (a:), s')-{-# INLINE consumeS #-}+STREAMING0(consume)  -- | Grouping input according to an equality function. --+-- Subject to fusion+-- -- Since 0.3.0-groupBy :: Monad m => (a -> a -> Bool) -> Conduit a m [a]-groupBy f =+groupBy, groupByC :: Monad m => (a -> a -> Bool) -> Conduit a m [a]+groupByC f =     start   where     start = await >>= maybe (return ()) (loop id)@@ -715,7 +643,7 @@         go y             | f x y     = loop (rest . (y:)) x             | otherwise = yield (x : rest []) >> loop id y-+STREAMING(groupBy, f)  -- | 'groupOn1' is similar to @groupBy id@ --@@ -728,11 +656,13 @@ -- > groupOn1 :: (Monad m, Eq b) => (a -> b) -> Conduit a m (NonEmpty a) -- > groupOn1 f = CL.groupOn1 f =$= CL.map (uncurry (:|)) --+-- Subject to fusion+-- -- Since 1.1.7-groupOn1 :: (Monad m, Eq b)-         => (a -> b)-         -> Conduit a m (a, [a])-groupOn1 f =+groupOn1, groupOn1C :: (Monad m, Eq b)+                     => (a -> b)+                     -> Conduit a m (a, [a])+groupOn1C f =     start   where     start = await >>= maybe (return ()) (loop id)@@ -743,7 +673,7 @@         go y             | f x == f y = loop (rest . (y:)) x             | otherwise  = yield (x, rest []) >> loop id y-+STREAMING(groupOn1, f)  -- | Ensure that the inner sink consumes no more than the given number of -- values. Note this this does /not/ ensure that the sink consumes all of those@@ -757,19 +687,25 @@ -- >     someOtherSink -- >     ... --+-- Subject to fusion+-- -- Since 0.3.0-isolate :: Monad m => Int -> Conduit a m a-isolate =+isolate, isolateC :: Monad m => Int -> Conduit a m a+isolateC =     loop   where     loop count | count <= 0 = return ()     loop count = await >>= maybe (return ()) (\x -> yield x >> loop (count - 1))+STREAMING(isolate, count)  -- | Keep only values in the stream passing a given predicate. --+-- Subject to fusion+-- -- Since 0.3.0-filter :: Monad m => (a -> Bool) -> Conduit a m a-filter f = awaitForever $ \i -> when (f i) (yield i)+filter, filterC :: Monad m => (a -> Bool) -> Conduit a m a+filterC f = awaitForever $ \i -> when (f i) (yield i)+STREAMING(filter, f)  filterFuseRight :: Monad m => Source m a -> (a -> Bool) -> Source m a filterFuseRight (CI.ConduitM src) f = CI.ConduitM $ \rest -> let@@ -784,16 +720,19 @@ -- Intermediate finalizers are dropped, but this is acceptable: the next -- yielded value would be demanded by downstream in any event, and that new -- finalizer will always override the existing finalizer.-{-# RULES "source/filter fusion =$=" forall f src. src =$= filter f = filterFuseRight src f #-}+{-# RULES "conduit: source/filter fusion =$=" forall f src. src =$= filter f = filterFuseRight src f #-} {-# INLINE filterFuseRight #-}  -- | Ignore the remainder of values in the source. Particularly useful when -- combined with 'isolate'. --+-- Subject to fusion+-- -- Since 0.3.0-sinkNull :: Monad m => Consumer a m ()-sinkNull = awaitForever $ \_ -> return ()-{-# RULES "connect to sinkNull" forall src. src $$ sinkNull = srcSinkNull src #-}+sinkNull, sinkNullC :: Monad m => Consumer a m ()+sinkNullC = awaitForever $ \_ -> return ()+{-# INLINE sinkNullC #-}+STREAMING0(sinkNull)  srcSinkNull :: Monad m => Source m a -> m () srcSinkNull (CI.ConduitM src) =@@ -805,13 +744,18 @@     go (CI.HaveOutput p _ _) = go p     go (CI.NeedInput _ c) = go (c ()) {-# INLINE srcSinkNull #-}+{-# RULES "conduit: connect to sinkNull" forall src. src $$ sinkNull = srcSinkNull src #-}  -- | A source that outputs no values. Note that this is just a type-restricted -- synonym for 'mempty'. --+-- Subject to fusion+-- -- Since 0.3.0-sourceNull :: Monad m => Producer m a-sourceNull = return ()+sourceNull, sourceNullC :: Monad m => Producer m a+sourceNullC = return ()+{-# INLINE sourceNullC #-}+STREAMING0(sourceNull)  -- | Run a @Pipe@ repeatedly, and output its result value downstream. Stops -- when no more input is available from upstream.

@@ -1,5 +1,5 @@ Name:                conduit-Version:             1.2.1+Version:             1.2.2 Synopsis:            Streaming data processing library. Description:     @conduit@ is a solution to the streaming data problem, allowing for production, transformation, and consumption of streams of data in constant memory. It is an alternative to lazy I\/O which guarantees deterministic resource handling, and fits in the same general solution space as @enumerator@\/@iteratee@ and @pipes@. For a tutorial, please visit <https://haskell.fpcomplete.com/user/snoyberg/library-documentation/conduit-overview>.@@ -13,15 +13,17 @@ Homepage:            http://github.com/snoyberg/conduit extra-source-files:  test/main.hs                    , changelog.md+                   , fusion-macros.h  Library   Exposed-modules:     Data.Conduit                        Data.Conduit.List                        Data.Conduit.Internal                        Data.Conduit.Lift+                       Data.Conduit.Internal.Fusion+                       Data.Conduit.Internal.List.Stream   other-modules:       Data.Conduit.Internal.Pipe                        Data.Conduit.Internal.Conduit-                       Data.Conduit.Internal.Fusion   Build-depends:       base                     >= 4.3          && < 5                      , resourcet                >= 1.1          && < 1.2                      , exceptions               >= 0.6@@ -33,12 +35,14 @@                      , mtl                      , void                     >= 0.5.5                      , mmorph-  ghc-options:     -Wall+  ghc-options:         -Wall+  include-dirs:        .  test-suite test     hs-source-dirs: test     main-is: main.hs     other-modules: Data.Conduit.Extra.ZipConduitSpec+                 , Data.Conduit.StreamSpec     type: exitcode-stdio-1.0     cpp-options:   -DTEST     build-depends:   conduit@@ -51,6 +55,7 @@                    , void                    , containers                    , exceptions >= 0.6+                   , safe     ghc-options:     -Wall  --test-suite doctests

@@ -0,0 +1,23 @@+#define INLINE_RULE0(new,old)             ;\+    new = old                            ;\+    {-# INLINE [0] new #-}               ;\+    {-# RULES "inline new" new = old #-}++#define INLINE_RULE(new,vars,body)                           ;\+    new vars = body                                         ;\+    {-# INLINE [0] new #-}                                  ;\+    {-# RULES "inline new" forall vars. new vars = body #-}++#define STREAMING0(name)                                 ;\+    name = name/**/C                                     ;\+    {-# INLINE [0] name #-}                              ;\+    {-# RULES "unstream name"                             \+      name = unstream (streamConduit name/**/C name/**/S) \+      #-}++#define STREAMING(name,vars)                                                ;\+    name = name/**/C                                                        ;\+    {-# INLINE [0] name #-}                                                 ;\+    {-# RULES "unstream name" forall vars.                                   \+      name vars = unstream (streamConduit (name/**/C vars) (name/**/S vars)) \+      #-}

@@ -0,0 +1,596 @@+{-# LANGUAGE ViewPatterns #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE RankNTypes #-}+module Data.Conduit.StreamSpec where++import           Control.Applicative+import qualified Control.Monad+import           Control.Monad (MonadPlus(..), liftM)+import           Control.Monad.Identity (Identity, runIdentity)+import           Control.Monad.State (StateT(..), get, put)+import           Data.Conduit+import           Data.Conduit.Internal.Fusion+import           Data.Conduit.Internal.List.Stream+import           Data.Conduit.List+import qualified Data.Foldable as F+import qualified Data.Foldable+import           Data.Function (on)+import qualified Data.List+import qualified Data.Maybe+import           Data.Monoid (Monoid(..))+import           Prelude+    ((.), ($), (>>=), (=<<), return, (==), Int, id, Maybe(..), Monad, Bool(..),+     Eq, Show, String, Functor, fst, snd)+import qualified Prelude+import qualified Safe+import           Test.Hspec+import           Test.QuickCheck++spec :: Spec+spec = describe "Comparing list function to" $ do+    qit "unfold" $+        \(getBlind -> f, initial :: Int) ->+            unfold f initial `checkInfiniteProducer`+            (Data.List.unfoldr f initial :: [Int])+    qit "unfoldS" $+        \(getBlind -> f, initial :: Int) ->+            unfoldS f initial `checkInfiniteStreamProducer`+            (Data.List.unfoldr f initial :: [Int])+    qit "unfoldM" $+        \(getBlind -> f, initial :: Int) ->+            unfoldM f initial `checkInfiniteProducerM`+            (unfoldrM f initial :: M [Int])+    qit "unfoldMS" $+        \(getBlind -> f, initial :: Int) ->+            unfoldMS f initial `checkInfiniteStreamProducerM`+            (unfoldrM f initial :: M [Int])+    qit "sourceList" $+        \(xs :: [Int]) ->+            sourceList xs `checkProducer` xs+    qit "sourceListS" $+        \(xs :: [Int]) ->+            sourceListS xs `checkStreamProducer` xs+    qit "enumFromTo" $+        \(fr :: Small Int, to :: Small Int) ->+            enumFromTo fr to `checkProducer`+            Prelude.enumFromTo fr to+    qit "enumFromToS" $+        \(fr :: Small Int, to :: Small Int) ->+            enumFromToS fr to `checkStreamProducer`+            Prelude.enumFromTo fr to+    qit "enumFromToS_int" $+        \(getSmall -> fr :: Int, getSmall -> to :: Int) ->+            enumFromToS_int fr to `checkStreamProducer`+            Prelude.enumFromTo fr to+    qit "iterate" $+        \(getBlind -> f, initial :: Int) ->+            iterate f initial `checkInfiniteProducer`+            Prelude.iterate f initial+    qit "iterateS" $+        \(getBlind -> f, initial :: Int) ->+            iterateS f initial `checkInfiniteStreamProducer`+            Prelude.iterate f initial+    qit "replicate" $+        \(getSmall -> n, getSmall -> x) ->+            replicate n x `checkProducer`+            (Prelude.replicate n x :: [Int])+    qit "replicateS" $+        \(getSmall -> n, getSmall -> x) ->+            replicateS n x `checkStreamProducer`+            (Prelude.replicate n x :: [Int])+    qit "replicateM" $+        \(getSmall -> n, getBlind -> f) ->+            replicateM n f `checkProducerM`+            (Control.Monad.replicateM n f :: M [Int])+    qit "replicateMS" $+        \(getSmall -> n, getBlind -> f) ->+            replicateMS n f `checkStreamProducerM`+            (Control.Monad.replicateM n f :: M [Int])+    qit "fold" $+        \(getBlind -> f, initial :: Int) ->+            fold f initial `checkConsumer`+            Data.List.foldl' f initial+    qit "foldS" $+        \(getBlind -> f, initial :: Int) ->+            foldS f initial `checkStreamConsumer`+            Data.List.foldl' f initial+    qit "foldM" $+        \(getBlind -> f, initial :: Int) ->+            foldM f initial `checkConsumerM`+            (Control.Monad.foldM f initial :: [Int] -> M Int)+    qit "foldMS" $+        \(getBlind -> f, initial :: Int) ->+            foldMS f initial `checkStreamConsumerM`+            (Control.Monad.foldM f initial :: [Int] -> M Int)+    qit "foldMap" $+        \(getBlind -> (f :: Int -> Sum Int)) ->+            foldMap f `checkConsumer`+            Data.Foldable.foldMap f+    qit "mapM_" $+        \(getBlind -> (f :: Int -> M ())) ->+            mapM_ f `checkConsumerM`+            Prelude.mapM_ f+    qit "mapM_S" $+        \(getBlind -> (f :: Int -> M ())) ->+            mapM_S f `checkStreamConsumerM`+            Prelude.mapM_ f+    qit "take" $+        \(getSmall -> n) ->+            take n `checkConsumer`+            Prelude.take n+    qit "takeS" $+        \(getSmall -> n) ->+            takeS n `checkStreamConsumer`+            Prelude.take n+    qit "head" $+        \() ->+            head `checkConsumer`+            Safe.headMay+    qit "headS" $+        \() ->+            headS `checkStreamConsumer`+            Safe.headMay+    qit "peek" $+        \() ->+            peek `checkConsumer`+            Safe.headMay+    qit "map" $+        \(getBlind -> (f :: Int -> Int)) ->+            map f `checkConduit`+            Prelude.map f+    qit "mapS" $+        \(getBlind -> (f :: Int -> Int)) ->+            mapS f `checkStreamConduit`+            Prelude.map f+    qit "mapM" $+        \(getBlind -> (f :: Int -> M Int)) ->+            mapM f `checkConduitM`+            Prelude.mapM f+    qit "mapMS" $+        \(getBlind -> (f :: Int -> M Int)) ->+            mapMS f `checkStreamConduitM`+            Prelude.mapM f+    qit "iterM" $+        \(getBlind -> (f :: Int -> M ())) ->+            iterM f `checkConduitM`+            iterML f+    qit "iterMS" $+        \(getBlind -> (f :: Int -> M ())) ->+            iterMS f `checkStreamConduitM`+            iterML f+    qit "mapMaybe" $+        \(getBlind -> (f :: Int -> Maybe Int)) ->+            mapMaybe f `checkConduit`+            Data.Maybe.mapMaybe f+    qit "mapMaybeS" $+        \(getBlind -> (f :: Int -> Maybe Int)) ->+            mapMaybeS f `checkStreamConduit`+            Data.Maybe.mapMaybe f+    qit "mapMaybeM" $+        \(getBlind -> (f :: Int -> M (Maybe Int))) ->+            mapMaybeM f `checkConduitM`+            mapMaybeML f+    qit "mapMaybeMS" $+        \(getBlind -> (f :: Int -> M (Maybe Int))) ->+            mapMaybeMS f `checkStreamConduitM`+            mapMaybeML f+    qit "catMaybes" $+        \() ->+            catMaybes `checkConduit`+            (Data.Maybe.catMaybes :: [Maybe Int] -> [Int])+    qit "catMaybesS" $+        \() ->+            catMaybesS `checkStreamConduit`+            (Data.Maybe.catMaybes :: [Maybe Int] -> [Int])+    qit "concat" $+        \() ->+            concat `checkConduit`+            (Prelude.concat :: [[Int]] -> [Int])+    qit "concatS" $+        \() ->+            concatS `checkStreamConduit`+            (Prelude.concat :: [[Int]] -> [Int])+    qit "concatMap" $+        \(getBlind -> f) ->+            concatMap f `checkConduit`+            (Prelude.concatMap f :: [Int] -> [Int])+    qit "concatMapS" $+        \(getBlind -> f) ->+            concatMapS f `checkStreamConduit`+            (Prelude.concatMap f :: [Int] -> [Int])+    qit "concatMapM" $+        \(getBlind -> (f :: Int -> M [Int])) ->+            concatMapM f `checkConduitM`+            concatMapML f+    qit "concatMapMS" $+        \(getBlind -> (f :: Int -> M [Int])) ->+            concatMapMS f `checkStreamConduitM`+            concatMapML f+    qit "concatMapAccum" $+        \(getBlind -> (f :: Int -> Int -> (Int, [Int])), initial :: Int) ->+            concatMapAccum f initial `checkConduit`+            concatMapAccumL f initial+    qit "concatMapAccumS" $+        \(getBlind -> (f :: Int -> Int -> (Int, [Int])), initial :: Int) ->+            concatMapAccumS f initial `checkStreamConduit`+            concatMapAccumL f initial+    {-qit "mapAccum" $+        \(getBlind -> (f :: Int -> Int -> (Int, [Int])), initial :: Int) ->+            mapAccum f initial `checkConduitResult`+            mapAccumL f initial-}+    qit "mapAccumS" $+        \(getBlind -> (f :: Int -> Int -> (Int, [Int])), initial :: Int) ->+            mapAccumS f initial `checkStreamConduitResult`+            mapAccumL f initial+    {-qit "mapAccumM" $+        \(getBlind -> (f :: Int -> Int -> M (Int, [Int])), initial :: Int) ->+            mapAccumM f initial `checkConduitResultM`+            mapAccumML f initial-}+    qit "mapAccumMS" $+        \(getBlind -> (f :: Int -> Int -> M (Int, [Int])), initial :: Int) ->+            mapAccumMS f initial `checkStreamConduitResultM`+            mapAccumML f initial+    {-qit "scan" $+        \(getBlind -> (f :: Int -> Int -> Int), initial :: Int) ->+            scan f initial `checkConduitResult`+            scanL f initial-}+    {-qit "scanM" $+        \(getBlind -> (f :: Int -> Int -> M Int), initial :: Int) ->+            scanM f initial `checkConduitResultM`+            scanML f initial-}+    qit "mapFoldable" $+        \(getBlind -> (f :: Int -> [Int])) ->+            mapFoldable f `checkConduit`+            mapFoldableL f+    qit "mapFoldableS" $+        \(getBlind -> (f :: Int -> [Int])) ->+            mapFoldableS f `checkStreamConduit`+            mapFoldableL f+    qit "mapFoldableM" $+        \(getBlind -> (f :: Int -> M [Int])) ->+            mapFoldableM f `checkConduitM`+            mapFoldableML f+    qit "mapFoldableMS" $+        \(getBlind -> (f :: Int -> M [Int])) ->+            mapFoldableMS f `checkStreamConduitM`+            mapFoldableML f+    qit "consume" $+        \() ->+            consume `checkConsumer`+            id+    qit "consumeS" $+        \() ->+            consumeS `checkStreamConsumer`+            id+    qit "groupBy" $+        \(getBlind -> f) ->+            groupBy f `checkConduit`+            (Data.List.groupBy f :: [Int] -> [[Int]])+    qit "groupByS" $+        \(getBlind -> f) ->+            groupByS f `checkStreamConduit`+            (Data.List.groupBy f :: [Int] -> [[Int]])+    qit "groupOn1" $+        \(getBlind -> (f :: Int -> Int)) ->+            groupOn1 f `checkConduit`+            groupOn1L f+    qit "groupOn1S" $+        \(getBlind -> (f :: Int -> Int)) ->+            groupOn1S f `checkStreamConduit`+            groupOn1L f+    qit "isolate" $+        \n ->+            isolate n `checkConduit`+            (Data.List.take n :: [Int] -> [Int])+    qit "isolateS" $+        \n ->+            isolateS n `checkStreamConduit`+            (Data.List.take n :: [Int] -> [Int])+    qit "filter" $+        \(getBlind -> f) ->+            filter f `checkConduit`+            (Data.List.filter f :: [Int] -> [Int])+    qit "filterS" $+        \(getBlind -> f) ->+            filterS f `checkStreamConduit`+            (Data.List.filter f :: [Int] -> [Int])+    qit "sourceNull" $+        \() ->+            sourceNull `checkProducer`+            ([] :: [Int])+    qit "sourceNullS" $+        \() ->+            sourceNullS `checkStreamProducer`+            ([] :: [Int])++qit :: (Arbitrary a, Testable prop, Show a)+    => String -> (a -> prop) -> Spec+qit n f = it n $ property $ forAll arbitrary f++--------------------------------------------------------------------------------+-- Quickcheck utilities for pure conduits / streams++checkProducer :: (Show a, Eq a) => Source Identity a -> [a] -> Property+checkProducer c l  = checkProducerM' runIdentity c (return l)++checkStreamProducer :: (Show a, Eq a) => StreamSource Identity a -> [a] -> Property+checkStreamProducer s l = checkStreamProducerM' runIdentity s (return l)++checkInfiniteProducer :: (Show a, Eq a) => Source Identity a -> [a] -> Property+checkInfiniteProducer c l = checkInfiniteProducerM' runIdentity c (return l)++checkInfiniteStreamProducer :: (Show a, Eq a) => StreamSource Identity a -> [a] -> Property+checkInfiniteStreamProducer s l = checkInfiniteStreamProducerM' runIdentity s (return l)++checkConsumer :: (Show b, Eq b) => Consumer Int Identity b -> ([Int] -> b) -> Property+checkConsumer c l = checkConsumerM' runIdentity c (return . l)++checkStreamConsumer :: (Show b, Eq b) => StreamConsumer Int Identity b -> ([Int] -> b) -> Property+checkStreamConsumer c l = checkStreamConsumerM' runIdentity c (return . l)++checkConduit :: (Show a, Arbitrary a, Show b, Eq b) => Conduit a Identity b -> ([a] -> [b]) -> Property+checkConduit c l = checkConduitM' runIdentity c (return . l)++checkStreamConduit :: (Show a, Arbitrary a, Show b, Eq b) => StreamConduit a Identity b -> ([a] -> [b]) -> Property+checkStreamConduit c l = checkStreamConduitM' runIdentity c (return . l)++-- checkConduitResult :: (Show a, Arbitrary a, Show b, Eq b, Show r, Eq r) => ConduitM a b Identity r -> ([a] -> ([b], r)) -> Property+-- checkConduitResult c l = checkConduitResultM' runIdentity c (return . l)++checkStreamConduitResult :: (Show a, Arbitrary a, Show b, Eq b, Show r, Eq r) => StreamConduitM a b Identity r -> ([a] -> ([b], r)) -> Property+checkStreamConduitResult c l = checkStreamConduitResultM' runIdentity c (return . l)++--------------------------------------------------------------------------------+-- Quickcheck utilities for conduits / streams in the M monad.++checkProducerM :: (Show a, Eq a) => Source M a -> M [a] -> Property+checkProducerM = checkProducerM' runM++checkStreamProducerM :: (Show a, Eq a) => StreamSource M a -> M [a] -> Property+checkStreamProducerM = checkStreamProducerM' runM++checkInfiniteProducerM :: (Show a, Eq a) => Source M a -> M [a] -> Property+checkInfiniteProducerM = checkInfiniteProducerM' (fst . runM)++checkInfiniteStreamProducerM :: (Show a, Eq a) => StreamSource M a -> M [a] -> Property+checkInfiniteStreamProducerM = checkInfiniteStreamProducerM' (fst . runM)++checkConsumerM :: (Show b, Eq b) => Consumer Int M b -> ([Int] -> M b) -> Property+checkConsumerM  = checkConsumerM' runM++checkStreamConsumerM :: (Show b, Eq b) => StreamConsumer Int M b -> ([Int] -> M b) -> Property+checkStreamConsumerM  = checkStreamConsumerM' runM++checkConduitM :: (Show a, Arbitrary a, Show b, Eq b) => Conduit a M b -> ([a] -> M [b]) -> Property+checkConduitM = checkConduitM' runM++checkStreamConduitM :: (Show a, Arbitrary a, Show b, Eq b) => StreamConduit a M b -> ([a] -> M [b]) -> Property+checkStreamConduitM = checkStreamConduitM' runM++-- checkConduitResultM :: (Show a, Arbitrary a, Show b, Eq b, Show r, Eq r) => ConduitM a b M r -> ([a] -> M ([b], r)) -> Property+-- checkConduitResultM = checkConduitResultM' runM++checkStreamConduitResultM :: (Show a, Arbitrary a, Show b, Eq b, Show r, Eq r) => StreamConduitM a b M r -> ([a] -> M ([b], r)) -> Property+checkStreamConduitResultM = checkStreamConduitResultM' runM++--------------------------------------------------------------------------------+-- Quickcheck utilities for monadic streams / conduits+-- These are polymorphic in which Monad is used.++checkProducerM' :: (Show a, Monad m, Show b, Eq b)+                => (m [a] -> b)+                -> Source m a+                -> m [a]+                -> Property+checkProducerM' f c l =+    f (preventFusion c $$ consume)+    ===+    f l++checkStreamProducerM' :: (Show a, Monad m, Show b, Eq b)+                      => (m [a] -> b)+                      -> StreamSource m a+                      -> m [a]+                      -> Property+checkStreamProducerM' f s l =+    f (liftM fst $ evalStream $ s emptyStream)+    ===+    f l++checkInfiniteProducerM' :: (Show a, Monad m, Show b, Eq b)+                        => (m [a] -> b)+                        -> Source m a+                        -> m [a]+                        -> Property+checkInfiniteProducerM' f s l =+    checkProducerM' f+        (preventFusion s $= isolate 10)+        (liftM (Prelude.take 10) l)++checkInfiniteStreamProducerM' :: (Show a, Monad m, Show b, Eq b)+                              => (m [a] -> b)+                              -> StreamSource m a+                              -> m [a]+                              -> Property+checkInfiniteStreamProducerM' f s l =+    f (liftM snd $ evalStream $ takeS 10 $ s emptyStream)+    ===+    f (liftM (Prelude.take 10) l)++checkConsumerM' :: (Show a, Monad m, Show b, Eq b)+                => (m a -> b)+                -> Consumer Int m a+                -> ([Int] -> m a)+                -> Property+checkConsumerM' f c l = forAll arbitrary $ \xs ->+    f (sourceList xs $$ preventFusion c)+    ===+    f (l xs)++checkStreamConsumerM' :: (Show a, Monad m, Show b, Eq b)+                      => (m a -> b)+                      -> StreamConsumer Int m a+                      -> ([Int] -> m a)+                      -> Property+checkStreamConsumerM' f s l = forAll arbitrary $ \xs ->+    f (liftM snd $ evalStream $ s $ sourceListS xs emptyStream)+    ===+    f (l xs)++checkConduitM' :: (Show a, Arbitrary a, Monad m, Show c, Eq c)+               => (m [b] -> c)+               -> Conduit a m b+               -> ([a] -> m [b])+               -> Property+checkConduitM' f c l = forAll arbitrary $ \xs ->+    f (sourceList xs $= preventFusion c $$ consume)+    ===+    f (l xs)++checkStreamConduitM' :: (Show a, Arbitrary a, Monad m, Show c, Eq c)+                     =>  (m [b] -> c)+                     -> StreamConduit a m b+                     -> ([a] -> m [b])+                     -> Property+checkStreamConduitM' f s l = forAll arbitrary $ \xs ->+    f (liftM fst $ evalStream $ s $ sourceListS xs emptyStream)+    ===+    f (l xs)++-- TODO: Fixing this would allow comparing conduit consumers against+-- their list versions.+--+-- checkConduitResultM' :: (Show a, Arbitrary a, Monad m, Show c, Eq c)+--                      => (m ([b], r) -> c)+--                      -> ConduitM a b m r+--                      -> ([a] -> m ([b], r))+--                      -> Property+-- checkConduitResultM' f c l = FIXME forAll arbitrary $ \xs ->+--     f (sourceList xs $= preventFusion c $$ consume)+--     ===+--     f (l xs)++checkStreamConduitResultM' :: (Show a, Arbitrary a, Monad m, Show c, Eq c)+                           =>  (m ([b], r) -> c)+                           -> StreamConduitM a b m r+                           -> ([a] -> m ([b], r))+                           -> Property+checkStreamConduitResultM' f s l = forAll arbitrary $ \xs ->+    f (evalStream $ s $ sourceListS xs emptyStream)+    ===+    f (l xs)++emptyStream :: Monad m => Stream m () ()+emptyStream = Stream (\_ -> return $ Stop ()) (return ())++evalStream :: Monad m => Stream m o r -> m ([o], r)+evalStream (Stream step s0) = go =<< s0+  where+    go s = do+        res <- step s+        case res of+            Stop r -> return ([], r)+            Skip s' -> go s'+            Emit s' x -> liftM (\(l, r) -> (x:l, r)) (go s')++--------------------------------------------------------------------------------+-- Misc utilities++-- Prefer this to creating an orphan instance for Data.Monoid.Sum:++newtype Sum a = Sum a+  deriving (Eq, Show, Arbitrary)++instance Prelude.Num a => Monoid (Sum a) where+  mempty = Sum 0+  mappend (Sum x) (Sum y) = Sum $ x Prelude.+ y++preventFusion :: a -> a+preventFusion = id+{-# INLINE [0] preventFusion #-}++newtype M a = M (StateT Int Identity a)+  deriving (Functor, Applicative, Monad)++instance Arbitrary a => Arbitrary (M a) where+    arbitrary = do+        f <- arbitrary+        return $ do+            s <- M get+            let (x, s') = f s+            M (put s')+            return x++runM :: M a -> (a, Int)+runM (M m) = runIdentity $ runStateT m 0++--------------------------------------------------------------------------------+-- List versions of some functions++iterML :: Monad m => (a -> m ()) -> [a] -> m [a]+iterML f = Prelude.mapM (\a -> f a >>= \() -> return a)++mapMaybeML :: Monad m => (a -> m (Maybe b)) -> [a] -> m [b]+mapMaybeML f = liftM Data.Maybe.catMaybes . Prelude.mapM f++concatMapML :: Monad m => (a -> m [b]) -> [a] -> m [b]+concatMapML f = liftM Prelude.concat . Prelude.mapM f++concatMapAccumL :: (a -> s -> (s, [b])) -> s -> [a] -> [b]+concatMapAccumL f acc0 =+    runIdentity . concatMapAccumML (\a acc -> return $ f a acc) acc0++mapAccumL :: (a -> s -> (s, b)) -> s -> [a] -> ([b], s)+mapAccumL f acc0 =+    runIdentity . mapAccumML (\a acc -> return $ f a acc) acc0++concatMapAccumML :: Monad m => (a -> s -> m (s, [b])) -> s -> [a] -> m [b]+concatMapAccumML f acc0 =+    liftM (Prelude.concat . fst) . mapAccumML f acc0++scanL :: (a -> b -> b) -> b -> [a] -> ([b], b)+scanL f = mapAccumL (\a b -> let r = f a b in (r, r))++scanML :: Monad m => (a -> b -> m b) -> b -> [a] -> m ([b], b)+scanML f = mapAccumML (\a b -> f a b >>= \r -> return (r, r))++mapFoldableL :: F.Foldable f => (a -> f b) -> [a] -> [b]+mapFoldableL f = runIdentity . mapFoldableML (return . f)++mapFoldableML :: (Monad m, F.Foldable f) => (a -> m (f b)) -> [a] -> m [b]+mapFoldableML f = concatMapML (liftM F.toList . f)++groupOn1L :: Eq b => (a -> b) -> [a] -> [(a, [a])]+groupOn1L f =+    Data.List.map (\(x:xs) -> (x, xs)) . Data.List.groupBy ((==) `on` f)++mapAccumML :: Monad m => (a -> s -> m (s, b)) -> s -> [a] -> m ([b], s)+mapAccumML f s0 = go s0+  where+    go s [] = return ([], s)+    go s (x:xs) = do+        (s', r) <- f x s+        liftM (\(l, o) -> (r:l, o)) $ go s' xs++--------------------------------------------------------------------------------+-- Utilities taken from monad-loops package++-- http://hackage.haskell.org/package/monad-loops++-- |See 'Data.List.unfoldr'.  This is a monad-friendly version of that.+unfoldrM :: (Monad m) => (a -> m (Maybe (b,a))) -> a -> m [b]+unfoldrM = unfoldrM'++-- |See 'Data.List.unfoldr'.  This is a monad-friendly version of that, with a+-- twist.  Rather than returning a list, it returns any MonadPlus type of your+-- choice.+unfoldrM' :: (Monad m, MonadPlus f) => (a -> m (Maybe (b,a))) -> a -> m (f b)+unfoldrM' f = go+    where go z = do+            x <- f z+            case x of+                Nothing         -> return mzero+                Just (x', z')   -> do+                        xs <- go z'+                        return (return x' `mplus` xs)

@@ -32,6 +32,7 @@ import Control.Monad.Error (catchError, throwError, Error) import qualified Data.Map as Map import qualified Data.Conduit.Extra.ZipConduitSpec as ZipConduit+import qualified Data.Conduit.StreamSpec as Stream  (@=?) :: (Eq a, Show a) => a -> a -> IO () (@=?) = flip shouldBe@@ -943,6 +944,7 @@             res `shouldBe` [1, 3]      ZipConduit.spec+    Stream.spec  it' :: String -> IO () -> Spec it' = it