diff --git a/Data/Conduit/Internal/Conduit.hs b/Data/Conduit/Internal/Conduit.hs
--- a/Data/Conduit/Internal/Conduit.hs
+++ b/Data/Conduit/Internal/Conduit.hs
@@ -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) #-}
     -}
diff --git a/Data/Conduit/Internal/Fusion.hs b/Data/Conduit/Internal/Fusion.hs
--- a/Data/Conduit/Internal/Fusion.hs
+++ b/Data/Conduit/Internal/Fusion.hs
@@ -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 ->
diff --git a/Data/Conduit/Internal/List/Stream.hs b/Data/Conduit/Internal/List/Stream.hs
new file mode 100644
--- /dev/null
+++ b/Data/Conduit/Internal/List/Stream.hs
@@ -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 #-}
diff --git a/Data/Conduit/Internal/Pipe.hs b/Data/Conduit/Internal/Pipe.hs
--- a/Data/Conduit/Internal/Pipe.hs
+++ b/Data/Conduit/Internal/Pipe.hs
@@ -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) #-}
diff --git a/Data/Conduit/List.hs b/Data/Conduit/List.hs
--- a/Data/Conduit/List.hs
+++ b/Data/Conduit/List.hs
@@ -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.
diff --git a/conduit.cabal b/conduit.cabal
--- a/conduit.cabal
+++ b/conduit.cabal
@@ -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
diff --git a/fusion-macros.h b/fusion-macros.h
new file mode 100644
--- /dev/null
+++ b/fusion-macros.h
@@ -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)) \
+      #-}
diff --git a/test/Data/Conduit/StreamSpec.hs b/test/Data/Conduit/StreamSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Data/Conduit/StreamSpec.hs
@@ -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)
diff --git a/test/main.hs b/test/main.hs
--- a/test/main.hs
+++ b/test/main.hs
@@ -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