conduit 1.1.7 → 1.2.0

raw patch · 13 files changed

+2920/−1697 lines, 13 filesdep +criteriondep +deepseqdep +kan-extensionsdep −bytestringdep −textdep −text-stream-decodedep ~basedep ~exceptionsdep ~hspecPVP ok

version bump matches the API change (PVP)

Dependencies added: criterion, deepseq, kan-extensions, mwc-random, vector

Dependencies removed: bytestring, text, text-stream-decode

Dependency ranges changed: base, exceptions, hspec, mmorph, transformers

API changes (from Hackage documentation)

Files

• Data/Conduit.hs +1/−423
• Data/Conduit/Internal.hs +16/−1148
• Data/Conduit/Internal/Conduit.hs +1194/−0
• Data/Conduit/Internal/Fusion.hs +173/−0
• Data/Conduit/Internal/Pipe.hs +632/−0
• Data/Conduit/Lift.hs +69/−69
• Data/Conduit/List.hs +273/−28
• benchmarks/optimize-201408.hs +427/−0
• benchmarks/unfused.hs +73/−0
• benchmarks/utf8-memory-usage.hs +0/−12
• changelog.md +2/−0
• conduit.cabal +37/−8
• test/main.hs +23/−9

@@ -88,426 +88,4 @@     , sequenceConduits     ) where -import Control.Monad.Trans.Class (lift)-import Data.Conduit.Internal hiding (await, awaitForever, yield, yieldOr, leftover, bracketP, addCleanup, transPipe, mapOutput, mapOutputMaybe, mapInput, yieldM)-import qualified Data.Conduit.Internal as CI-import Control.Monad.Morph (hoist)-import Control.Monad (liftM, forever, when, unless)-import Control.Applicative (Applicative (..))-import Data.Traversable (Traversable (..))-import Control.Monad.Trans.Resource (MonadResource)---- Define fixity of all our operators-infixr 0 $$-infixl 1 $=-infixr 2 =$-infixr 2 =$=-infixr 0 $$+-infixr 0 $$++-infixr 0 $$+--infixl 1 $=+---- | The connect operator, which pulls data from a source and pushes to a sink.--- If you would like to keep the @Source@ open to be used for other--- operations, use the connect-and-resume operator '$$+'.------ Since 0.4.0-($$) :: Monad m => Source m a -> Sink a m b -> m b-src $$ sink = do-    (rsrc, res) <- src $$+ sink-    rsrc $$+- return ()-    return res-{-# INLINE ($$) #-}---- | Left fuse, combining a source and a conduit together into a new source.------ Both the @Source@ and @Conduit@ will be closed when the newly-created--- @Source@ is closed.------ Leftover data from the @Conduit@ will be discarded.------ Note: Since version 1.0.18, this operator has been generalized to be--- identical to @=$=@.------ Since 0.4.0-($=) :: Monad m => Conduit a m b -> ConduitM b c m r -> ConduitM a c m r-ConduitM src $= ConduitM con = ConduitM $ pipeL src con-{-# INLINE ($=) #-}---- | Right fuse, combining a conduit and a sink together into a new sink.------ Both the @Conduit@ and @Sink@ will be closed when the newly-created @Sink@--- is closed.------ Leftover data returned from the @Sink@ will be discarded.------ Note: Since version 1.0.18, this operator has been generalized to be--- identical to @=$=@.------ Since 0.4.0-(=$) :: Monad m => Conduit a m b -> ConduitM b c m r -> ConduitM a c m r-ConduitM con =$ ConduitM sink = ConduitM $ pipeL con sink-{-# INLINE (=$) #-}---- | Fusion operator, combining two @Conduit@s together into a new @Conduit@.------ Both @Conduit@s will be closed when the newly-created @Conduit@ is closed.------ Leftover data returned from the right @Conduit@ will be discarded.------ Since 0.4.0-(=$=) :: Monad m => Conduit a m b -> ConduitM b c m r -> ConduitM a c m r-ConduitM left =$= ConduitM right = ConduitM $ pipeL left right-{-# INLINE (=$=) #-}---- | Wait for a single input value from upstream. If no data is available,--- returns @Nothing@.------ Since 0.5.0-await :: Monad m => Consumer i m (Maybe i)-await = ConduitM CI.await-{-# RULES "await >>= maybe" forall x y. await >>= maybe x y = ConduitM (NeedInput (unConduitM . y) (unConduitM . const x)) #-}-{-# INLINE [1] await #-}---- | Send a value downstream to the next component to consume. If the--- downstream component terminates, this call will never return control. If you--- would like to register a cleanup function, please use 'yieldOr' instead.------ Since 0.5.0-yield :: Monad m-      => o -- ^ output value-      -> ConduitM i o m ()-yield = ConduitM . CI.yield-{-# INLINE [1] yield #-}--yieldM :: Monad m => m o -> ConduitM i o m ()-yieldM = ConduitM . CI.yieldM-{-# INLINE [1] yieldM #-}--{-# RULES-    "yield o >> p" forall o (p :: ConduitM i o m r). yield o >> p = ConduitM (HaveOutput (unConduitM p) (return ()) o)-  ; "mapM_ yield" mapM_ yield = ConduitM . sourceList-  ; "yieldOr o c >> p" forall o c (p :: ConduitM i o m r). yieldOr o c >> p =-        ConduitM (HaveOutput (unConduitM p) c o)-  ; "when yield next" forall b o p. when b (yield o) >> p =-        if b then ConduitM (HaveOutput (unConduitM p) (return ()) o) else p-  ; "unless yield next" forall b o p. unless b (yield o) >> p =-        if b then p else ConduitM (HaveOutput (unConduitM p) (return ()) o)-  ; "lift m >>= yield" forall m. lift m >>= yield = yieldM m-   #-}---- | Provide a single piece of leftover input to be consumed by the next--- component in the current monadic binding.------ /Note/: it is highly encouraged to only return leftover values from input--- already consumed from upstream.------ Since 0.5.0-leftover :: i -> ConduitM i o m ()-leftover = ConduitM . CI.leftover-{-# INLINE [1] leftover #-}-{-# RULES "leftover l >> p" forall l (p :: ConduitM i o m r). leftover l >> p =-    ConduitM (Leftover (unConduitM p) l) #-}---- | Perform some allocation and run an inner component. Two guarantees are--- given about resource finalization:------ 1. It will be /prompt/. The finalization will be run as early as possible.------ 2. It is exception safe. Due to usage of @resourcet@, the finalization will--- be run in the event of any exceptions.------ Since 0.5.0-bracketP :: MonadResource m-         => IO a-         -> (a -> IO ())-         -> (a -> ConduitM i o m r)-         -> ConduitM i o m r-bracketP alloc free inside = ConduitM $ CI.bracketP alloc free $ unConduitM . inside---- | Add some code to be run when the given component cleans up.------ The supplied cleanup function will be given a @True@ if the component ran to--- completion, or @False@ if it terminated early due to a downstream component--- terminating.------ Note that this function is not exception safe. For that, please use--- 'bracketP'.------ Since 0.4.1-addCleanup :: Monad m-           => (Bool -> m ())-           -> ConduitM i o m r-           -> ConduitM i o m r-addCleanup f = ConduitM . CI.addCleanup f . unConduitM---- | Similar to 'yield', but additionally takes a finalizer to be run if the--- downstream component terminates.------ Since 0.5.0-yieldOr :: Monad m-        => o-        -> m () -- ^ finalizer-        -> ConduitM i o m ()-yieldOr o m = ConduitM $ CI.yieldOr o m-{-# INLINE [1] yieldOr #-}---- | Wait for input forever, calling the given inner component for each piece of--- new input. Returns the upstream result type.------ This function is provided as a convenience for the common pattern of--- @await@ing input, checking if it's @Just@ and then looping.------ Since 0.5.0-awaitForever :: Monad m => (i -> ConduitM i o m r) -> ConduitM i o m ()-awaitForever f = ConduitM $ CI.awaitForever (unConduitM . f)---- | Transform the monad that a @ConduitM@ lives in.------ Note that the monad transforming function will be run multiple times,--- resulting in unintuitive behavior in some cases. For a fuller treatment,--- please see:------ <https://github.com/snoyberg/conduit/wiki/Dealing-with-monad-transformers>------ This function is just a synonym for 'hoist'.------ Since 0.4.0-transPipe :: Monad m => (forall a. m a -> n a) -> ConduitM i o m r -> ConduitM i o n r-transPipe = hoist---- | Apply a function to all the output values of a @ConduitM@.------ This mimics the behavior of `fmap` for a `Source` and `Conduit` in pre-0.4--- days. It can also be simulated by fusing with the @map@ conduit from--- "Data.Conduit.List".------ Since 0.4.1-mapOutput :: Monad m => (o1 -> o2) -> ConduitM i o1 m r -> ConduitM i o2 m r-mapOutput f (ConduitM p) = ConduitM $ CI.mapOutput f p---- | Same as 'mapOutput', but use a function that returns @Maybe@ values.------ Since 0.5.0-mapOutputMaybe :: Monad m => (o1 -> Maybe o2) -> ConduitM i o1 m r -> ConduitM i o2 m r-mapOutputMaybe f (ConduitM p) = ConduitM $ CI.mapOutputMaybe f p---- | Apply a function to all the input values of a @ConduitM@.------ Since 0.5.0-mapInput :: Monad m-         => (i1 -> i2) -- ^ map initial input to new input-         -> (i2 -> Maybe i1) -- ^ map new leftovers to initial leftovers-         -> ConduitM i2 o m r-         -> ConduitM i1 o m r-mapInput f g (ConduitM p) = ConduitM $ CI.mapInput f g p---- | The connect-and-resume operator. This does not close the @Source@, but--- instead returns it to be used again. This allows a @Source@ to be used--- incrementally in a large program, without forcing the entire program to live--- in the @Sink@ monad.------ Mnemonic: connect + do more.------ Since 0.5.0-($$+) :: Monad m => Source m a -> Sink a m b -> m (ResumableSource m a, b)-src $$+ sink = connectResume (ResumableSource src (return ())) sink-{-# INLINE ($$+) #-}---- | Continue processing after usage of @$$+@.------ Since 0.5.0-($$++) :: Monad m => ResumableSource m a -> Sink a m b -> m (ResumableSource m a, b)-($$++) = connectResume-{-# INLINE ($$++) #-}---- | Complete processing of a @ResumableSource@. This will run the finalizer--- associated with the @ResumableSource@. In order to guarantee process resource--- finalization, you /must/ use this operator after using @$$+@ and @$$++@.------ Since 0.5.0-($$+-) :: Monad m => ResumableSource m a -> Sink a m b -> m b-rsrc $$+- sink = do-    (ResumableSource _ final, res) <- connectResume rsrc sink-    final-    return res-{-# INLINE ($$+-) #-}---- | Left fusion for a resumable source.------ Since 1.0.16-($=+) :: Monad m => ResumableSource m a -> Conduit a m b -> ResumableSource m b-ResumableSource src final $=+ sink = ResumableSource (src $= sink) final---- | Execute the finalizer associated with a @ResumableSource@, rendering the--- @ResumableSource@ invalid for further use.------ This is just a more explicit version of @$$+- return ()@.------ Since 1.1.3-closeResumableSource :: Monad m => ResumableSource m a -> m ()-closeResumableSource = ($$+- return ())---- | Provide for a stream of data that can be flushed.------ A number of @Conduit@s (e.g., zlib compression) need the ability to flush--- the stream at some point. This provides a single wrapper datatype to be used--- in all such circumstances.------ Since 0.3.0-data Flush a = Chunk a | Flush-    deriving (Show, Eq, Ord)-instance Functor Flush where-    fmap _ Flush = Flush-    fmap f (Chunk a) = Chunk (f a)---- | A wrapper for defining an 'Applicative' instance for 'Source's which allows--- to combine sources together, generalizing 'zipSources'. A combined source--- will take input yielded from each of its @Source@s until any of them stop--- producing output.------ Since 1.0.13-newtype ZipSource m o = ZipSource { getZipSource :: Source m o }--instance Monad m => Functor (ZipSource m) where-    fmap f = ZipSource . mapOutput f . getZipSource-instance Monad m => Applicative (ZipSource m) where-    pure  = ZipSource . forever . yield-    (ZipSource f) <*> (ZipSource x) = ZipSource $ zipSourcesApp f x---- | Coalesce all values yielded by all of the @Source@s.------ Implemented on top of @ZipSource@, see that data type for more details.------ Since 1.0.13-sequenceSources :: (Traversable f, Monad m) => f (Source m o) -> Source m (f o)-sequenceSources = getZipSource . sequenceA . fmap ZipSource---- | A wrapper for defining an 'Applicative' instance for 'Sink's which allows--- to combine sinks together, generalizing 'zipSinks'. A combined sink--- distributes the input to all its participants and when all finish, produces--- the result. This allows to define functions like------ @--- sequenceSinks :: (Monad m)---           => [Sink i m r] -> Sink i m [r]--- sequenceSinks = getZipSink . sequenceA . fmap ZipSink--- @------ Note that the standard 'Applicative' instance for conduits works--- differently. It feeds one sink with input until it finishes, then switches--- to another, etc., and at the end combines their results.------ Since 1.0.13-newtype ZipSink i m r = ZipSink { getZipSink :: Sink i m r }--instance Monad m => Functor (ZipSink i m) where-    fmap f (ZipSink x) = ZipSink (liftM f x)-instance Monad m => Applicative (ZipSink i m) where-    pure  = ZipSink . return-    (ZipSink f) <*> (ZipSink x) =-         ZipSink $ liftM (uncurry ($)) $ zipSinks f x---- | Send incoming values to all of the @Sink@ providing, and ultimately--- coalesce together all return values.------ Implemented on top of @ZipSink@, see that data type for more details.------ Since 1.0.13-sequenceSinks :: (Traversable f, Monad m) => f (Sink i m r) -> Sink i m (f r)-sequenceSinks = getZipSink . sequenceA . fmap ZipSink---- | The connect-and-resume operator. This does not close the @Conduit@, but--- instead returns it to be used again. This allows a @Conduit@ to be used--- incrementally in a large program, without forcing the entire program to live--- in the @Sink@ monad.------ Leftover data returned from the @Sink@ will be discarded.------ Mnemonic: connect + do more.------ Since 1.0.17-(=$$+) :: Monad m => Conduit a m b -> Sink b m r -> Sink a m (ResumableConduit a m b, r)-(=$$+) conduit = connectResumeConduit (ResumableConduit conduit (return ()))-{-# INLINE (=$$+) #-}---- | Continue processing after usage of '=$$+'. Connect a 'ResumableConduit' to--- a sink and return the output of the sink together with a new--- 'ResumableConduit'.------ Since 1.0.17-(=$$++) :: Monad m => ResumableConduit i m o -> Sink o m r -> Sink i m (ResumableConduit i m o, r)-(=$$++) = connectResumeConduit-{-# INLINE (=$$++) #-}---- | Complete processing of a 'ResumableConduit'. This will run the finalizer--- associated with the @ResumableConduit@. In order to guarantee process--- resource finalization, you /must/ use this operator after using '=$$+' and--- '=$$++'.------ Since 1.0.17-(=$$+-) :: Monad m => ResumableConduit i m o -> Sink o m r -> Sink i m r-rsrc =$$+- sink = do-    (ResumableConduit _ final, res) <- connectResumeConduit rsrc sink-    lift final-    return res-{-# INLINE (=$$+-) #-}---infixr 0 =$$+-infixr 0 =$$++-infixr 0 =$$+----- | Provides an alternative @Applicative@ instance for @ConduitM@. In this instance,--- every incoming value is provided to all @ConduitM@s, and output is coalesced together.--- Leftovers from individual @ConduitM@s will be used within that component, and then discarded--- at the end of their computation. Output and finalizers will both be handled in a left-biased manner.------ As an example, take the following program:------ @--- main :: IO ()--- main = do---     let src = mapM_ yield [1..3 :: Int]---         conduit1 = CL.map (+1)---         conduit2 = CL.concatMap (replicate 2)---         conduit = getZipConduit $ ZipConduit conduit1 <* ZipConduit conduit2---         sink = CL.mapM_ print---     src $$ conduit =$ sink--- @------ It will produce the output: 2, 1, 1, 3, 2, 2, 4, 3, 3------ Since 1.0.17-newtype ZipConduit i o m r = ZipConduit { getZipConduit :: ConduitM i o m r }-    deriving Functor-instance Monad m => Applicative (ZipConduit i o m) where-    pure = ZipConduit . pure-    ZipConduit left <*> ZipConduit right = ZipConduit (zipConduitApp left right)---- | Provide identical input to all of the @Conduit@s and combine their outputs--- into a single stream.------ Implemented on top of @ZipConduit@, see that data type for more details.------ Since 1.0.17-sequenceConduits :: (Traversable f, Monad m) => f (ConduitM i o m r) -> ConduitM i o m (f r)-sequenceConduits = getZipConduit . sequenceA . fmap ZipConduit---- | Fuse two @ConduitM@s together, and provide the return value of both. Note--- that this will force the entire upstream @ConduitM@ to be run to produce the--- result value, even if the downstream terminates early.------ Since 1.1.5-fuseBoth :: Monad m => ConduitM a b m r1 -> ConduitM b c m r2 -> ConduitM a c m (r1, r2)-fuseBoth (ConduitM up) (ConduitM down) =-    ConduitM $ pipeL up (withUpstream $ generalizeUpstream down)-{-# INLINE fuseBoth #-}---- | Same as @fuseBoth@, but ignore the return value from the downstream--- @Conduit@. Same caveats of forced consumption apply.------ Since 1.1.5-fuseUpstream :: Monad m => ConduitM a b m r -> Conduit b m c -> ConduitM a c m r-fuseUpstream up down = fmap fst (fuseBoth up down)-{-# INLINE fuseUpstream #-}+import Data.Conduit.Internal.Conduit

@@ -1,1149 +1,17 @@ {-# OPTIONS_HADDOCK not-home #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE CPP #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE UndecidableInstances #-}-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE TupleSections #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE ImpredicativeTypes #-}-module Data.Conduit.Internal-    ( -- * Types-      Pipe (..)-    , ConduitM (..)-    , Source-    , Producer-    , Sink-    , Consumer-    , Conduit-    , ResumableSource (..)-    , ResumableConduit (..)-      -- * Primitives-    , await-    , awaitE-    , awaitForever-    , yield-    , yieldM-    , yieldOr-    , leftover-      -- * Finalization-    , bracketP-    , addCleanup-      -- * Composition-    , idP-    , pipe-    , pipeL-    , connectResume-    , connectResumeConduit-    , runPipe-    , injectLeftovers-    , (>+>)-    , (<+<)-    , fuseLeftovers-    , fuseReturnLeftovers-      -- * Generalizing-    , sourceToPipe-    , sinkToPipe-    , conduitToPipe-    , toProducer-    , toConsumer-      -- * Exceptions-    , catchP-    , handleP-    , tryP-    , catchC-    , handleC-    , tryC-      -- * Utilities-    , transPipe-    , mapOutput-    , mapOutputMaybe-    , mapInput-    , sourceList-    , withUpstream-    , unwrapResumable-    , unwrapResumableConduit-    , newResumableSource-    , newResumableConduit-    , Data.Conduit.Internal.enumFromTo-    , zipSinks-    , zipSources-    , zipSourcesApp-    , zipConduitApp-    , passthroughSink-    , generalizeUpstream-    ) where--import Control.Applicative (Applicative (..))-import Control.Exception.Lifted as E (Exception, catch)-import Control.Monad ((>=>), liftM, ap, when, liftM2)-import Control.Monad.Error.Class(MonadError(..))-import Control.Monad.Reader.Class(MonadReader(..))-import Control.Monad.RWS.Class(MonadRWS())-import Control.Monad.Writer.Class(MonadWriter(..))-import Control.Monad.State.Class(MonadState(..))-import Control.Monad.Trans.Class (MonadTrans (lift))-import Control.Monad.IO.Class (MonadIO (liftIO))-import Control.Monad.Base (MonadBase (liftBase))-import Data.Void (Void, absurd)-import Data.Monoid (Monoid (mappend, mempty))-import Control.Monad.Trans.Resource-import qualified GHC.Exts-import qualified Data.IORef as I-import Control.Monad.Morph (MFunctor (..))-#if MIN_VERSION_exceptions(0, 6, 0)-import qualified Control.Monad.Catch as Catch-#endif---- | The underlying datatype for all the types in this package.  In has six--- type parameters:------ * /l/ is the type of values that may be left over from this @Pipe@. A @Pipe@--- with no leftovers would use @Void@ here, and one with leftovers would use--- the same type as the /i/ parameter. Leftovers are automatically provided to--- the next @Pipe@ in the monadic chain.------ * /i/ is the type of values for this @Pipe@'s input stream.------ * /o/ is the type of values for this @Pipe@'s output stream.------ * /u/ is the result type from the upstream @Pipe@.------ * /m/ is the underlying monad.------ * /r/ is the result type.------ A basic intuition is that every @Pipe@ produces a stream of output values--- (/o/), and eventually indicates that this stream is terminated by sending a--- result (/r/). On the receiving end of a @Pipe@, these become the /i/ and /u/--- parameters.------ Since 0.5.0-data Pipe l i o u m r =-    -- | Provide new output to be sent downstream. This constructor has three-    -- fields: the next @Pipe@ to be used, a finalization function, and the-    -- output value.-    HaveOutput (Pipe l i o u m r) (m ()) o-    -- | Request more input from upstream. The first field takes a new input-    -- value and provides a new @Pipe@. The second takes an upstream result-    -- value, which indicates that upstream is producing no more results.-  | NeedInput (i -> Pipe l i o u m r) (u -> Pipe l i o u m r)-    -- | Processing with this @Pipe@ is complete, providing the final result.-  | Done r-    -- | Require running of a monadic action to get the next @Pipe@.-  | PipeM (m (Pipe l i o u m r))-    -- | Return leftover input, which should be provided to future operations.-  | Leftover (Pipe l i o u m r) l--instance Monad m => Functor (Pipe l i o u m) where-    fmap = liftM--instance Monad m => Applicative (Pipe l i o u m) where-    pure = return-    (<*>) = ap--instance Monad m => Monad (Pipe l i o u m) where-    return = Done--    HaveOutput p c o >>= fp = HaveOutput (p >>= fp)            c          o-    NeedInput p c    >>= fp = NeedInput  (p >=> fp)            (c >=> fp)-    Done x           >>= fp = fp x-    PipeM mp         >>= fp = PipeM      ((>>= fp) `liftM` mp)-    Leftover p i     >>= fp = Leftover   (p >>= fp)            i--instance MonadBase base m => MonadBase base (Pipe l i o u m) where-    liftBase = lift . liftBase--instance MonadTrans (Pipe l i o u) where-    lift mr = PipeM (Done `liftM` mr)--instance MonadIO m => MonadIO (Pipe l i o u m) where-    liftIO = lift . liftIO--instance MonadThrow m => MonadThrow (Pipe l i o u m) where-    throwM = lift . throwM--#if MIN_VERSION_exceptions(0, 6, 0)-instance Catch.MonadCatch m => Catch.MonadCatch (Pipe l i o u m) where-    catch p0 onErr =-        go p0-      where-        go (Done r) = Done r-        go (PipeM mp) = PipeM $ Catch.catch (liftM go mp) (return . onErr)-        go (Leftover p i) = Leftover (go p) i-        go (NeedInput x y) = NeedInput (go . x) (go . y)-        go (HaveOutput p c o) = HaveOutput (go p) c o-    {-# INLINE catch #-}-#endif--instance Monad m => Monoid (Pipe l i o u m ()) where-    mempty = return ()-    mappend = (>>)--instance MonadResource m => MonadResource (Pipe l i o u m) where-    liftResourceT = lift . liftResourceT--instance MonadReader r m => MonadReader r (Pipe l i o u m) where-    ask = lift ask-    local f (HaveOutput p c o) = HaveOutput (local f p) c o-    local f (NeedInput p c) = NeedInput (\i -> local f (p i)) (\u -> local f (c u))-    local _ (Done x) = Done x-    local f (PipeM mp) = PipeM (local f mp)-    local f (Leftover p i) = Leftover (local f p) i---- Provided for doctest-#ifndef MIN_VERSION_mtl-#define MIN_VERSION_mtl(x, y, z) 0-#endif--instance MonadWriter w m => MonadWriter w (Pipe l i o u m) where-#if MIN_VERSION_mtl(2, 1, 0)-    writer = lift . writer-#endif--    tell = lift . tell--    listen (HaveOutput p c o) = HaveOutput (listen p) c o-    listen (NeedInput p c) = NeedInput (\i -> listen (p i)) (\u -> listen (c u))-    listen (Done x) = Done (x,mempty)-    listen (PipeM mp) =-      PipeM $-      do (p,w) <- listen mp-         return $ do (x,w') <- listen p-                     return (x, w `mappend` w')-    listen (Leftover p i) = Leftover (listen p) i--    pass (HaveOutput p c o) = HaveOutput (pass p) c o-    pass (NeedInput p c) = NeedInput (\i -> pass (p i)) (\u -> pass (c u))-    pass (PipeM mp) = PipeM $ mp >>= (return . pass)-    pass (Done (x,_)) = Done x-    pass (Leftover p i) = Leftover (pass p) i--instance MonadState s m => MonadState s (Pipe l i o u m) where-    get = lift get-    put = lift . put-#if MIN_VERSION_mtl(2, 1, 0)-    state = lift . state-#endif--instance MonadRWS r w s m => MonadRWS r w s (Pipe l i o u m)--instance MonadError e m => MonadError e (Pipe l i o u m) where-    throwError = lift . throwError-    catchError (HaveOutput p c o) f = HaveOutput (catchError p f) c o-    catchError (NeedInput p c) f = NeedInput (\i -> catchError (p i) f) (\u -> catchError (c u) f)-    catchError (Done x) _ = Done x-    catchError (PipeM mp) f =-      PipeM $ catchError (liftM (flip catchError f) mp) (\e -> return (f e))-    catchError (Leftover p i) f = Leftover (catchError p f) i---- | Core datatype of the conduit package. This type represents a general--- component which can consume a stream of input values @i@, produce a stream--- of output values @o@, perform actions in the @m@ monad, and produce a final--- result @r@. The type synonyms provided here are simply wrappers around this--- type.------ Since 1.0.0-newtype ConduitM i o m r = ConduitM { unConduitM :: Pipe i i o () m r }-    deriving (Functor, Applicative, Monad, MonadIO, MonadTrans, MonadThrow, MFunctor-#if MIN_VERSION_exceptions(0, 6, 0)-    , Catch.MonadCatch-#endif-    )--instance MonadReader r m => MonadReader r (ConduitM i o m) where-    ask = ConduitM ask-    local f (ConduitM m) = ConduitM (local f m)--instance MonadWriter w m => MonadWriter w (ConduitM i o m) where-#if MIN_VERSION_mtl(2, 1, 0)-    writer = ConduitM . writer-#endif-    tell = ConduitM . tell-    listen (ConduitM m) = ConduitM (listen m)-    pass (ConduitM m) = ConduitM (pass m)--instance MonadState s m => MonadState s (ConduitM i o m) where-    get = ConduitM get-    put = ConduitM . put-#if MIN_VERSION_mtl(2, 1, 0)-    state = ConduitM . state-#endif--instance MonadRWS r w s m => MonadRWS r w s (ConduitM i o m)--instance MonadError e m => MonadError e (ConduitM i o m) where-    throwError = ConduitM . throwError-    catchError (ConduitM m) f = ConduitM $ catchError m (unConduitM . f)--instance MonadBase base m => MonadBase base (ConduitM i o m) where-    liftBase = lift . liftBase--instance MonadResource m => MonadResource (ConduitM i o m) where-    liftResourceT = lift . liftResourceT--instance Monad m => Monoid (ConduitM i o m ()) where-    mempty = return ()-    mappend = (>>)---- | Provides a stream of output values, without consuming any input or--- producing a final result.------ Since 0.5.0-type Source m o = ConduitM () o m ()---- | A component which produces a stream of output values, regardless of the--- input stream. A @Producer@ is a generalization of a @Source@, and can be--- used as either a @Source@ or a @Conduit@.------ Since 1.0.0-type Producer m o = forall i. ConduitM i o m ()---- | Consumes a stream of input values and produces a final result, without--- producing any output.------ > type Sink i m r = ConduitM i Void m r------ Since 0.5.0-type Sink i = ConduitM i Void---- | A component which consumes a stream of input values and produces a final--- result, regardless of the output stream. A @Consumer@ is a generalization of--- a @Sink@, and can be used as either a @Sink@ or a @Conduit@.------ Since 1.0.0-type Consumer i m r = forall o. ConduitM i o m r---- | Consumes a stream of input values and produces a stream of output values,--- without producing a final result.------ Since 0.5.0-type Conduit i m o = ConduitM i o m ()---- | A @Source@ which has been started, but has not yet completed.------ This type contains both the current state of the @Source@, and the finalizer--- to be run to close it.------ Since 0.5.0-data ResumableSource m o = ResumableSource (Source m o) (m ())---- | Since 1.0.13-instance MFunctor ResumableSource where-    hoist nat (ResumableSource src m) = ResumableSource (hoist nat src) (nat m)---- | Wait for a single input value from upstream.------ 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) #-}-{-# INLINE [1] await #-}---- | This is similar to @await@, but will return the upstream result value as--- @Left@ if available.------ 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 #-}-{-# INLINE [1] awaitE #-}---- | Wait for input forever, calling the given inner @Pipe@ for each piece of--- new input. Returns the upstream result type.------ Since 0.5.0-awaitForever :: Monad m => (i -> Pipe l i o r m r') -> Pipe l i o r m r-awaitForever inner =-    self-  where-    self = awaitE >>= either return (\i -> inner i >> self)-{-# INLINE [1] awaitForever #-}---- | Send a single output value downstream. If the downstream @Pipe@--- terminates, this @Pipe@ will terminate as well.------ Since 0.5.0-yield :: Monad m-      => o -- ^ output value-      -> Pipe l i o u m ()-yield = HaveOutput (Done ()) (return ())-{-# INLINE [1] yield #-}--yieldM :: Monad m => m o -> Pipe l i o u m ()-yieldM = PipeM . liftM (HaveOutput (Done ()) (return ()))-{-# INLINE [1] yieldM #-}---- | Similar to @yield@, but additionally takes a finalizer to be run if the--- downstream @Pipe@ terminates.------ Since 0.5.0-yieldOr :: Monad m-        => o-        -> m () -- ^ finalizer-        -> Pipe l i o u m ()-yieldOr o f = HaveOutput (Done ()) f o-{-# INLINE [1] yieldOr #-}--{-# RULES-    "CI.yield o >> p" forall o (p :: Pipe l i o u m r). yield o >> p = HaveOutput p (return ()) o-  ; "mapM_ CI.yield" mapM_ yield = sourceList-  ; "CI.yieldOr o c >> p" forall o c (p :: Pipe l i o u m r). yieldOr o c >> p = HaveOutput p c o-  ; "lift m >>= CI.yield" forall m. lift m >>= yield = yieldM m-  #-}---- | Provide a single piece of leftover input to be consumed by the next pipe--- in the current monadic binding.------ /Note/: it is highly encouraged to only return leftover values from input--- already consumed from upstream.------ Since 0.5.0-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 #-}---- | Perform some allocation and run an inner @Pipe@. Two guarantees are given--- about resource finalization:------ 1. It will be /prompt/. The finalization will be run as early as possible.------ 2. It is exception safe. Due to usage of @resourcet@, the finalization will---    be run in the event of any exceptions.------ Since 0.5.0-bracketP :: MonadResource m-         => IO a-         -> (a -> IO ())-         -> (a -> Pipe l i o u m r)-         -> Pipe l i o u m r-bracketP alloc free inside =-    PipeM start-  where-    start = do-        (key, seed) <- allocate alloc free-        return $ addCleanup (const $ release key) (inside seed)---- | Add some code to be run when the given @Pipe@ cleans up.------ Since 0.4.1-addCleanup :: Monad m-           => (Bool -> m ()) -- ^ @True@ if @Pipe@ ran to completion, @False@ for early termination.-           -> Pipe l i o u m r-           -> Pipe l i o u m r-addCleanup cleanup (Done r) = PipeM (cleanup True >> return (Done r))-addCleanup cleanup (HaveOutput src close x) = HaveOutput-    (addCleanup cleanup src)-    (cleanup False >> close)-    x-addCleanup cleanup (PipeM msrc) = PipeM (liftM (addCleanup cleanup) msrc)-addCleanup cleanup (NeedInput p c) = NeedInput-    (addCleanup cleanup . p)-    (addCleanup cleanup . c)-addCleanup cleanup (Leftover p i) = Leftover (addCleanup cleanup p) i---- | The identity @Pipe@.------ Since 0.5.0-idP :: Monad m => Pipe l a a r m r-idP = NeedInput (HaveOutput idP (return ())) Done---- | Compose a left and right pipe together into a complete pipe. The left pipe--- will be automatically closed when the right pipe finishes.------ Since 0.5.0-pipe :: Monad m => Pipe l a b r0 m r1 -> Pipe Void b c r1 m r2 -> Pipe l a c r0 m r2-pipe =-    goRight (return ())-  where-    goRight final left right =-        case right of-            HaveOutput p c o -> HaveOutput (recurse p) (c >> final) o-            NeedInput rp rc  -> goLeft rp rc final left-            Done r2          -> PipeM (final >> return (Done r2))-            PipeM mp         -> PipeM (liftM recurse mp)-            Leftover _ i     -> absurd i-      where-        recurse = goRight final left--    goLeft rp rc final left =-        case left of-            HaveOutput left' final' o -> goRight final' left' (rp o)-            NeedInput left' lc        -> NeedInput (recurse . left') (recurse . lc)-            Done r1                   -> goRight (return ()) (Done r1) (rc r1)-            PipeM mp                  -> PipeM (liftM recurse mp)-            Leftover left' i          -> Leftover (recurse left') i-      where-        recurse = goLeft rp rc final---- | Same as 'pipe', but automatically applies 'injectLeftovers' to the right @Pipe@.------ Since 0.5.0-pipeL :: Monad m => Pipe l a b r0 m r1 -> Pipe b b c r1 m r2 -> Pipe l a c r0 m r2--- Note: The following should be equivalent to the simpler:------     pipeL l r = l `pipe` injectLeftovers r------ However, this version tested as being significantly more efficient.-pipeL =-    goRight (return ())-  where-    goRight final left right =-        case right of-            HaveOutput p c o  -> HaveOutput (recurse p) (c >> final) o-            NeedInput rp rc   -> goLeft rp rc final left-            Done r2           -> PipeM (final >> return (Done r2))-            PipeM mp          -> PipeM (liftM recurse mp)-            Leftover right' i -> goRight final (HaveOutput left final i) right'-      where-        recurse = goRight final left--    goLeft rp rc final left =-        case left of-            HaveOutput left' final' o -> goRight final' left' (rp o)-            NeedInput left' lc        -> NeedInput (recurse . left') (recurse . lc)-            Done r1                   -> goRight (return ()) (Done r1) (rc r1)-            PipeM mp                  -> PipeM (liftM recurse mp)-            Leftover left' i          -> Leftover (recurse left') i-      where-        recurse = goLeft rp rc final---- | Connect a @Source@ to a @Sink@ until the latter closes. Returns both the--- most recent state of the @Source@ and the result of the @Sink@.------ We use a @ResumableSource@ to keep track of the most recent finalizer--- provided by the @Source@.------ Since 0.5.0-connectResume :: Monad m-              => ResumableSource m o-              -> Sink o m r-              -> m (ResumableSource m o, r)-connectResume (ResumableSource (ConduitM left0) leftFinal0) (ConduitM right0) =-    goRight leftFinal0 left0 right0-  where-    goRight leftFinal left right =-        case right of-            HaveOutput _ _ o -> absurd o-            NeedInput rp rc  -> goLeft rp rc leftFinal left-            Done r2          -> return (ResumableSource (ConduitM left) leftFinal, r2)-            PipeM mp         -> mp >>= goRight leftFinal left-            Leftover p i     -> goRight leftFinal (HaveOutput left leftFinal i) p--    goLeft rp rc leftFinal left =-        case left of-            HaveOutput left' leftFinal' o -> goRight leftFinal' left' (rp o)-            NeedInput _ lc                -> recurse (lc ())-            Done ()                       -> goRight (return ()) (Done ()) (rc ())-            PipeM mp                      -> mp >>= recurse-            Leftover p ()                 -> recurse p-      where-        recurse = goLeft rp rc leftFinal---- | Run a pipeline until processing completes.------ Since 0.5.0-runPipe :: Monad m => Pipe Void () Void () m r -> m r-runPipe (HaveOutput _ _ o) = absurd o-runPipe (NeedInput _ c) = runPipe (c ())-runPipe (Done r) = return r-runPipe (PipeM mp) = mp >>= runPipe-runPipe (Leftover _ i) = absurd i---- | Transforms a @Pipe@ that provides leftovers to one which does not,--- allowing it to be composed.------ This function will provide any leftover values within this @Pipe@ to any--- calls to @await@. If there are more leftover values than are demanded, the--- remainder are discarded.------ Since 0.5.0-injectLeftovers :: Monad m => Pipe i i o u m r -> Pipe l i o u m r-injectLeftovers =-    go []-  where-    go ls (HaveOutput p c o) = HaveOutput (go ls p) c o-    go (l:ls) (NeedInput p _) = go ls $ p l-    go [] (NeedInput p c) = NeedInput (go [] . p) (go [] . c)-    go _ (Done r) = Done r-    go ls (PipeM mp) = PipeM (liftM (go ls) mp)-    go ls (Leftover p l) = go (l:ls) p---- | Transform the monad that a @Pipe@ lives in.------ Note that the monad transforming function will be run multiple times,--- resulting in unintuitive behavior in some cases. For a fuller treatment,--- please see:------ <https://github.com/snoyberg/conduit/wiki/Dealing-with-monad-transformers>------ This function is just a synonym for 'hoist'.------ Since 0.4.0-transPipe :: Monad m => (forall a. m a -> n a) -> Pipe l i o u m r -> Pipe l i o u n r-transPipe f (HaveOutput p c o) = HaveOutput (transPipe f p) (f c) o-transPipe f (NeedInput p c) = NeedInput (transPipe f . p) (transPipe f . c)-transPipe _ (Done r) = Done r-transPipe f (PipeM mp) =-    PipeM (f $ liftM (transPipe f) $ collapse mp)-  where-    -- Combine a series of monadic actions into a single action.  Since we-    -- throw away side effects between different actions, an arbitrary break-    -- between actions will lead to a violation of the monad transformer laws.-    -- Example available at:-    ---    -- http://hpaste.org/75520-    collapse mpipe = do-        pipe' <- mpipe-        case pipe' of-            PipeM mpipe' -> collapse mpipe'-            _ -> return pipe'-transPipe f (Leftover p i) = Leftover (transPipe f p) i---- | Apply a function to all the output values of a @Pipe@.------ This mimics the behavior of `fmap` for a `Source` and `Conduit` in pre-0.4--- days.------ Since 0.4.1-mapOutput :: Monad m => (o1 -> o2) -> Pipe l i o1 u m r -> Pipe l i o2 u m r-mapOutput f =-    go-  where-    go (HaveOutput p c o) = HaveOutput (go p) c (f o)-    go (NeedInput p c) = NeedInput (go . p) (go . c)-    go (Done r) = Done r-    go (PipeM mp) = PipeM (liftM (go) mp)-    go (Leftover p i) = Leftover (go p) i-{-# INLINE mapOutput #-}---- | Same as 'mapOutput', but use a function that returns @Maybe@ values.------ Since 0.5.0-mapOutputMaybe :: Monad m => (o1 -> Maybe o2) -> Pipe l i o1 u m r -> Pipe l i o2 u m r-mapOutputMaybe f =-    go-  where-    go (HaveOutput p c o) = maybe id (\o' p' -> HaveOutput p' c o') (f o) (mapOutputMaybe f p)-    go (NeedInput p c) = NeedInput (go . p) (go . c)-    go (Done r) = Done r-    go (PipeM mp) = PipeM (liftM (go) mp)-    go (Leftover p i) = Leftover (go p) i-{-# INLINE mapOutputMaybe #-}---- | Apply a function to all the input values of a @Pipe@.------ Since 0.5.0-mapInput :: Monad m-         => (i1 -> i2) -- ^ map initial input to new input-         -> (l2 -> Maybe l1) -- ^ map new leftovers to initial leftovers-         -> Pipe l2 i2 o u m r-         -> Pipe l1 i1 o u m r-mapInput f f' (HaveOutput p c o) = HaveOutput (mapInput f f' p) c o-mapInput f f' (NeedInput p c)    = NeedInput (mapInput f f' . p . f) (mapInput f f' . c)-mapInput _ _  (Done r)           = Done r-mapInput f f' (PipeM mp)         = PipeM (liftM (mapInput f f') mp)-mapInput f f' (Leftover p i)     = maybe id (flip Leftover) (f' i) $ mapInput f f' p--enumFromTo :: (Enum o, Eq o, Monad m)-           => o-           -> o-           -> Pipe l i o u m ()-enumFromTo start stop =-    loop start-  where-    loop i-        | i == stop = HaveOutput (Done ()) (return ()) i-        | otherwise = HaveOutput (loop (succ i)) (return ()) i-{-# INLINE enumFromTo #-}---- | Convert a list into a source.------ Since 0.3.0-sourceList :: Monad m => [a] -> Pipe l i a u m ()-sourceList =-    go-  where-    go [] = Done ()-    go (o:os) = HaveOutput (go os) (return ()) o-{-# INLINE [1] sourceList #-}---- | The equivalent of @GHC.Exts.build@ for @Pipe@.------ Since 0.4.2-build :: Monad m => (forall b. (o -> b -> b) -> b -> b) -> Pipe l i o u m ()-build g = g (\o p -> HaveOutput p (return ()) o) (return ())--{-# RULES-    "sourceList/build" forall (f :: (forall b. (a -> b -> b) -> b -> b)). sourceList (GHC.Exts.build f) = build f #-}--sourceToPipe :: Monad m => Source m o -> Pipe l i o u m ()-sourceToPipe =-    go . unConduitM-  where-    go (HaveOutput p c o) = HaveOutput (go p) c o-    go (NeedInput _ c) = go $ c ()-    go (Done ()) = Done ()-    go (PipeM mp) = PipeM (liftM go mp)-    go (Leftover p ()) = go p--sinkToPipe :: Monad m => Sink i m r -> Pipe l i o u m r-sinkToPipe =-    go . injectLeftovers . unConduitM-  where-    go (HaveOutput _ _ o) = absurd o-    go (NeedInput p c) = NeedInput (go . p) (const $ go $ c ())-    go (Done r) = Done r-    go (PipeM mp) = PipeM (liftM go mp)-    go (Leftover _ l) = absurd l--conduitToPipe :: Monad m => Conduit i m o -> Pipe l i o u m ()-conduitToPipe =-    go . injectLeftovers . unConduitM-  where-    go (HaveOutput p c o) = HaveOutput (go p) c o-    go (NeedInput p c) = NeedInput (go . p) (const $ go $ c ())-    go (Done ()) = Done ()-    go (PipeM mp) = PipeM (liftM go mp)-    go (Leftover _ l) = absurd l---- | Returns a tuple of the upstream and downstream results. Note that this--- will force consumption of the entire input stream.------ Since 0.5.0-withUpstream :: Monad m-             => Pipe l i o u m r-             -> Pipe l i o u m (u, r)-withUpstream down =-    down >>= go-  where-    go r =-        loop-      where-        loop = awaitE >>= either (\u -> return (u, r)) (\_ -> loop)---- | Unwraps a @ResumableSource@ into a @Source@ and a finalizer.------ A @ResumableSource@ represents a @Source@ which has already been run, and--- therefore has a finalizer registered. As a result, if we want to turn it--- into a regular @Source@, we need to ensure that the finalizer will be run--- appropriately. By appropriately, I mean:------ * If a new finalizer is registered, the old one should not be called.------ * If the old one is called, it should not be called again.------ This function returns both a @Source@ and a finalizer which ensures that the--- above two conditions hold. Once you call that finalizer, the @Source@ is--- invalidated and cannot be used.------ Since 0.5.2-unwrapResumable :: MonadIO m => ResumableSource m o -> m (Source m o, m ())-unwrapResumable (ResumableSource src final) = do-    ref <- liftIO $ I.newIORef True-    let final' = do-            x <- liftIO $ I.readIORef ref-            when x final-    return (liftIO (I.writeIORef ref False) >> src, final')---- | Turn a @Source@ into a @ResumableSource@ with no attached finalizer.------ Since 1.1.4-newResumableSource :: Monad m => Source m o -> ResumableSource m o-newResumableSource s = ResumableSource s (return ())--infixr 9 <+<-infixl 9 >+>---- | Fuse together two @Pipe@s, connecting the output from the left to the--- input of the right.------ Notice that the /leftover/ parameter for the @Pipe@s must be @Void@. This--- ensures that there is no accidental data loss of leftovers during fusion. If--- you have a @Pipe@ with leftovers, you must first call 'injectLeftovers'.------ Since 0.5.0-(>+>) :: Monad m => Pipe l a b r0 m r1 -> Pipe Void b c r1 m r2 -> Pipe l a c r0 m r2-(>+>) = pipe-{-# INLINE (>+>) #-}---- | Same as '>+>', but reverse the order of the arguments.------ Since 0.5.0-(<+<) :: Monad m => Pipe Void b c r1 m r2 -> Pipe l a b r0 m r1 -> Pipe l a c r0 m r2-(<+<) = flip pipe-{-# INLINE (<+<) #-}---- | Generalize a 'Source' to a 'Producer'.------ Since 1.0.0-toProducer :: Monad m => Source m a -> Producer m a-toProducer =-    ConduitM . go . unConduitM-  where-    go (HaveOutput p c o) = HaveOutput (go p) c o-    go (NeedInput _ c) = go (c ())-    go (Done r) = Done r-    go (PipeM mp) = PipeM (liftM go mp)-    go (Leftover p ()) = go p---- | Generalize a 'Sink' to a 'Consumer'.------ Since 1.0.0-toConsumer :: Monad m => Sink a m b -> Consumer a m b-toConsumer =-    ConduitM . go . unConduitM-  where-    go (HaveOutput _ _ o) = absurd o-    go (NeedInput p c) = NeedInput (go . p) (go . c)-    go (Done r) = Done r-    go (PipeM mp) = PipeM (liftM go mp)-    go (Leftover p l) = Leftover (go p) l---- | Since 1.0.4-instance MFunctor (Pipe l i o u) where-    hoist = transPipe---- | See 'catchC' for more details.------ Since 1.0.11-catchP :: (MonadBaseControl IO m, Exception e)-       => Pipe l i o u m r-       -> (e -> Pipe l i o u m r)-       -> Pipe l i o u m r-catchP p0 onErr =-    go p0-  where-    go (Done r) = Done r-    go (PipeM mp) = PipeM $ E.catch (liftM go mp) (return . onErr)-    go (Leftover p i) = Leftover (go p) i-    go (NeedInput x y) = NeedInput (go . x) (go . y)-    go (HaveOutput p c o) = HaveOutput (go p) c o-{-# INLINABLE catchP #-}---- | The same as @flip catchP@.------ Since 1.0.11-handleP :: (MonadBaseControl IO m, Exception e)-        => (e -> Pipe l i o u m r)-        -> Pipe l i o u m r-        -> Pipe l i o u m r-handleP = flip catchP-{-# INLINE handleP #-}---- | See 'tryC' for more details.------ Since 1.0.11-tryP :: (MonadBaseControl IO m, Exception e)-     => Pipe l i o u m r-     -> Pipe l i o u m (Either e r)-tryP =-    go-  where-    go (Done r) = Done (Right r)-    go (PipeM mp) = PipeM $ E.catch (liftM go mp) (return . Done . Left)-    go (Leftover p i) = Leftover (go p) i-    go (NeedInput x y) = NeedInput (go . x) (go . y)-    go (HaveOutput p c o) = HaveOutput (go p) c o-{-# INLINABLE tryP #-}---- | Catch all exceptions thrown by the current component of the pipeline.------ Note: this will /not/ catch exceptions thrown by other components! For--- example, if an exception is thrown in a @Source@ feeding to a @Sink@, and--- the @Sink@ uses @catchC@, the exception will /not/ be caught.------ Due to this behavior (as well as lack of async exception handling), you--- should not try to implement combinators such as @onException@ in terms of this--- primitive function.------ Note also that the exception handling will /not/ be applied to any--- finalizers generated by this conduit.------ Since 1.0.11-catchC :: (MonadBaseControl IO m, Exception e)-       => ConduitM i o m r-       -> (e -> ConduitM i o m r)-       -> ConduitM i o m r-catchC (ConduitM p) f = ConduitM (catchP p (unConduitM . f))-{-# INLINE catchC #-}---- | The same as @flip catchC@.------ Since 1.0.11-handleC :: (MonadBaseControl IO m, Exception e)-        => (e -> ConduitM i o m r)-        -> ConduitM i o m r-        -> ConduitM i o m r-handleC = flip catchC-{-# INLINE handleC #-}---- | A version of @try@ for use within a pipeline. See the comments in @catchC@--- for more details.------ Since 1.0.11-tryC :: (MonadBaseControl IO m, Exception e)-     => ConduitM i o m r-     -> ConduitM i o m (Either e r)-tryC = ConduitM . tryP . unConduitM-{-# INLINE tryC #-}---- | Combines two sinks. The new sink will complete when both input sinks have---   completed.------ Any leftovers are discarded.------ Since 0.4.1-zipSinks :: Monad m => Sink i m r -> Sink i m r' -> Sink i m (r, r')-zipSinks (ConduitM x0) (ConduitM y0) =-    ConduitM $ injectLeftovers x0 >< injectLeftovers y0-  where-    (><) :: Monad m => Pipe Void i Void () m r1 -> Pipe Void i Void () m r2 -> Pipe l i o () m (r1, r2)--    Leftover _  i    >< _                = absurd i-    _                >< Leftover _  i    = absurd i-    HaveOutput _ _ o >< _                = absurd o-    _                >< HaveOutput _ _ o = absurd o--    PipeM mx         >< y                = PipeM (liftM (>< y) mx)-    x                >< PipeM my         = PipeM (liftM (x ><) my)-    Done x           >< Done y           = Done (x, y)-    NeedInput px cx  >< NeedInput py cy  = NeedInput (\i -> px i >< py i) (\() -> cx () >< cy ())-    NeedInput px cx  >< y@Done{}         = NeedInput (\i -> px i >< y)    (\u -> cx u >< y)-    x@Done{}         >< NeedInput py cy  = NeedInput (\i -> x >< py i)    (\u -> x >< cy u)---- | Combines two sources. The new source will stop producing once either---   source has been exhausted.------ Since 1.0.13-zipSources :: Monad m => Source m a -> Source m b -> Source m (a, b)-zipSources (ConduitM left0) (ConduitM right0) =-    ConduitM $ go left0 right0-  where-    go (Leftover left ()) right = go left right-    go left (Leftover right ())  = go left right-    go (Done ()) (Done ()) = Done ()-    go (Done ()) (HaveOutput _ close _) = PipeM (close >> return (Done ()))-    go (HaveOutput _ close _) (Done ()) = PipeM (close >> return (Done ()))-    go (Done ()) (PipeM _) = Done ()-    go (PipeM _) (Done ()) = Done ()-    go (PipeM mx) (PipeM my) = PipeM (liftM2 go mx my)-    go (PipeM mx) y@HaveOutput{} = PipeM (liftM (\x -> go x y) mx)-    go x@HaveOutput{} (PipeM my) = PipeM (liftM (go x) my)-    go (HaveOutput srcx closex x) (HaveOutput srcy closey y) = HaveOutput (go srcx srcy) (closex >> closey) (x, y)-    go (NeedInput _ c) right = go (c ()) right-    go left (NeedInput _ c) = go left (c ())---- | Combines two sources. The new source will stop producing once either---   source has been exhausted.------ Since 1.0.13-zipSourcesApp :: Monad m => Source m (a -> b) -> Source m a -> Source m b-zipSourcesApp (ConduitM left0) (ConduitM right0) =-    ConduitM $ go left0 right0-  where-    go (Leftover left ()) right = go left right-    go left (Leftover right ())  = go left right-    go (Done ()) (Done ()) = Done ()-    go (Done ()) (HaveOutput _ close _) = PipeM (close >> return (Done ()))-    go (HaveOutput _ close _) (Done ()) = PipeM (close >> return (Done ()))-    go (Done ()) (PipeM _) = Done ()-    go (PipeM _) (Done ()) = Done ()-    go (PipeM mx) (PipeM my) = PipeM (liftM2 go mx my)-    go (PipeM mx) y@HaveOutput{} = PipeM (liftM (\x -> go x y) mx)-    go x@HaveOutput{} (PipeM my) = PipeM (liftM (go x) my)-    go (HaveOutput srcx closex x) (HaveOutput srcy closey y) = HaveOutput (go srcx srcy) (closex >> closey) (x y)-    go (NeedInput _ c) right = go (c ()) right-    go left (NeedInput _ c) = go left (c ())---- |------ Since 1.0.17-zipConduitApp-    :: Monad m-    => ConduitM i o m (x -> y)-    -> ConduitM i o m x-    -> ConduitM i o m y-zipConduitApp (ConduitM left0) (ConduitM right0) =-    ConduitM $ go (return ()) (return ()) (injectLeftovers left0) (injectLeftovers right0)-  where-    go _ _ (Done f) (Done x) = Done (f x)-    go _ finalY (HaveOutput x finalX o) y = HaveOutput-        (go finalX finalY x y)-        (finalX >> finalY)-        o-    go finalX _ x (HaveOutput y finalY o) = HaveOutput-        (go finalX finalY x y)-        (finalX >> finalY)-        o-    go _ _ (Leftover _ i) _ = absurd i-    go _ _ _ (Leftover _ i) = absurd i-    go finalX finalY (PipeM mx) y = PipeM (flip (go finalX finalY) y `liftM` mx)-    go finalX finalY x (PipeM my) = PipeM (go finalX finalY x `liftM` my)-    go finalX finalY (NeedInput px cx) (NeedInput py cy) = NeedInput-        (\i -> go finalX finalY (px i) (py i))-        (\u -> go finalX finalY (cx u) (cy u))-    go finalX finalY (NeedInput px cx) (Done y) = NeedInput-        (\i -> go finalX finalY (px i) (Done y))-        (\u -> go finalX finalY (cx u) (Done y))-    go finalX finalY (Done x) (NeedInput py cy) = NeedInput-        (\i -> go finalX finalY (Done x) (py i))-        (\u -> go finalX finalY (Done x) (cy u))---- | Same as normal fusion (e.g. @=$=@), except instead of discarding leftovers--- from the downstream component, return them.------ Since 1.0.17-fuseReturnLeftovers :: Monad m-                    => ConduitM a b m ()-                    -> ConduitM b c m r-                    -> ConduitM a c m (r, [b])-fuseReturnLeftovers (ConduitM left0) (ConduitM right0) =-    ConduitM $ goRight (return ()) [] left0 right0-  where-    goRight final bs left right =-        case right of-            HaveOutput p c o -> HaveOutput (recurse p) (c >> final) o-            NeedInput rp rc  ->-                case bs of-                    [] -> goLeft rp rc final left-                    b:bs' -> goRight final bs' left (rp b)-            Done r2          -> PipeM (final >> return (Done (r2, bs)))-            PipeM mp         -> PipeM (liftM recurse mp)-            Leftover p b     -> goRight final (b:bs) left p-      where-        recurse = goRight final bs left--    goLeft rp rc final left =-        case left of-            HaveOutput left' final' o -> goRight final' [] left' (rp o)-            NeedInput left' lc        -> NeedInput (recurse . left') (recurse . lc)-            Done r1                   -> goRight (return ()) [] (Done r1) (rc r1)-            PipeM mp                  -> PipeM (liftM recurse mp)-            Leftover left' i          -> Leftover (recurse left') i-      where-        recurse = goLeft rp rc final---- | Similar to @fuseReturnLeftovers@, but use the provided function to convert--- downstream leftovers to upstream leftovers.------ Since 1.0.17-fuseLeftovers-    :: Monad m-    => ([b] -> [a])-    -> ConduitM a b m ()-    -> ConduitM b c m r-    -> ConduitM a c m r-fuseLeftovers f left right = do-    (r, bs) <- fuseReturnLeftovers left right-    ConduitM $ mapM_ leftover $ reverse $ f bs-    return r---- | A generalization of 'ResumableSource'. Allows to resume an arbitrary--- conduit, keeping its state and using it later (or finalizing it).------ Since 1.0.17-data ResumableConduit i m o =-    ResumableConduit (Conduit i m o) (m ())---- | Connect a 'ResumableConduit' to a sink and return the output of the sink--- together with a new 'ResumableConduit'.------ Since 1.0.17-connectResumeConduit-    :: Monad m-    => ResumableConduit i m o-    -> Sink o m r-    -> Sink i m (ResumableConduit i m o, r)-connectResumeConduit (ResumableConduit (ConduitM left0) leftFinal0) (ConduitM right0) =-    ConduitM $ goRight leftFinal0 left0 right0-  where-    goRight leftFinal left right =-        case right of-            HaveOutput _ _ o -> absurd o-            NeedInput rp rc -> goLeft rp rc leftFinal left-            Done r2 -> Done (ResumableConduit (ConduitM left) leftFinal, r2)-            PipeM mp -> PipeM (liftM (goRight leftFinal left) mp)-            Leftover p i -> goRight leftFinal (HaveOutput left leftFinal i) p--    goLeft rp rc leftFinal left =-        case left of-            HaveOutput left' leftFinal' o -> goRight leftFinal' left' (rp o)-            NeedInput left' lc -> NeedInput (recurse . left') (recurse . lc)-            Done () -> goRight (return ()) (Done ()) (rc ())-            PipeM mp -> PipeM (liftM recurse mp)-            Leftover left' i -> Leftover (recurse left') i -- recurse p-      where-        recurse = goLeft rp rc leftFinal---- | Unwraps a @ResumableConduit@ into a @Conduit@ and a finalizer.------ Since 'unwrapResumable' for more information.------ Since 1.0.17-unwrapResumableConduit :: MonadIO m => ResumableConduit i m o -> m (Conduit i m o, m ())-unwrapResumableConduit (ResumableConduit src final) = do-    ref <- liftIO $ I.newIORef True-    let final' = do-            x <- liftIO $ I.readIORef ref-            when x final-    return (liftIO (I.writeIORef ref False) >> src, final')---- | Turn a @Conduit@ into a @ResumableConduit@ with no attached finalizer.------ Since 1.1.4-newResumableConduit :: Monad m => Conduit i m o -> ResumableConduit i m o-newResumableConduit c = ResumableConduit c (return ())---- | Turn a @Sink@ into a @Conduit@ in the following way:------ * All input passed to the @Sink@ is yielded downstream.------ * When the @Sink@ finishes processing, the result is passed to the provided to the finalizer function.------ Note that the @Sink@ will stop receiving input as soon as the downstream it--- is connected to shuts down.------ An example usage would be to write the result of a @Sink@ to some mutable--- variable while allowing other processing to continue.------ Since 1.1.0-passthroughSink :: Monad m-                => Sink i m r-                -> (r -> m ()) -- ^ finalizer-                -> Conduit i m i-passthroughSink (ConduitM sink0) final =-    ConduitM $ go [] sink0-  where-    go _ (Done r) = do-        lift $ final r-        awaitForever yield-    go is (Leftover sink i) = go (i:is) sink-    go _ (HaveOutput _ _ o) = absurd o-    go is (PipeM mx) = do-        x <- lift mx-        go is x-    go (i:is) (NeedInput next _) = go is (next i)-    go [] (NeedInput next done) = do-        mx <- await-        case mx of-            Nothing -> go [] (done ())-            Just x -> do-                yield x-                go [] (next x)---- | Generalize the upstream return value for a @Pipe@ from unit to any type.------ Since 1.1.5-generalizeUpstream :: Monad m => Pipe l i o () m r -> Pipe l i o u m r-generalizeUpstream =-    go-  where-    go (HaveOutput p f o) = HaveOutput (go p) f o-    go (NeedInput x y) = NeedInput (go . x) (\_ -> go (y ()))-    go (Done r) = Done r-    go (PipeM mp) = PipeM (liftM go mp)-    go (Leftover p l) = Leftover (go p) l-{-# INLINE generalizeUpstream #-}+module Data.Conduit.Internal+    ( -- * Pipe+      module Data.Conduit.Internal.Pipe+      -- * Conduit+    , module Data.Conduit.Internal.Conduit+      -- * Fusion (highly experimental!!!)+    , module Data.Conduit.Internal.Fusion+    ) where++import           Data.Conduit.Internal.Conduit hiding (addCleanup, await,+                                                awaitForever, bracketP,+                                                leftover, mapInput, mapOutput,+                                                mapOutputMaybe, transPipe,+                                                yield, yieldM, yieldOr)+import           Data.Conduit.Internal.Pipe+import           Data.Conduit.Internal.Fusion

@@ -0,0 +1,1194 @@+{-# OPTIONS_HADDOCK not-home #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ImpredicativeTypes #-}+module Data.Conduit.Internal.Conduit+    ( -- ** Types+      ConduitM (..)+    , Source+    , Producer+    , Sink+    , Consumer+    , Conduit+    , ResumableSource (..)+    , ResumableConduit (..)+    , Flush (..)+      -- *** Newtype wrappers+    , ZipSource (..)+    , ZipSink (..)+    , ZipConduit (..)+      -- ** Primitives+    , await+    , awaitForever+    , yield+    , yieldM+    , yieldOr+    , leftover+      -- ** Composition+    , connectResume+    , connectResumeConduit+    , fuseLeftovers+    , fuseReturnLeftovers+    , ($$+)+    , ($$++)+    , ($$+-)+    , ($=+)+    , (=$$+)+    , (=$$++)+    , (=$$+-)+    , ($$)+    , ($=)+    , (=$)+    , (=$=)+      -- ** Generalizing+    , sourceToPipe+    , sinkToPipe+    , conduitToPipe+    , toProducer+    , toConsumer+      -- ** Cleanup+    , bracketP+    , addCleanup+      -- ** Exceptions+    , catchC+    , handleC+    , tryC+      -- ** Utilities+    , Data.Conduit.Internal.Conduit.transPipe+    , Data.Conduit.Internal.Conduit.mapOutput+    , Data.Conduit.Internal.Conduit.mapOutputMaybe+    , Data.Conduit.Internal.Conduit.mapInput+    , Data.Conduit.Internal.Conduit.closeResumableSource+    , unwrapResumable+    , unwrapResumableConduit+    , newResumableSource+    , newResumableConduit+    , zipSinks+    , zipSources+    , zipSourcesApp+    , zipConduitApp+    , passthroughSink+    , fuseBoth+    , fuseUpstream+    , sequenceSources+    , sequenceSinks+    , sequenceConduits+    ) where++import Control.Applicative (Applicative (..))+import Control.Exception.Lifted as E (Exception)+import qualified Control.Exception.Lifted as E (catch)+import Control.Monad (liftM, when, liftM2, ap)+import Control.Monad.Error.Class(MonadError(..))+import Control.Monad.Reader.Class(MonadReader(..))+import Control.Monad.RWS.Class(MonadRWS())+import Control.Monad.Writer.Class(MonadWriter(..))+import Control.Monad.State.Class(MonadState(..))+import Control.Monad.Trans.Class (MonadTrans (lift))+import Control.Monad.IO.Class (MonadIO (liftIO))+import Control.Monad.Base (MonadBase (liftBase))+import Data.Void (Void, absurd)+import Data.Monoid (Monoid (mappend, mempty))+import Control.Monad.Trans.Resource+import qualified Data.IORef as I+import Control.Monad.Morph (MFunctor (..))+import Data.Conduit.Internal.Pipe hiding (yield, mapOutput, leftover, yieldM, yieldOr, await, awaitForever, addCleanup, bracketP)+import qualified Data.Conduit.Internal.Pipe as CI+import Control.Monad (forever)+import Data.Traversable (Traversable (..))+import Control.Monad.Catch (MonadCatch, catch)++-- | Core datatype of the conduit package. This type represents a general+-- component which can consume a stream of input values @i@, produce a stream+-- of output values @o@, perform actions in the @m@ monad, and produce a final+-- result @r@. The type synonyms provided here are simply wrappers around this+-- type.+--+-- Since 1.0.0+newtype ConduitM i o m r = ConduitM+    { unConduitM :: forall b.+                    (r -> Pipe i i o () m b) -> Pipe i i o () m b+    }++instance Functor (ConduitM i o m) where+    fmap f (ConduitM c) = ConduitM $ \rest -> c (rest . f)++instance Applicative (ConduitM i o m) where+    pure = return+    {-# INLINE pure #-}+    (<*>) = ap+    {-# INLINE (<*>) #-}++instance Monad (ConduitM i o m) where+    return x = ConduitM ($ x)+    ConduitM f >>= g = ConduitM $ \h -> f $ \a -> unConduitM (g a) h++instance MonadThrow m => MonadThrow (ConduitM i o m) where+    throwM = lift . throwM++instance MFunctor (ConduitM i o) where+    hoist f (ConduitM c0) = ConduitM $ \rest -> let+        go (HaveOutput p c o) = HaveOutput (go p) (f c) o+        go (NeedInput p c) = NeedInput (go . p) (go . c)+        go (Done r) = rest r+        go (PipeM mp) =+            PipeM (f $ liftM go $ collapse mp)+          where+            -- Combine a series of monadic actions into a single action.  Since we+            -- throw away side effects between different actions, an arbitrary break+            -- between actions will lead to a violation of the monad transformer laws.+            -- Example available at:+            --+            -- http://hpaste.org/75520+            collapse mpipe = do+                pipe' <- mpipe+                case pipe' of+                    PipeM mpipe' -> collapse mpipe'+                    _ -> return pipe'+        go (Leftover p i) = Leftover (go p) i+        in go (c0 Done)++instance MonadCatch m => MonadCatch (ConduitM i o m) where+    catch (ConduitM p0) onErr = ConduitM $ \rest -> let+        go (Done r) = rest r+        go (PipeM mp) = PipeM $ catch (liftM go mp) (return . flip unConduitM rest . onErr)+        go (Leftover p i) = Leftover (go p) i+        go (NeedInput x y) = NeedInput (go . x) (go . y)+        go (HaveOutput p c o) = HaveOutput (go p) c o+        in go (p0 Done)++instance MonadIO m => MonadIO (ConduitM i o m) where+    liftIO = lift . liftIO+    {-# INLINE liftIO #-}++instance MonadReader r m => MonadReader r (ConduitM i o m) where+    ask = lift ask+    {-# INLINE ask #-}++    local f (ConduitM c0) = ConduitM $ \rest ->+        let go (HaveOutput p c o) = HaveOutput (go p) c o+            go (NeedInput p c) = NeedInput (\i -> go (p i)) (\u -> go (c u))+            go (Done x) = rest x+            go (PipeM mp) = PipeM (liftM go $ local f mp)+            go (Leftover p i) = Leftover (go p) i+         in go (c0 Done)++#ifndef MIN_VERSION_mtl+#define MIN_VERSION_mtl(x, y, z) 0+#endif++instance MonadWriter w m => MonadWriter w (ConduitM i o m) where+#if MIN_VERSION_mtl(2, 1, 0)+    writer = lift . writer+#endif+    tell = lift . tell++    listen (ConduitM c0) = ConduitM $ \rest ->+        let go front (HaveOutput p c o) = HaveOutput (go front p) c o+            go front (NeedInput p c) = NeedInput (\i -> go front (p i)) (\u -> go front (c u))+            go front (Done x) = rest (x, front)+            go front (PipeM mp) = PipeM $ do+                (p,w) <- listen mp+                return $ go (front `mappend` w) p+            go front (Leftover p i) = Leftover (go front p) i+         in go mempty (c0 Done)++    pass (ConduitM c0) = ConduitM $ \rest ->+        let go (HaveOutput p c o) = HaveOutput (go p) c o+            go (NeedInput p c) = NeedInput (\i -> go (p i)) (\u -> go (c u))+            go (PipeM mp) = PipeM $ mp >>= (return . go)+            go (Done (x,_)) = rest x+            go (Leftover p i) = Leftover (go p) i+         in go (c0 Done)++instance MonadState s m => MonadState s (ConduitM i o m) where+    get = lift get+    put = lift . put+#if MIN_VERSION_mtl(2, 1, 0)+    state = lift . state+#endif++instance MonadRWS r w s m => MonadRWS r w s (ConduitM i o m)++instance MonadError e m => MonadError e (ConduitM i o m) where+    throwError = lift . throwError+    catchError (ConduitM c0) f = ConduitM $ \rest ->+        let go (HaveOutput p c o) = HaveOutput (go p) c o+            go (NeedInput p c) = NeedInput (\i -> go (p i)) (\u -> go (c u))+            go (Done x) = rest x+            go (PipeM mp) =+              PipeM $ catchError (liftM go mp) $ \e -> do+                return $ unConduitM (f e) rest+            go (Leftover p i) = Leftover (go p) i+         in go (c0 Done)++instance MonadBase base m => MonadBase base (ConduitM i o m) where+    liftBase = lift . liftBase+    {-# INLINE liftBase #-}++instance MonadTrans (ConduitM i o) where+    lift mr = ConduitM $ \rest -> PipeM (liftM rest mr)+    {-# INLINE [1] lift #-}++instance MonadResource m => MonadResource (ConduitM i o m) where+    liftResourceT = lift . liftResourceT+    {-# INLINE liftResourceT #-}++instance Monad m => Monoid (ConduitM i o m ()) where+    mempty = return ()+    {-# INLINE mempty #-}+    mappend = (>>)+    {-# INLINE mappend #-}++-- | Provides a stream of output values, without consuming any input or+-- producing a final result.+--+-- Since 0.5.0+type Source m o = ConduitM () o m ()++-- | A component which produces a stream of output values, regardless of the+-- input stream. A @Producer@ is a generalization of a @Source@, and can be+-- used as either a @Source@ or a @Conduit@.+--+-- Since 1.0.0+type Producer m o = forall i. ConduitM i o m ()++-- | Consumes a stream of input values and produces a final result, without+-- producing any output.+--+-- > type Sink i m r = ConduitM i Void m r+--+-- Since 0.5.0+type Sink i = ConduitM i Void++-- | A component which consumes a stream of input values and produces a final+-- result, regardless of the output stream. A @Consumer@ is a generalization of+-- a @Sink@, and can be used as either a @Sink@ or a @Conduit@.+--+-- Since 1.0.0+type Consumer i m r = forall o. ConduitM i o m r++-- | Consumes a stream of input values and produces a stream of output values,+-- without producing a final result.+--+-- Since 0.5.0+type Conduit i m o = ConduitM i o m ()++-- | A @Source@ which has been started, but has not yet completed.+--+-- This type contains both the current state of the @Source@, and the finalizer+-- to be run to close it.+--+-- Since 0.5.0+data ResumableSource m o = ResumableSource (Pipe () () o () m ()) (m ())++-- | Since 1.0.13+instance MFunctor ResumableSource where+    hoist nat (ResumableSource src m) = ResumableSource (hoist nat src) (nat m)++-- | Connect a @Source@ to a @Sink@ until the latter closes. Returns both the+-- most recent state of the @Source@ and the result of the @Sink@.+--+-- We use a @ResumableSource@ to keep track of the most recent finalizer+-- provided by the @Source@.+--+-- Since 0.5.0+connectResume :: Monad m+              => ResumableSource m o+              -> Sink o m r+              -> m (ResumableSource m o, r)+connectResume (ResumableSource left0 leftFinal0) (ConduitM right0) =+    goRight leftFinal0 left0 (right0 Done)+  where+    goRight leftFinal left right =+        case right of+            HaveOutput _ _ o -> absurd o+            NeedInput rp rc  -> goLeft rp rc leftFinal left+            Done r2          -> return (ResumableSource left leftFinal, r2)+            PipeM mp         -> mp >>= goRight leftFinal left+            Leftover p i     -> goRight leftFinal (HaveOutput left leftFinal i) p++    goLeft rp rc leftFinal left =+        case left of+            HaveOutput left' leftFinal' o -> goRight leftFinal' left' (rp o)+            NeedInput _ lc                -> recurse (lc ())+            Done ()                       -> goRight (return ()) (Done ()) (rc ())+            PipeM mp                      -> mp >>= recurse+            Leftover p ()                 -> recurse p+      where+        recurse = goLeft rp rc leftFinal++sourceToPipe :: Monad m => Source m o -> Pipe l i o u m ()+sourceToPipe =+    go . flip unConduitM Done+  where+    go (HaveOutput p c o) = HaveOutput (go p) c o+    go (NeedInput _ c) = go $ c ()+    go (Done ()) = Done ()+    go (PipeM mp) = PipeM (liftM go mp)+    go (Leftover p ()) = go p++sinkToPipe :: Monad m => Sink i m r -> Pipe l i o u m r+sinkToPipe =+    go . injectLeftovers . flip unConduitM Done+  where+    go (HaveOutput _ _ o) = absurd o+    go (NeedInput p c) = NeedInput (go . p) (const $ go $ c ())+    go (Done r) = Done r+    go (PipeM mp) = PipeM (liftM go mp)+    go (Leftover _ l) = absurd l++conduitToPipe :: Monad m => Conduit i m o -> Pipe l i o u m ()+conduitToPipe =+    go . injectLeftovers . flip unConduitM Done+  where+    go (HaveOutput p c o) = HaveOutput (go p) c o+    go (NeedInput p c) = NeedInput (go . p) (const $ go $ c ())+    go (Done ()) = Done ()+    go (PipeM mp) = PipeM (liftM go mp)+    go (Leftover _ l) = absurd l++-- | Unwraps a @ResumableSource@ into a @Source@ and a finalizer.+--+-- A @ResumableSource@ represents a @Source@ which has already been run, and+-- therefore has a finalizer registered. As a result, if we want to turn it+-- into a regular @Source@, we need to ensure that the finalizer will be run+-- appropriately. By appropriately, I mean:+--+-- * If a new finalizer is registered, the old one should not be called.+--+-- * If the old one is called, it should not be called again.+--+-- This function returns both a @Source@ and a finalizer which ensures that the+-- above two conditions hold. Once you call that finalizer, the @Source@ is+-- invalidated and cannot be used.+--+-- Since 0.5.2+unwrapResumable :: MonadIO m => ResumableSource m o -> m (Source m o, m ())+unwrapResumable (ResumableSource src final) = do+    ref <- liftIO $ I.newIORef True+    let final' = do+            x <- liftIO $ I.readIORef ref+            when x final+    return (liftIO (I.writeIORef ref False) >> (ConduitM (src >>=)), final')++-- | Turn a @Source@ into a @ResumableSource@ with no attached finalizer.+--+-- Since 1.1.4+newResumableSource :: Monad m => Source m o -> ResumableSource m o+newResumableSource (ConduitM s) = ResumableSource (s Done) (return ())++-- | Generalize a 'Source' to a 'Producer'.+--+-- Since 1.0.0+toProducer :: Monad m => Source m a -> Producer m a+toProducer (ConduitM c0) = ConduitM $ \rest -> let+    go (HaveOutput p c o) = HaveOutput (go p) c o+    go (NeedInput _ c) = go (c ())+    go (Done r) = rest r+    go (PipeM mp) = PipeM (liftM go mp)+    go (Leftover p ()) = go p+    in go (c0 Done)++-- | Generalize a 'Sink' to a 'Consumer'.+--+-- Since 1.0.0+toConsumer :: Monad m => Sink a m b -> Consumer a m b+toConsumer (ConduitM c0) = ConduitM $ \rest -> let+    go (HaveOutput _ _ o) = absurd o+    go (NeedInput p c) = NeedInput (go . p) (go . c)+    go (Done r) = rest r+    go (PipeM mp) = PipeM (liftM go mp)+    go (Leftover p l) = Leftover (go p) l+    in go (c0 Done)++-- | Catch all exceptions thrown by the current component of the pipeline.+--+-- Note: this will /not/ catch exceptions thrown by other components! For+-- example, if an exception is thrown in a @Source@ feeding to a @Sink@, and+-- the @Sink@ uses @catchC@, the exception will /not/ be caught.+--+-- Due to this behavior (as well as lack of async exception handling), you+-- should not try to implement combinators such as @onException@ in terms of this+-- primitive function.+--+-- Note also that the exception handling will /not/ be applied to any+-- finalizers generated by this conduit.+--+-- Since 1.0.11+catchC :: (MonadBaseControl IO m, Exception e)+       => ConduitM i o m r+       -> (e -> ConduitM i o m r)+       -> ConduitM i o m r+catchC (ConduitM p0) onErr = ConduitM $ \rest -> let+    go (Done r) = rest r+    go (PipeM mp) = PipeM $ E.catch (liftM go mp)+        (return . flip unConduitM rest . onErr)+    go (Leftover p i) = Leftover (go p) i+    go (NeedInput x y) = NeedInput (go . x) (go . y)+    go (HaveOutput p c o) = HaveOutput (go p) c o+    in go (p0 Done)+{-# INLINE catchC #-}++-- | The same as @flip catchC@.+--+-- Since 1.0.11+handleC :: (MonadBaseControl IO m, Exception e)+        => (e -> ConduitM i o m r)+        -> ConduitM i o m r+        -> ConduitM i o m r+handleC = flip catchC+{-# INLINE handleC #-}++-- | A version of @try@ for use within a pipeline. See the comments in @catchC@+-- for more details.+--+-- Since 1.0.11+tryC :: (MonadBaseControl IO m, Exception e)+     => ConduitM i o m r+     -> ConduitM i o m (Either e r)+tryC (ConduitM c0) = ConduitM $ \rest -> let+    go (Done r) = rest (Right r)+    go (PipeM mp) = PipeM $ E.catch (liftM go mp) (return . rest . Left)+    go (Leftover p i) = Leftover (go p) i+    go (NeedInput x y) = NeedInput (go . x) (go . y)+    go (HaveOutput p c o) = HaveOutput (go p) c o+    in go (c0 Done)+{-# INLINE tryC #-}++-- | Combines two sinks. The new sink will complete when both input sinks have+--   completed.+--+-- Any leftovers are discarded.+--+-- Since 0.4.1+zipSinks :: Monad m => Sink i m r -> Sink i m r' -> Sink i m (r, r')+zipSinks (ConduitM x0) (ConduitM y0) = ConduitM $ \rest -> let+    Leftover _  i    >< _                = absurd i+    _                >< Leftover _  i    = absurd i+    HaveOutput _ _ o >< _                = absurd o+    _                >< HaveOutput _ _ o = absurd o++    PipeM mx         >< y                = PipeM (liftM (>< y) mx)+    x                >< PipeM my         = PipeM (liftM (x ><) my)+    Done x           >< Done y           = rest (x, y)+    NeedInput px cx  >< NeedInput py cy  = NeedInput (\i -> px i >< py i) (\() -> cx () >< cy ())+    NeedInput px cx  >< y@Done{}         = NeedInput (\i -> px i >< y)    (\u -> cx u >< y)+    x@Done{}         >< NeedInput py cy  = NeedInput (\i -> x >< py i)    (\u -> x >< cy u)+    in injectLeftovers (x0 Done) >< injectLeftovers (y0 Done)++-- | Combines two sources. The new source will stop producing once either+--   source has been exhausted.+--+-- Since 1.0.13+zipSources :: Monad m => Source m a -> Source m b -> Source m (a, b)+zipSources (ConduitM left0) (ConduitM right0) = ConduitM $ \rest -> let+    go (Leftover left ()) right = go left right+    go left (Leftover right ())  = go left right+    go (Done ()) (Done ()) = rest ()+    go (Done ()) (HaveOutput _ close _) = PipeM (close >> return (rest ()))+    go (HaveOutput _ close _) (Done ()) = PipeM (close >> return (rest ()))+    go (Done ()) (PipeM _) = rest ()+    go (PipeM _) (Done ()) = rest ()+    go (PipeM mx) (PipeM my) = PipeM (liftM2 go mx my)+    go (PipeM mx) y@HaveOutput{} = PipeM (liftM (\x -> go x y) mx)+    go x@HaveOutput{} (PipeM my) = PipeM (liftM (go x) my)+    go (HaveOutput srcx closex x) (HaveOutput srcy closey y) = HaveOutput (go srcx srcy) (closex >> closey) (x, y)+    go (NeedInput _ c) right = go (c ()) right+    go left (NeedInput _ c) = go left (c ())+    in go (left0 Done) (right0 Done)++-- | Combines two sources. The new source will stop producing once either+--   source has been exhausted.+--+-- Since 1.0.13+zipSourcesApp :: Monad m => Source m (a -> b) -> Source m a -> Source m b+zipSourcesApp (ConduitM left0) (ConduitM right0) = ConduitM $ \rest -> let+    go (Leftover left ()) right = go left right+    go left (Leftover right ())  = go left right+    go (Done ()) (Done ()) = rest ()+    go (Done ()) (HaveOutput _ close _) = PipeM (close >> return (rest ()))+    go (HaveOutput _ close _) (Done ()) = PipeM (close >> return (rest ()))+    go (Done ()) (PipeM _) = rest ()+    go (PipeM _) (Done ()) = rest ()+    go (PipeM mx) (PipeM my) = PipeM (liftM2 go mx my)+    go (PipeM mx) y@HaveOutput{} = PipeM (liftM (\x -> go x y) mx)+    go x@HaveOutput{} (PipeM my) = PipeM (liftM (go x) my)+    go (HaveOutput srcx closex x) (HaveOutput srcy closey y) = HaveOutput (go srcx srcy) (closex >> closey) (x y)+    go (NeedInput _ c) right = go (c ()) right+    go left (NeedInput _ c) = go left (c ())+    in go (left0 Done) (right0 Done)++-- |+--+-- Since 1.0.17+zipConduitApp+    :: Monad m+    => ConduitM i o m (x -> y)+    -> ConduitM i o m x+    -> ConduitM i o m y+zipConduitApp (ConduitM left0) (ConduitM right0) = ConduitM $ \rest -> let+    go _ _ (Done f) (Done x) = rest (f x)+    go _ finalY (HaveOutput x finalX o) y = HaveOutput+        (go finalX finalY x y)+        (finalX >> finalY)+        o+    go finalX _ x (HaveOutput y finalY o) = HaveOutput+        (go finalX finalY x y)+        (finalX >> finalY)+        o+    go _ _ (Leftover _ i) _ = absurd i+    go _ _ _ (Leftover _ i) = absurd i+    go finalX finalY (PipeM mx) y = PipeM (flip (go finalX finalY) y `liftM` mx)+    go finalX finalY x (PipeM my) = PipeM (go finalX finalY x `liftM` my)+    go finalX finalY (NeedInput px cx) (NeedInput py cy) = NeedInput+        (\i -> go finalX finalY (px i) (py i))+        (\u -> go finalX finalY (cx u) (cy u))+    go finalX finalY (NeedInput px cx) (Done y) = NeedInput+        (\i -> go finalX finalY (px i) (Done y))+        (\u -> go finalX finalY (cx u) (Done y))+    go finalX finalY (Done x) (NeedInput py cy) = NeedInput+        (\i -> go finalX finalY (Done x) (py i))+        (\u -> go finalX finalY (Done x) (cy u))+  in go (return ()) (return ()) (injectLeftovers $ left0 Done) (injectLeftovers $ right0 Done)++-- | Same as normal fusion (e.g. @=$=@), except instead of discarding leftovers+-- from the downstream component, return them.+--+-- Since 1.0.17+fuseReturnLeftovers :: Monad m+                    => ConduitM a b m ()+                    -> ConduitM b c m r+                    -> ConduitM a c m (r, [b])+fuseReturnLeftovers (ConduitM left0) (ConduitM right0) = ConduitM $ \rest -> let+    goRight final bs left right =+        case right of+            HaveOutput p c o -> HaveOutput (recurse p) (c >> final) o+            NeedInput rp rc  ->+                case bs of+                    [] -> goLeft rp rc final left+                    b:bs' -> goRight final bs' left (rp b)+            Done r2          -> PipeM (final >> return (rest (r2, bs)))+            PipeM mp         -> PipeM (liftM recurse mp)+            Leftover p b     -> goRight final (b:bs) left p+      where+        recurse = goRight final bs left++    goLeft rp rc final left =+        case left of+            HaveOutput left' final' o -> goRight final' [] left' (rp o)+            NeedInput left' lc        -> NeedInput (recurse . left') (recurse . lc)+            Done r1                   -> goRight (return ()) [] (Done r1) (rc r1)+            PipeM mp                  -> PipeM (liftM recurse mp)+            Leftover left' i          -> Leftover (recurse left') i+      where+        recurse = goLeft rp rc final+    in goRight (return ()) [] (left0 Done) (right0 Done)++-- | Similar to @fuseReturnLeftovers@, but use the provided function to convert+-- downstream leftovers to upstream leftovers.+--+-- Since 1.0.17+fuseLeftovers+    :: Monad m+    => ([b] -> [a])+    -> ConduitM a b m ()+    -> ConduitM b c m r+    -> ConduitM a c m r+fuseLeftovers f left right = do+    (r, bs) <- fuseReturnLeftovers left right+    mapM_ leftover $ reverse $ f bs+    return r++-- | A generalization of 'ResumableSource'. Allows to resume an arbitrary+-- conduit, keeping its state and using it later (or finalizing it).+--+-- Since 1.0.17+data ResumableConduit i m o =+    ResumableConduit (Pipe i i o () m ()) (m ())++-- | Connect a 'ResumableConduit' to a sink and return the output of the sink+-- together with a new 'ResumableConduit'.+--+-- Since 1.0.17+connectResumeConduit+    :: Monad m+    => ResumableConduit i m o+    -> Sink o m r+    -> Sink i m (ResumableConduit i m o, r)+connectResumeConduit (ResumableConduit left0 leftFinal0) (ConduitM right0) = ConduitM $ \rest -> let+    goRight leftFinal left right =+        case right of+            HaveOutput _ _ o -> absurd o+            NeedInput rp rc -> goLeft rp rc leftFinal left+            Done r2 -> rest (ResumableConduit left leftFinal, r2)+            PipeM mp -> PipeM (liftM (goRight leftFinal left) mp)+            Leftover p i -> goRight leftFinal (HaveOutput left leftFinal i) p++    goLeft rp rc leftFinal left =+        case left of+            HaveOutput left' leftFinal' o -> goRight leftFinal' left' (rp o)+            NeedInput left' lc -> NeedInput (recurse . left') (recurse . lc)+            Done () -> goRight (return ()) (Done ()) (rc ())+            PipeM mp -> PipeM (liftM recurse mp)+            Leftover left' i -> Leftover (recurse left') i -- recurse p+      where+        recurse = goLeft rp rc leftFinal+    in goRight leftFinal0 left0 (right0 Done)++-- | Unwraps a @ResumableConduit@ into a @Conduit@ and a finalizer.+--+-- Since 'unwrapResumable' for more information.+--+-- Since 1.0.17+unwrapResumableConduit :: MonadIO m => ResumableConduit i m o -> m (Conduit i m o, m ())+unwrapResumableConduit (ResumableConduit src final) = do+    ref <- liftIO $ I.newIORef True+    let final' = do+            x <- liftIO $ I.readIORef ref+            when x final+    return (ConduitM ((liftIO (I.writeIORef ref False) >> src) >>=), final')++-- | Turn a @Conduit@ into a @ResumableConduit@ with no attached finalizer.+--+-- Since 1.1.4+newResumableConduit :: Monad m => Conduit i m o -> ResumableConduit i m o+newResumableConduit (ConduitM c) = ResumableConduit (c Done) (return ())++-- | Turn a @Sink@ into a @Conduit@ in the following way:+--+-- * All input passed to the @Sink@ is yielded downstream.+--+-- * When the @Sink@ finishes processing, the result is passed to the provided to the finalizer function.+--+-- Note that the @Sink@ will stop receiving input as soon as the downstream it+-- is connected to shuts down.+--+-- An example usage would be to write the result of a @Sink@ to some mutable+-- variable while allowing other processing to continue.+--+-- Since 1.1.0+passthroughSink :: Monad m+                => Sink i m r+                -> (r -> m ()) -- ^ finalizer+                -> Conduit i m i+passthroughSink (ConduitM sink0) final = ConduitM $ \rest -> let+    go _ (Done r) = do+        lift $ final r+        unConduitM (awaitForever yield) rest+    go is (Leftover sink i) = go (i:is) sink+    go _ (HaveOutput _ _ o) = absurd o+    go is (PipeM mx) = do+        x <- lift mx+        go is x+    go (i:is) (NeedInput next _) = go is (next i)+    go [] (NeedInput next done) = do+        mx <- CI.await+        case mx of+            Nothing -> go [] (done ())+            Just x -> do+                CI.yield x+                go [] (next x)+    in go [] (sink0 Done)++-- Define fixity of all our operators+infixr 0 $$+infixl 1 $=+infixr 2 =$+infixr 2 =$=+infixr 0 $$++infixr 0 $$+++infixr 0 $$+-+infixl 1 $=+++-- | The connect operator, which pulls data from a source and pushes to a sink.+-- If you would like to keep the @Source@ open to be used for other+-- operations, use the connect-and-resume operator '$$+'.+--+-- Since 0.4.0+($$) :: Monad m => Source m a -> Sink a m b -> m b+src $$ sink = do+    (rsrc, res) <- src $$+ sink+    rsrc $$+- return ()+    return res+{-# INLINE [1] ($$) #-}++-- | Left fuse, combining a source and a conduit together into a new source.+--+-- Both the @Source@ and @Conduit@ will be closed when the newly-created+-- @Source@ is closed.+--+-- Leftover data from the @Conduit@ will be discarded.+--+-- Note: Since version 1.0.18, this operator has been generalized to be+-- identical to @=$=@.+--+-- Since 0.4.0+($=) :: Monad m => Conduit a m b -> ConduitM b c m r -> ConduitM a c m r+($=) = (=$=)+{-# INLINE [0] ($=) #-}+{-# RULES "$= is =$=" ($=) = (=$=) #-}++-- | Right fuse, combining a conduit and a sink together into a new sink.+--+-- Both the @Conduit@ and @Sink@ will be closed when the newly-created @Sink@+-- is closed.+--+-- Leftover data returned from the @Sink@ will be discarded.+--+-- Note: Since version 1.0.18, this operator has been generalized to be+-- identical to @=$=@.+--+-- Since 0.4.0+(=$) :: Monad m => Conduit a m b -> ConduitM b c m r -> ConduitM a c m r+(=$) = (=$=)+{-# INLINE [0] (=$) #-}+{-# RULES "=$ is =$=" (=$) = (=$=) #-}++-- | Fusion operator, combining two @Conduit@s together into a new @Conduit@.+--+-- Both @Conduit@s will be closed when the newly-created @Conduit@ is closed.+--+-- Leftover data returned from the right @Conduit@ will be discarded.+--+-- Since 0.4.0+(=$=) :: Monad m => Conduit a m b -> ConduitM b c m r -> ConduitM a c m r+ConduitM left0 =$= ConduitM right0 = ConduitM $ \rest ->+    let goRight final left right =+            case right of+                HaveOutput p c o  -> HaveOutput (recurse p) (c >> final) o+                NeedInput rp rc   -> goLeft rp rc final left+                Done r2           -> PipeM (final >> return (rest r2))+                PipeM mp          -> PipeM (liftM recurse mp)+                Leftover right' i -> goRight final (HaveOutput left final i) right'+          where+            recurse = goRight final left++        goLeft rp rc final left =+            case left of+                HaveOutput left' final' o -> goRight final' left' (rp o)+                NeedInput left' lc        -> NeedInput (recurse . left') (recurse . lc)+                Done r1                   -> goRight (return ()) (Done r1) (rc r1)+                PipeM mp                  -> PipeM (liftM recurse mp)+                Leftover left' i          -> Leftover (recurse left') i+          where+            recurse = goLeft rp rc final+     in goRight (return ()) (left0 Done) (right0 Done)+  where+{-# INLINE [1] (=$=) #-}++-- | Wait for a single input value from upstream. If no data is available,+-- returns @Nothing@.+--+-- Since 0.5.0+await :: Monad m => Consumer i m (Maybe i)+await = ConduitM $ \f -> NeedInput (f . Just) (const $ f Nothing)+{-# INLINE [0] await #-}++await' :: Monad m+       => ConduitM i o m r+       -> (i -> ConduitM i o m r)+       -> ConduitM i o m r+await' f g = ConduitM $ \rest -> NeedInput+    (\i -> unConduitM (g i) rest)+    (const $ unConduitM f rest)+{-# INLINE await' #-}+{-# RULES "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+-- would like to register a cleanup function, please use 'yieldOr' instead.+--+-- Since 0.5.0+yield :: Monad m+      => o -- ^ output value+      -> ConduitM i o m ()+yield o = yieldOr o (return ())+{-# INLINE yield #-}++yieldM :: Monad m => m o -> ConduitM i o m ()+yieldM mo = lift mo >>= yield+{-# INLINE yieldM #-}++  -- FIXME rule won't fire, see FIXME in .Pipe; "mapM_ yield" mapM_ yield = ConduitM . sourceList++-- | Provide a single piece of leftover input to be consumed by the next+-- component in the current monadic binding.+--+-- /Note/: it is highly encouraged to only return leftover values from input+-- already consumed from upstream.+--+-- Since 0.5.0+leftover :: i -> ConduitM i o m ()+leftover i = ConduitM $ \rest -> Leftover (rest ()) i+{-# INLINE leftover #-}++-- | Perform some allocation and run an inner component. Two guarantees are+-- given about resource finalization:+--+-- 1. It will be /prompt/. The finalization will be run as early as possible.+--+-- 2. It is exception safe. Due to usage of @resourcet@, the finalization will+-- be run in the event of any exceptions.+--+-- Since 0.5.0+bracketP :: MonadResource m+         => IO a+         -> (a -> IO ())+         -> (a -> ConduitM i o m r)+         -> ConduitM i o m r+bracketP alloc free inside = ConduitM $ \rest -> PipeM $ do+    (key, seed) <- allocate alloc free+    return $ unConduitM (addCleanup (const $ release key) (inside seed)) rest++-- | Add some code to be run when the given component cleans up.+--+-- The supplied cleanup function will be given a @True@ if the component ran to+-- completion, or @False@ if it terminated early due to a downstream component+-- terminating.+--+-- Note that this function is not exception safe. For that, please use+-- 'bracketP'.+--+-- Since 0.4.1+addCleanup :: Monad m+           => (Bool -> m ())+           -> ConduitM i o m r+           -> ConduitM i o m r+addCleanup cleanup (ConduitM c0) = ConduitM $ \rest -> let+    go (Done r) = PipeM (cleanup True >> return (rest r))+    go (HaveOutput src close x) = HaveOutput+        (go src)+        (cleanup False >> close)+        x+    go (PipeM msrc) = PipeM (liftM (go) msrc)+    go (NeedInput p c) = NeedInput+        (go . p)+        (go . c)+    go (Leftover p i) = Leftover (go p) i+    in go (c0 Done)++-- | Similar to 'yield', but additionally takes a finalizer to be run if the+-- downstream component terminates.+--+-- Since 0.5.0+yieldOr :: Monad m+        => o+        -> m () -- ^ finalizer+        -> ConduitM i o m ()+yieldOr o m = ConduitM $ \rest -> HaveOutput (rest ()) m o+{-# INLINE yieldOr #-}++-- | Wait for input forever, calling the given inner component for each piece of+-- new input. Returns the upstream result type.+--+-- This function is provided as a convenience for the common pattern of+-- @await@ing input, checking if it's @Just@ and then looping.+--+-- Since 0.5.0+awaitForever :: Monad m => (i -> ConduitM i o m r) -> ConduitM i o m ()+awaitForever f = ConduitM $ \rest ->+    let go = NeedInput (\i -> unConduitM (f i) (const go)) rest+     in go++-- | Transform the monad that a @ConduitM@ lives in.+--+-- Note that the monad transforming function will be run multiple times,+-- resulting in unintuitive behavior in some cases. For a fuller treatment,+-- please see:+--+-- <https://github.com/snoyberg/conduit/wiki/Dealing-with-monad-transformers>+--+-- This function is just a synonym for 'hoist'.+--+-- Since 0.4.0+transPipe :: Monad m => (forall a. m a -> n a) -> ConduitM i o m r -> ConduitM i o n r+transPipe = hoist++-- | Apply a function to all the output values of a @ConduitM@.+--+-- This mimics the behavior of `fmap` for a `Source` and `Conduit` in pre-0.4+-- days. It can also be simulated by fusing with the @map@ conduit from+-- "Data.Conduit.List".+--+-- Since 0.4.1+mapOutput :: Monad m => (o1 -> o2) -> ConduitM i o1 m r -> ConduitM i o2 m r+mapOutput f (ConduitM c0) = ConduitM $ \rest -> let+    go (HaveOutput p c o) = HaveOutput (go p) c (f o)+    go (NeedInput p c) = NeedInput (go . p) (go . c)+    go (Done r) = rest r+    go (PipeM mp) = PipeM (liftM (go) mp)+    go (Leftover p i) = Leftover (go p) i+    in go (c0 Done)++-- | Same as 'mapOutput', but use a function that returns @Maybe@ values.+--+-- Since 0.5.0+mapOutputMaybe :: Monad m => (o1 -> Maybe o2) -> ConduitM i o1 m r -> ConduitM i o2 m r+mapOutputMaybe f (ConduitM c0) = ConduitM $ \rest -> let+    go (HaveOutput p c o) = maybe id (\o' p' -> HaveOutput p' c o') (f o) (go p)+    go (NeedInput p c) = NeedInput (go . p) (go . c)+    go (Done r) = rest r+    go (PipeM mp) = PipeM (liftM (go) mp)+    go (Leftover p i) = Leftover (go p) i+    in go (c0 Done)++-- | Apply a function to all the input values of a @ConduitM@.+--+-- Since 0.5.0+mapInput :: Monad m+         => (i1 -> i2) -- ^ map initial input to new input+         -> (i2 -> Maybe i1) -- ^ map new leftovers to initial leftovers+         -> ConduitM i2 o m r+         -> ConduitM i1 o m r+mapInput f f' (ConduitM c0) = ConduitM $ \rest -> let+    go (HaveOutput p c o) = HaveOutput (go p) c o+    go (NeedInput p c) = NeedInput (go . p . f) (go . c)+    go (Done r) = rest r+    go (PipeM mp) = PipeM $ liftM go mp+    go (Leftover p i) = maybe id (flip Leftover) (f' i) (go p)+    in go (c0 Done)++-- | The connect-and-resume operator. This does not close the @Source@, but+-- instead returns it to be used again. This allows a @Source@ to be used+-- incrementally in a large program, without forcing the entire program to live+-- in the @Sink@ monad.+--+-- Mnemonic: connect + do more.+--+-- Since 0.5.0+($$+) :: Monad m => Source m a -> Sink a m b -> m (ResumableSource m a, b)+ConduitM src $$+ sink =+    connectResume (ResumableSource (src Done) (return ())) sink+{-# INLINE ($$+) #-}++-- | Continue processing after usage of @$$+@.+--+-- Since 0.5.0+($$++) :: Monad m => ResumableSource m a -> Sink a m b -> m (ResumableSource m a, b)+($$++) = connectResume+{-# INLINE ($$++) #-}++-- | Complete processing of a @ResumableSource@. This will run the finalizer+-- associated with the @ResumableSource@. In order to guarantee process resource+-- finalization, you /must/ use this operator after using @$$+@ and @$$++@.+--+-- Since 0.5.0+($$+-) :: Monad m => ResumableSource m a -> Sink a m b -> m b+rsrc $$+- sink = do+    (ResumableSource _ final, res) <- connectResume rsrc sink+    final+    return res+{-# INLINE ($$+-) #-}++-- | Left fusion for a resumable source.+--+-- Since 1.0.16+($=+) :: Monad m => ResumableSource m a -> Conduit a m b -> ResumableSource m b+ResumableSource src final $=+ ConduitM sink =+    ResumableSource (src `pipeL` sink Done) final++-- | Execute the finalizer associated with a @ResumableSource@, rendering the+-- @ResumableSource@ invalid for further use.+--+-- This is just a more explicit version of @$$+- return ()@.+--+-- Since 1.1.3+closeResumableSource :: Monad m => ResumableSource m a -> m ()+closeResumableSource = ($$+- return ())++-- | Provide for a stream of data that can be flushed.+--+-- A number of @Conduit@s (e.g., zlib compression) need the ability to flush+-- the stream at some point. This provides a single wrapper datatype to be used+-- in all such circumstances.+--+-- Since 0.3.0+data Flush a = Chunk a | Flush+    deriving (Show, Eq, Ord)+instance Functor Flush where+    fmap _ Flush = Flush+    fmap f (Chunk a) = Chunk (f a)++-- | A wrapper for defining an 'Applicative' instance for 'Source's which allows+-- to combine sources together, generalizing 'zipSources'. A combined source+-- will take input yielded from each of its @Source@s until any of them stop+-- producing output.+--+-- Since 1.0.13+newtype ZipSource m o = ZipSource { getZipSource :: Source m o }++instance Monad m => Functor (ZipSource m) where+    fmap f = ZipSource . mapOutput f . getZipSource+instance Monad m => Applicative (ZipSource m) where+    pure  = ZipSource . forever . yield+    (ZipSource f) <*> (ZipSource x) = ZipSource $ zipSourcesApp f x++-- | Coalesce all values yielded by all of the @Source@s.+--+-- Implemented on top of @ZipSource@, see that data type for more details.+--+-- Since 1.0.13+sequenceSources :: (Traversable f, Monad m) => f (Source m o) -> Source m (f o)+sequenceSources = getZipSource . sequenceA . fmap ZipSource++-- | A wrapper for defining an 'Applicative' instance for 'Sink's which allows+-- to combine sinks together, generalizing 'zipSinks'. A combined sink+-- distributes the input to all its participants and when all finish, produces+-- the result. This allows to define functions like+--+-- @+-- sequenceSinks :: (Monad m)+--           => [Sink i m r] -> Sink i m [r]+-- sequenceSinks = getZipSink . sequenceA . fmap ZipSink+-- @+--+-- Note that the standard 'Applicative' instance for conduits works+-- differently. It feeds one sink with input until it finishes, then switches+-- to another, etc., and at the end combines their results.+--+-- Since 1.0.13+newtype ZipSink i m r = ZipSink { getZipSink :: Sink i m r }++instance Monad m => Functor (ZipSink i m) where+    fmap f (ZipSink x) = ZipSink (liftM f x)+instance Monad m => Applicative (ZipSink i m) where+    pure  = ZipSink . return+    (ZipSink f) <*> (ZipSink x) =+         ZipSink $ liftM (uncurry ($)) $ zipSinks f x++-- | Send incoming values to all of the @Sink@ providing, and ultimately+-- coalesce together all return values.+--+-- Implemented on top of @ZipSink@, see that data type for more details.+--+-- Since 1.0.13+sequenceSinks :: (Traversable f, Monad m) => f (Sink i m r) -> Sink i m (f r)+sequenceSinks = getZipSink . sequenceA . fmap ZipSink++-- | The connect-and-resume operator. This does not close the @Conduit@, but+-- instead returns it to be used again. This allows a @Conduit@ to be used+-- incrementally in a large program, without forcing the entire program to live+-- in the @Sink@ monad.+--+-- Leftover data returned from the @Sink@ will be discarded.+--+-- Mnemonic: connect + do more.+--+-- Since 1.0.17+(=$$+) :: Monad m => Conduit a m b -> Sink b m r -> Sink a m (ResumableConduit a m b, r)+(=$$+) (ConduitM conduit) = connectResumeConduit (ResumableConduit (conduit Done) (return ()))+{-# INLINE (=$$+) #-}++-- | Continue processing after usage of '=$$+'. Connect a 'ResumableConduit' to+-- a sink and return the output of the sink together with a new+-- 'ResumableConduit'.+--+-- Since 1.0.17+(=$$++) :: Monad m => ResumableConduit i m o -> Sink o m r -> Sink i m (ResumableConduit i m o, r)+(=$$++) = connectResumeConduit+{-# INLINE (=$$++) #-}++-- | Complete processing of a 'ResumableConduit'. This will run the finalizer+-- associated with the @ResumableConduit@. In order to guarantee process+-- resource finalization, you /must/ use this operator after using '=$$+' and+-- '=$$++'.+--+-- Since 1.0.17+(=$$+-) :: Monad m => ResumableConduit i m o -> Sink o m r -> Sink i m r+rsrc =$$+- sink = do+    (ResumableConduit _ final, res) <- connectResumeConduit rsrc sink+    lift final+    return res+{-# INLINE (=$$+-) #-}+++infixr 0 =$$++infixr 0 =$$+++infixr 0 =$$+-++-- | Provides an alternative @Applicative@ instance for @ConduitM@. In this instance,+-- every incoming value is provided to all @ConduitM@s, and output is coalesced together.+-- Leftovers from individual @ConduitM@s will be used within that component, and then discarded+-- at the end of their computation. Output and finalizers will both be handled in a left-biased manner.+--+-- As an example, take the following program:+--+-- @+-- main :: IO ()+-- main = do+--     let src = mapM_ yield [1..3 :: Int]+--         conduit1 = CL.map (+1)+--         conduit2 = CL.concatMap (replicate 2)+--         conduit = getZipConduit $ ZipConduit conduit1 <* ZipConduit conduit2+--         sink = CL.mapM_ print+--     src $$ conduit =$ sink+-- @+--+-- It will produce the output: 2, 1, 1, 3, 2, 2, 4, 3, 3+--+-- Since 1.0.17+newtype ZipConduit i o m r = ZipConduit { getZipConduit :: ConduitM i o m r }+    deriving Functor+instance Monad m => Applicative (ZipConduit i o m) where+    pure = ZipConduit . pure+    ZipConduit left <*> ZipConduit right = ZipConduit (zipConduitApp left right)++-- | Provide identical input to all of the @Conduit@s and combine their outputs+-- into a single stream.+--+-- Implemented on top of @ZipConduit@, see that data type for more details.+--+-- Since 1.0.17+sequenceConduits :: (Traversable f, Monad m) => f (ConduitM i o m r) -> ConduitM i o m (f r)+sequenceConduits = getZipConduit . sequenceA . fmap ZipConduit++-- | Fuse two @ConduitM@s together, and provide the return value of both. Note+-- that this will force the entire upstream @ConduitM@ to be run to produce the+-- result value, even if the downstream terminates early.+--+-- Since 1.1.5+fuseBoth :: Monad m => ConduitM a b m r1 -> ConduitM b c m r2 -> ConduitM a c m (r1, r2)+fuseBoth (ConduitM up) (ConduitM down) =+    ConduitM (pipeL (up Done) (withUpstream $ generalizeUpstream $ down Done) >>=)+{-# INLINE fuseBoth #-}++-- | Same as @fuseBoth@, but ignore the return value from the downstream+-- @Conduit@. Same caveats of forced consumption apply.+--+-- Since 1.1.5+fuseUpstream :: Monad m => ConduitM a b m r -> Conduit b m c -> ConduitM a c m r+fuseUpstream up down = fmap fst (fuseBoth up down)+{-# INLINE fuseUpstream #-}++-- 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 "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 "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+    "yield o >> p" forall o (p :: ConduitM i o m r). yield o >> p = ConduitM (HaveOutput (unConduitM p) (return ()) o)+  ; "yieldOr o c >> p" forall o c (p :: ConduitM i o m r). yieldOr o c >> p =+        ConduitM (HaveOutput (unConduitM p) c o)+  ; "when yield next" forall b o p. when b (yield o) >> p =+        if b then ConduitM (HaveOutput (unConduitM p) (return ()) o) else p+  ; "unless yield next" forall b o p. unless b (yield o) >> p =+        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 =+    ConduitM (Leftover (unConduitM p) l) #-}+    -}

@@ -0,0 +1,173 @@+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE BangPatterns #-}+module Data.Conduit.Internal.Fusion+    ( -- ** Types+      Step (..)+    , Stream (..)+    , StreamConduit+      -- ** Functions+    , streamConduit+    , streamSource+    , streamSourcePure+    , unstream+    ) where++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+-- avoid confusion with conduit names.+data Step s o r+    = Emit s o+    | Skip s+    | Stop r++data Stream m o r = forall s. Stream+    (s -> m (Step s o r))+    (m s)++data StreamConduit i o m r = StreamConduit+    (ConduitM i o m r)+    (Stream m i () -> Stream m o r)++unstream :: StreamConduit i o m r -> ConduitM i o m r+unstream (StreamConduit 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)+{-# INLINE fuseStream #-}++{-# RULES "fuseStream" forall left right.+        unstream left =$= unstream right = unstream (fuseStream left right)+  #-}++connectStream :: Monad m+              => StreamConduit () i    m ()+              -> StreamConduit i  Void m r+              -> m r+connectStream (StreamConduit _ stream) (StreamConduit _ f) =+    run $ f $ stream $ Stream emptyStep (return ())+  where+    emptyStep _ = return $ Stop ()+    run (Stream step ms0) =+        ms0 >>= loop+      where+        loop s = do+            res <- step s+            case res of+                Stop r -> return r+                Skip s' -> loop s'+                Emit _ o -> absurd o+{-# INLINE connectStream #-}++{-# RULES "connectStream" forall left right.+        unstream left $$ unstream right = connectStream left right+  #-}++connectStream1 :: Monad m+               => StreamConduit () i    m ()+               -> ConduitM      i  Void m r+               -> m r+connectStream1 (StreamConduit _ fstream) (ConduitM sink0) =+    case fstream $ Stream (const $ return $ Stop ()) (return ()) of+        Stream step ms0 ->+            let loop _ (Done r) _ = return r+                loop ls (PipeM mp) s = mp >>= flip (loop ls) s+                loop ls (Leftover p l) s = loop (l:ls) p s+                loop _ (HaveOutput _ _ o) _ = absurd o+                loop (l:ls) (NeedInput p _) s = loop ls (p l) s+                loop [] (NeedInput p c) s = do+                    res <- step s+                    case res of+                        Stop () -> loop [] (c ()) s+                        Skip s' -> loop [] (NeedInput p c) s'+                        Emit s' i -> loop [] (p i) s'+             in ms0 >>= loop [] (sink0 Done)+{-# INLINE connectStream1 #-}++{-# RULES "connectStream1" forall left right.+        unstream left $$ right = connectStream1 left right+  #-}++{-++Not only will this rule not fire reliably, but due to finalizers, it can change+behavior unless implemented very carefully. Odds are that the careful+implementation won't be any faster, so leaving this commented out for now.++connectStream2 :: Monad m+               => ConduitM      () i    m ()+               -> StreamConduit i  Void m r+               -> m r+connectStream2 (ConduitM src0) (StreamConduit _ fstream) =+    run $ fstream $ Stream step' $ return (return (), src0 Done)+  where+    step' (_, Done ()) = return $ Stop ()+    {-# INLINE step' #-}++    run (Stream step ms0) =+        ms0 >>= loop+      where+        loop s = do+            res <- step s+            case res of+                Stop r -> return r+                Emit _ o -> absurd o+                Skip s' -> loop s'+{-# INLINE connectStream2 #-}++{-# RULES "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+{-# INLINE CONLIKE streamConduit #-}++streamSource+    :: Monad m+    => Stream m o ()+    -> StreamConduit i o m ()+streamSource str@(Stream step ms0) =+    StreamConduit con (const str)+  where+    con = ConduitM $ \rest -> PipeM $ do+        s0 <- ms0+        let loop s = do+                res <- step s+                case res of+                    Stop () -> return $ rest ()+                    Emit s' o -> return $ HaveOutput (PipeM $ loop s') (return ()) o+                    Skip s' -> loop s'+        loop s0+{-# INLINE streamSource #-}++streamSourcePure+    :: Monad m+    => Stream Identity o ()+    -> StreamConduit i o m ()+streamSourcePure (Stream step ms0) =+    StreamConduit con (const $ Stream (return . runIdentity . step) (return s0))+  where+    s0 = runIdentity ms0+    con = ConduitM $ \rest ->+        let loop s =+                case runIdentity $ step s of+                    Stop () -> rest ()+                    Emit s' o -> HaveOutput (loop s') (return ()) o+                    Skip s' -> loop s'+         in loop s0+{-# INLINE streamSourcePure #-}

@@ -0,0 +1,632 @@+{-# OPTIONS_HADDOCK not-home #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ImpredicativeTypes #-}+module Data.Conduit.Internal.Pipe+    ( -- ** Types+      Pipe (..)+      -- ** Primitives+    , await+    , awaitE+    , awaitForever+    , yield+    , yieldM+    , yieldOr+    , leftover+      -- ** Finalization+    , bracketP+    , addCleanup+      -- ** Composition+    , idP+    , pipe+    , pipeL+    , runPipe+    , injectLeftovers+    , (>+>)+    , (<+<)+      -- ** Exceptions+    , catchP+    , handleP+    , tryP+      -- ** Utilities+    , transPipe+    , mapOutput+    , mapOutputMaybe+    , mapInput+    , sourceList+    , withUpstream+    , Data.Conduit.Internal.Pipe.enumFromTo+    , generalizeUpstream+    ) where++import Control.Applicative (Applicative (..))+import Control.Exception.Lifted as E (Exception, catch)+import Control.Monad ((>=>), liftM, ap)+import Control.Monad.Error.Class(MonadError(..))+import Control.Monad.Reader.Class(MonadReader(..))+import Control.Monad.RWS.Class(MonadRWS())+import Control.Monad.Writer.Class(MonadWriter(..))+import Control.Monad.State.Class(MonadState(..))+import Control.Monad.Trans.Class (MonadTrans (lift))+import Control.Monad.IO.Class (MonadIO (liftIO))+import Control.Monad.Base (MonadBase (liftBase))+import Data.Void (Void, absurd)+import Data.Monoid (Monoid (mappend, mempty))+import Control.Monad.Trans.Resource+import qualified GHC.Exts+import Control.Monad.Morph (MFunctor (..))+import qualified Control.Monad.Catch as Catch++-- | The underlying datatype for all the types in this package.  In has six+-- type parameters:+--+-- * /l/ is the type of values that may be left over from this @Pipe@. A @Pipe@+-- with no leftovers would use @Void@ here, and one with leftovers would use+-- the same type as the /i/ parameter. Leftovers are automatically provided to+-- the next @Pipe@ in the monadic chain.+--+-- * /i/ is the type of values for this @Pipe@'s input stream.+--+-- * /o/ is the type of values for this @Pipe@'s output stream.+--+-- * /u/ is the result type from the upstream @Pipe@.+--+-- * /m/ is the underlying monad.+--+-- * /r/ is the result type.+--+-- A basic intuition is that every @Pipe@ produces a stream of output values+-- (/o/), and eventually indicates that this stream is terminated by sending a+-- result (/r/). On the receiving end of a @Pipe@, these become the /i/ and /u/+-- parameters.+--+-- Since 0.5.0+data Pipe l i o u m r =+    -- | Provide new output to be sent downstream. This constructor has three+    -- fields: the next @Pipe@ to be used, a finalization function, and the+    -- output value.+    HaveOutput (Pipe l i o u m r) (m ()) o+    -- | Request more input from upstream. The first field takes a new input+    -- value and provides a new @Pipe@. The second takes an upstream result+    -- value, which indicates that upstream is producing no more results.+  | NeedInput (i -> Pipe l i o u m r) (u -> Pipe l i o u m r)+    -- | Processing with this @Pipe@ is complete, providing the final result.+  | Done r+    -- | Require running of a monadic action to get the next @Pipe@.+  | PipeM (m (Pipe l i o u m r))+    -- | Return leftover input, which should be provided to future operations.+  | Leftover (Pipe l i o u m r) l++instance Monad m => Functor (Pipe l i o u m) where+    fmap = liftM+    {-# INLINE fmap #-}++instance Monad m => Applicative (Pipe l i o u m) where+    pure = return+    {-# INLINE pure #-}+    (<*>) = ap+    {-# INLINE (<*>) #-}++instance Monad m => Monad (Pipe l i o u m) where+    return = Done+    {-# INLINE return #-}++    HaveOutput p c o >>= fp = HaveOutput (p >>= fp)            c          o+    NeedInput p c    >>= fp = NeedInput  (p >=> fp)            (c >=> fp)+    Done x           >>= fp = fp x+    PipeM mp         >>= fp = PipeM      ((>>= fp) `liftM` mp)+    Leftover p i     >>= fp = Leftover   (p >>= fp)            i++instance MonadBase base m => MonadBase base (Pipe l i o u m) where+    liftBase = lift . liftBase+    {-# INLINE liftBase #-}++instance MonadTrans (Pipe l i o u) where+    lift mr = PipeM (Done `liftM` mr)+    {-# INLINE [1] lift #-}++instance MonadIO m => MonadIO (Pipe l i o u m) where+    liftIO = lift . liftIO+    {-# INLINE liftIO #-}++instance MonadThrow m => MonadThrow (Pipe l i o u m) where+    throwM = lift . throwM+    {-# INLINE throwM #-}++instance Catch.MonadCatch m => Catch.MonadCatch (Pipe l i o u m) where+    catch p0 onErr =+        go p0+      where+        go (Done r) = Done r+        go (PipeM mp) = PipeM $ Catch.catch (liftM go mp) (return . onErr)+        go (Leftover p i) = Leftover (go p) i+        go (NeedInput x y) = NeedInput (go . x) (go . y)+        go (HaveOutput p c o) = HaveOutput (go p) c o+    {-# INLINE catch #-}++instance Monad m => Monoid (Pipe l i o u m ()) where+    mempty = return ()+    {-# INLINE mempty #-}+    mappend = (>>)+    {-# INLINE mappend #-}++instance MonadResource m => MonadResource (Pipe l i o u m) where+    liftResourceT = lift . liftResourceT+    {-# INLINE liftResourceT #-}++instance MonadReader r m => MonadReader r (Pipe l i o u m) where+    ask = lift ask+    {-# INLINE ask #-}+    local f (HaveOutput p c o) = HaveOutput (local f p) c o+    local f (NeedInput p c) = NeedInput (\i -> local f (p i)) (\u -> local f (c u))+    local _ (Done x) = Done x+    local f (PipeM mp) = PipeM (liftM (local f) $ local f mp)+    local f (Leftover p i) = Leftover (local f p) i++-- Provided for doctest+#ifndef MIN_VERSION_mtl+#define MIN_VERSION_mtl(x, y, z) 0+#endif++instance MonadWriter w m => MonadWriter w (Pipe l i o u m) where+#if MIN_VERSION_mtl(2, 1, 0)+    writer = lift . writer+#endif++    tell = lift . tell++    listen (HaveOutput p c o) = HaveOutput (listen p) c o+    listen (NeedInput p c) = NeedInput (\i -> listen (p i)) (\u -> listen (c u))+    listen (Done x) = Done (x,mempty)+    listen (PipeM mp) =+      PipeM $+      do (p,w) <- listen mp+         return $ do (x,w') <- listen p+                     return (x, w `mappend` w')+    listen (Leftover p i) = Leftover (listen p) i++    pass (HaveOutput p c o) = HaveOutput (pass p) c o+    pass (NeedInput p c) = NeedInput (\i -> pass (p i)) (\u -> pass (c u))+    pass (PipeM mp) = PipeM $ mp >>= (return . pass)+    pass (Done (x,_)) = Done x+    pass (Leftover p i) = Leftover (pass p) i++instance MonadState s m => MonadState s (Pipe l i o u m) where+    get = lift get+    put = lift . put+#if MIN_VERSION_mtl(2, 1, 0)+    state = lift . state+#endif++instance MonadRWS r w s m => MonadRWS r w s (Pipe l i o u m)++instance MonadError e m => MonadError e (Pipe l i o u m) where+    throwError = lift . throwError+    catchError (HaveOutput p c o) f = HaveOutput (catchError p f) c o+    catchError (NeedInput p c) f = NeedInput (\i -> catchError (p i) f) (\u -> catchError (c u) f)+    catchError (Done x) _ = Done x+    catchError (PipeM mp) f =+      PipeM $ catchError (liftM (flip catchError f) mp) (\e -> return (f e))+    catchError (Leftover p i) f = Leftover (catchError p f) i++-- | Wait for a single input value from upstream.+--+-- 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) #-}+{-# INLINE [1] await #-}++-- | This is similar to @await@, but will return the upstream result value as+-- @Left@ if available.+--+-- 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 #-}+{-# INLINE [1] awaitE #-}++-- | Wait for input forever, calling the given inner @Pipe@ for each piece of+-- new input. Returns the upstream result type.+--+-- Since 0.5.0+awaitForever :: Monad m => (i -> Pipe l i o r m r') -> Pipe l i o r m r+awaitForever inner =+    self+  where+    self = awaitE >>= either return (\i -> inner i >> self)+{-# INLINE [1] awaitForever #-}++-- | Send a single output value downstream. If the downstream @Pipe@+-- terminates, this @Pipe@ will terminate as well.+--+-- Since 0.5.0+yield :: Monad m+      => o -- ^ output value+      -> Pipe l i o u m ()+yield = HaveOutput (Done ()) (return ())+{-# INLINE [1] yield #-}++yieldM :: Monad m => m o -> Pipe l i o u m ()+yieldM = PipeM . liftM (HaveOutput (Done ()) (return ()))+{-# INLINE [1] yieldM #-}++-- | Similar to @yield@, but additionally takes a finalizer to be run if the+-- downstream @Pipe@ terminates.+--+-- Since 0.5.0+yieldOr :: Monad m+        => o+        -> m () -- ^ finalizer+        -> Pipe l i o u m ()+yieldOr o f = HaveOutput (Done ()) f o+{-# INLINE [1] yieldOr #-}++{-# RULES+    "CI.yield o >> p" forall o (p :: Pipe l i o u m r). yield o >> p = HaveOutput p (return ()) o+  ; "CI.yieldOr o c >> p" forall o c (p :: Pipe l i o u m r). yieldOr o c >> p = HaveOutput p c o+  ; "lift m >>= CI.yield" forall m. lift m >>= yield = yieldM m+  #-}+  -- FIXME: Too much inlining on mapM_, can't enforce; "mapM_ CI.yield" mapM_ yield = sourceList+  -- Maybe we can get a rewrite rule on foldr instead? Need a benchmark to back this up.++-- | Provide a single piece of leftover input to be consumed by the next pipe+-- in the current monadic binding.+--+-- /Note/: it is highly encouraged to only return leftover values from input+-- already consumed from upstream.+--+-- Since 0.5.0+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 #-}++-- | Perform some allocation and run an inner @Pipe@. Two guarantees are given+-- about resource finalization:+--+-- 1. It will be /prompt/. The finalization will be run as early as possible.+--+-- 2. It is exception safe. Due to usage of @resourcet@, the finalization will+--    be run in the event of any exceptions.+--+-- Since 0.5.0+bracketP :: MonadResource m+         => IO a+         -> (a -> IO ())+         -> (a -> Pipe l i o u m r)+         -> Pipe l i o u m r+bracketP alloc free inside =+    PipeM start+  where+    start = do+        (key, seed) <- allocate alloc free+        return $ addCleanup (const $ release key) (inside seed)++-- | Add some code to be run when the given @Pipe@ cleans up.+--+-- Since 0.4.1+addCleanup :: Monad m+           => (Bool -> m ()) -- ^ @True@ if @Pipe@ ran to completion, @False@ for early termination.+           -> Pipe l i o u m r+           -> Pipe l i o u m r+addCleanup cleanup (Done r) = PipeM (cleanup True >> return (Done r))+addCleanup cleanup (HaveOutput src close x) = HaveOutput+    (addCleanup cleanup src)+    (cleanup False >> close)+    x+addCleanup cleanup (PipeM msrc) = PipeM (liftM (addCleanup cleanup) msrc)+addCleanup cleanup (NeedInput p c) = NeedInput+    (addCleanup cleanup . p)+    (addCleanup cleanup . c)+addCleanup cleanup (Leftover p i) = Leftover (addCleanup cleanup p) i++-- | The identity @Pipe@.+--+-- Since 0.5.0+idP :: Monad m => Pipe l a a r m r+idP = NeedInput (HaveOutput idP (return ())) Done++-- | Compose a left and right pipe together into a complete pipe. The left pipe+-- will be automatically closed when the right pipe finishes.+--+-- Since 0.5.0+pipe :: Monad m => Pipe l a b r0 m r1 -> Pipe Void b c r1 m r2 -> Pipe l a c r0 m r2+pipe =+    goRight (return ())+  where+    goRight final left right =+        case right of+            HaveOutput p c o -> HaveOutput (recurse p) (c >> final) o+            NeedInput rp rc  -> goLeft rp rc final left+            Done r2          -> PipeM (final >> return (Done r2))+            PipeM mp         -> PipeM (liftM recurse mp)+            Leftover _ i     -> absurd i+      where+        recurse = goRight final left++    goLeft rp rc final left =+        case left of+            HaveOutput left' final' o -> goRight final' left' (rp o)+            NeedInput left' lc        -> NeedInput (recurse . left') (recurse . lc)+            Done r1                   -> goRight (return ()) (Done r1) (rc r1)+            PipeM mp                  -> PipeM (liftM recurse mp)+            Leftover left' i          -> Leftover (recurse left') i+      where+        recurse = goLeft rp rc final++-- | Same as 'pipe', but automatically applies 'injectLeftovers' to the right @Pipe@.+--+-- Since 0.5.0+pipeL :: Monad m => Pipe l a b r0 m r1 -> Pipe b b c r1 m r2 -> Pipe l a c r0 m r2+-- Note: The following should be equivalent to the simpler:+--+--     pipeL l r = l `pipe` injectLeftovers r+--+-- However, this version tested as being significantly more efficient.+pipeL =+    goRight (return ())+  where+    goRight final left right =+        case right of+            HaveOutput p c o  -> HaveOutput (recurse p) (c >> final) o+            NeedInput rp rc   -> goLeft rp rc final left+            Done r2           -> PipeM (final >> return (Done r2))+            PipeM mp          -> PipeM (liftM recurse mp)+            Leftover right' i -> goRight final (HaveOutput left final i) right'+      where+        recurse = goRight final left++    goLeft rp rc final left =+        case left of+            HaveOutput left' final' o -> goRight final' left' (rp o)+            NeedInput left' lc        -> NeedInput (recurse . left') (recurse . lc)+            Done r1                   -> goRight (return ()) (Done r1) (rc r1)+            PipeM mp                  -> PipeM (liftM recurse mp)+            Leftover left' i          -> Leftover (recurse left') i+      where+        recurse = goLeft rp rc final++-- | Run a pipeline until processing completes.+--+-- Since 0.5.0+runPipe :: Monad m => Pipe Void () Void () m r -> m r+runPipe (HaveOutput _ _ o) = absurd o+runPipe (NeedInput _ c) = runPipe (c ())+runPipe (Done r) = return r+runPipe (PipeM mp) = mp >>= runPipe+runPipe (Leftover _ i) = absurd i++-- | Transforms a @Pipe@ that provides leftovers to one which does not,+-- allowing it to be composed.+--+-- This function will provide any leftover values within this @Pipe@ to any+-- calls to @await@. If there are more leftover values than are demanded, the+-- remainder are discarded.+--+-- Since 0.5.0+injectLeftovers :: Monad m => Pipe i i o u m r -> Pipe l i o u m r+injectLeftovers =+    go []+  where+    go ls (HaveOutput p c o) = HaveOutput (go ls p) c o+    go (l:ls) (NeedInput p _) = go ls $ p l+    go [] (NeedInput p c) = NeedInput (go [] . p) (go [] . c)+    go _ (Done r) = Done r+    go ls (PipeM mp) = PipeM (liftM (go ls) mp)+    go ls (Leftover p l) = go (l:ls) p++-- | Transform the monad that a @Pipe@ lives in.+--+-- Note that the monad transforming function will be run multiple times,+-- resulting in unintuitive behavior in some cases. For a fuller treatment,+-- please see:+--+-- <https://github.com/snoyberg/conduit/wiki/Dealing-with-monad-transformers>+--+-- This function is just a synonym for 'hoist'.+--+-- Since 0.4.0+transPipe :: Monad m => (forall a. m a -> n a) -> Pipe l i o u m r -> Pipe l i o u n r+transPipe f (HaveOutput p c o) = HaveOutput (transPipe f p) (f c) o+transPipe f (NeedInput p c) = NeedInput (transPipe f . p) (transPipe f . c)+transPipe _ (Done r) = Done r+transPipe f (PipeM mp) =+    PipeM (f $ liftM (transPipe f) $ collapse mp)+  where+    -- Combine a series of monadic actions into a single action.  Since we+    -- throw away side effects between different actions, an arbitrary break+    -- between actions will lead to a violation of the monad transformer laws.+    -- Example available at:+    --+    -- http://hpaste.org/75520+    collapse mpipe = do+        pipe' <- mpipe+        case pipe' of+            PipeM mpipe' -> collapse mpipe'+            _ -> return pipe'+transPipe f (Leftover p i) = Leftover (transPipe f p) i++-- | Apply a function to all the output values of a @Pipe@.+--+-- This mimics the behavior of `fmap` for a `Source` and `Conduit` in pre-0.4+-- days.+--+-- Since 0.4.1+mapOutput :: Monad m => (o1 -> o2) -> Pipe l i o1 u m r -> Pipe l i o2 u m r+mapOutput f =+    go+  where+    go (HaveOutput p c o) = HaveOutput (go p) c (f o)+    go (NeedInput p c) = NeedInput (go . p) (go . c)+    go (Done r) = Done r+    go (PipeM mp) = PipeM (liftM (go) mp)+    go (Leftover p i) = Leftover (go p) i+{-# INLINE mapOutput #-}++-- | Same as 'mapOutput', but use a function that returns @Maybe@ values.+--+-- Since 0.5.0+mapOutputMaybe :: Monad m => (o1 -> Maybe o2) -> Pipe l i o1 u m r -> Pipe l i o2 u m r+mapOutputMaybe f =+    go+  where+    go (HaveOutput p c o) = maybe id (\o' p' -> HaveOutput p' c o') (f o) (go p)+    go (NeedInput p c) = NeedInput (go . p) (go . c)+    go (Done r) = Done r+    go (PipeM mp) = PipeM (liftM (go) mp)+    go (Leftover p i) = Leftover (go p) i+{-# INLINE mapOutputMaybe #-}++-- | Apply a function to all the input values of a @Pipe@.+--+-- Since 0.5.0+mapInput :: Monad m+         => (i1 -> i2) -- ^ map initial input to new input+         -> (l2 -> Maybe l1) -- ^ map new leftovers to initial leftovers+         -> Pipe l2 i2 o u m r+         -> Pipe l1 i1 o u m r+mapInput f f' (HaveOutput p c o) = HaveOutput (mapInput f f' p) c o+mapInput f f' (NeedInput p c)    = NeedInput (mapInput f f' . p . f) (mapInput f f' . c)+mapInput _ _  (Done r)           = Done r+mapInput f f' (PipeM mp)         = PipeM (liftM (mapInput f f') mp)+mapInput f f' (Leftover p i)     = maybe id (flip Leftover) (f' i) $ mapInput f f' p++enumFromTo :: (Enum o, Eq o, Monad m)+           => o+           -> o+           -> Pipe l i o u m ()+enumFromTo start stop =+    loop start+  where+    loop i+        | i == stop = HaveOutput (Done ()) (return ()) i+        | otherwise = HaveOutput (loop (succ i)) (return ()) i+{-# INLINE enumFromTo #-}++-- | Convert a list into a source.+--+-- Since 0.3.0+sourceList :: Monad m => [a] -> Pipe l i a u m ()+sourceList =+    go+  where+    go [] = Done ()+    go (o:os) = HaveOutput (go os) (return ()) o+{-# INLINE [1] sourceList #-}++-- | The equivalent of @GHC.Exts.build@ for @Pipe@.+--+-- Since 0.4.2+build :: Monad m => (forall b. (o -> b -> b) -> b -> b) -> Pipe l i o u m ()+build g = g (\o p -> HaveOutput p (return ()) o) (return ())++{-# RULES+    "sourceList/build" forall (f :: (forall b. (a -> b -> b) -> b -> b)). sourceList (GHC.Exts.build f) = build f #-}++-- | Returns a tuple of the upstream and downstream results. Note that this+-- will force consumption of the entire input stream.+--+-- Since 0.5.0+withUpstream :: Monad m+             => Pipe l i o u m r+             -> Pipe l i o u m (u, r)+withUpstream down =+    down >>= go+  where+    go r =+        loop+      where+        loop = awaitE >>= either (\u -> return (u, r)) (\_ -> loop)++infixr 9 <+<+infixl 9 >+>++-- | Fuse together two @Pipe@s, connecting the output from the left to the+-- input of the right.+--+-- Notice that the /leftover/ parameter for the @Pipe@s must be @Void@. This+-- ensures that there is no accidental data loss of leftovers during fusion. If+-- you have a @Pipe@ with leftovers, you must first call 'injectLeftovers'.+--+-- Since 0.5.0+(>+>) :: Monad m => Pipe l a b r0 m r1 -> Pipe Void b c r1 m r2 -> Pipe l a c r0 m r2+(>+>) = pipe+{-# INLINE (>+>) #-}++-- | Same as '>+>', but reverse the order of the arguments.+--+-- Since 0.5.0+(<+<) :: Monad m => Pipe Void b c r1 m r2 -> Pipe l a b r0 m r1 -> Pipe l a c r0 m r2+(<+<) = flip pipe+{-# INLINE (<+<) #-}++-- | Since 1.0.4+instance MFunctor (Pipe l i o u) where+    hoist = transPipe++-- | See 'catchC' for more details.+--+-- Since 1.0.11+catchP :: (MonadBaseControl IO m, Exception e)+       => Pipe l i o u m r+       -> (e -> Pipe l i o u m r)+       -> Pipe l i o u m r+catchP p0 onErr =+    go p0+  where+    go (Done r) = Done r+    go (PipeM mp) = PipeM $ E.catch (liftM go mp) (return . onErr)+    go (Leftover p i) = Leftover (go p) i+    go (NeedInput x y) = NeedInput (go . x) (go . y)+    go (HaveOutput p c o) = HaveOutput (go p) c o+{-# INLINABLE catchP #-}++-- | The same as @flip catchP@.+--+-- Since 1.0.11+handleP :: (MonadBaseControl IO m, Exception e)+        => (e -> Pipe l i o u m r)+        -> Pipe l i o u m r+        -> Pipe l i o u m r+handleP = flip catchP+{-# INLINE handleP #-}++-- | See 'tryC' for more details.+--+-- Since 1.0.11+tryP :: (MonadBaseControl IO m, Exception e)+     => Pipe l i o u m r+     -> Pipe l i o u m (Either e r)+tryP =+    go+  where+    go (Done r) = Done (Right r)+    go (PipeM mp) = PipeM $ E.catch (liftM go mp) (return . Done . Left)+    go (Leftover p i) = Leftover (go p) i+    go (NeedInput x y) = NeedInput (go . x) (go . y)+    go (HaveOutput p c o) = HaveOutput (go p) c o+{-# INLINABLE tryP #-}++-- | Generalize the upstream return value for a @Pipe@ from unit to any type.+--+-- Since 1.1.5+generalizeUpstream :: Monad m => Pipe l i o () m r -> Pipe l i o u m r+generalizeUpstream =+    go+  where+    go (HaveOutput p f o) = HaveOutput (go p) f o+    go (NeedInput x y) = NeedInput (go . x) (\_ -> go (y ()))+    go (Done r) = Done r+    go (PipeM mp) = PipeM (liftM go mp)+    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) #-}

@@ -138,18 +138,18 @@ runErrorC   :: (Monad m, E.Error e) =>      ConduitM i o (E.ErrorT e m) r -> ConduitM i o m (Either e r)-runErrorC =-    ConduitM . go . unConduitM-  where-    go (Done r) = Done (Right r)-    go (PipeM mp) = PipeM $ do-        eres <- E.runErrorT mp-        return $ case eres of-            Left e -> Done $ Left e-            Right p -> go p-    go (Leftover p i) = Leftover (go p) i-    go (HaveOutput p f o) = HaveOutput (go p) (E.runErrorT f >> return ()) o-    go (NeedInput x y) = NeedInput (go . x) (go . y)+runErrorC (ConduitM c0) =+    ConduitM $ \rest ->+        let go (Done r) = rest (Right r)+            go (PipeM mp) = PipeM $ do+                eres <- E.runErrorT mp+                return $ case eres of+                    Left e -> rest $ Left e+                    Right p -> go p+            go (Leftover p i) = Leftover (go p) i+            go (HaveOutput p f o) = HaveOutput (go p) (E.runErrorT f >> return ()) o+            go (NeedInput x y) = NeedInput (go . x) (go . y)+         in go (c0 Done) {-# INLINABLE runErrorC #-}  -- | Catch an error in the base monad@@ -161,17 +161,18 @@      -> (e -> ConduitM i o (E.ErrorT e m) r)      -> ConduitM i o (E.ErrorT e m) r catchErrorC c0 h =-    ConduitM $ go $ unConduitM c0+    ConduitM $ \rest ->+        let go (Done r) = rest r+            go (PipeM mp) = PipeM $ do+                eres <- lift $ E.runErrorT mp+                return $ case eres of+                    Left e -> unConduitM (h e) rest+                    Right p -> go p+            go (Leftover p i) = Leftover (go p) i+            go (HaveOutput p f o) = HaveOutput (go p) f o+            go (NeedInput x y) = NeedInput (go . x) (go . y)+         in go $ unConduitM c0 Done   where-    go (Done r) = Done r-    go (PipeM mp) = PipeM $ do-        eres <- lift $ E.runErrorT mp-        return $ case eres of-            Left e -> unConduitM $ h e-            Right p -> go p-    go (Leftover p i) = Leftover (go p) i-    go (HaveOutput p f o) = HaveOutput (go p) f o-    go (NeedInput x y) = NeedInput (go . x) (go . y) {-# INLINABLE catchErrorC #-}  -- | Run 'CatchT' in the base monad@@ -180,18 +181,18 @@ runCatchC   :: Monad m =>      ConduitM i o (CatchT m) r -> ConduitM i o m (Either SomeException r)-runCatchC =-    ConduitM . go . unConduitM-  where-    go (Done r) = Done (Right r)-    go (PipeM mp) = PipeM $ do-        eres <- runCatchT mp-        return $ case eres of-            Left e -> Done $ Left e-            Right p -> go p-    go (Leftover p i) = Leftover (go p) i-    go (HaveOutput p f o) = HaveOutput (go p) (runCatchT f >> return ()) o-    go (NeedInput x y) = NeedInput (go . x) (go . y)+runCatchC c0 =+    ConduitM $ \rest ->+        let go (Done r) = rest (Right r)+            go (PipeM mp) = PipeM $ do+                eres <- runCatchT mp+                return $ case eres of+                    Left e -> rest $ Left e+                    Right p -> go p+            go (Leftover p i) = Leftover (go p) i+            go (HaveOutput p f o) = HaveOutput (go p) (runCatchT f >> return ()) o+            go (NeedInput x y) = NeedInput (go . x) (go . y)+         in go $ unConduitM c0 Done {-# INLINABLE runCatchC #-}  -- | Catch an exception in the base monad@@ -202,18 +203,18 @@      ConduitM i o (CatchT m) r      -> (SomeException -> ConduitM i o (CatchT m) r)      -> ConduitM i o (CatchT m) r-catchCatchC c0 h =-    ConduitM $ go $ unConduitM c0-  where-    go (Done r) = Done r-    go (PipeM mp) = PipeM $ do-        eres <- lift $ runCatchT mp-        return $ case eres of-            Left e -> unConduitM $ h e-            Right p -> go p-    go (Leftover p i) = Leftover (go p) i-    go (HaveOutput p f o) = HaveOutput (go p) f o-    go (NeedInput x y) = NeedInput (go . x) (go . y)+catchCatchC (ConduitM c0) h =+    ConduitM $ \rest ->+        let go (Done r) = rest r+            go (PipeM mp) = PipeM $ do+                eres <- lift $ runCatchT mp+                return $ case eres of+                    Left e -> unConduitM (h e) rest+                    Right p -> go p+            go (Leftover p i) = Leftover (go p) i+            go (HaveOutput p f o) = HaveOutput (go p) f o+            go (NeedInput x y) = NeedInput (go . x) (go . y)+         in go (c0 Done) {-# INLINABLE catchCatchC #-}  -- | Wrap the base monad in 'M.MaybeT'@@ -235,18 +236,18 @@ runMaybeC   :: Monad m =>      ConduitM i o (M.MaybeT m) r -> ConduitM i o m (Maybe r)-runMaybeC =-    ConduitM . go . unConduitM-  where-    go (Done r) = Done (Just r)-    go (PipeM mp) = PipeM $ do-        mres <- M.runMaybeT mp-        return $ case mres of-            Nothing -> Done Nothing-            Just p -> go p-    go (Leftover p i) = Leftover (go p) i-    go (HaveOutput p c o) = HaveOutput (go p) (M.runMaybeT c >> return ()) o-    go (NeedInput x y) = NeedInput (go . x) (go . y)+runMaybeC (ConduitM c0) =+    ConduitM $ \rest ->+        let go (Done r) = rest (Just r)+            go (PipeM mp) = PipeM $ do+                mres <- M.runMaybeT mp+                return $ case mres of+                    Nothing -> rest Nothing+                    Just p -> go p+            go (Leftover p i) = Leftover (go p) i+            go (HaveOutput p c o) = HaveOutput (go p) (M.runMaybeT c >> return ()) o+            go (NeedInput x y) = NeedInput (go . x) (go . y)+         in go (c0 Done) {-# INLINABLE runMaybeC #-}  -- | Wrap the base monad in 'R.ReaderT'@@ -293,16 +294,16 @@        -> s        -> ConduitM i o (t m) r        -> ConduitM i o m res-thread toRes runM s0 =-    ConduitM . go s0 . unConduitM-  where-    go s (Done r) = Done (toRes r s)-    go s (PipeM mp) = PipeM $ do-        (p, s') <- runM mp s-        return $ go s' p-    go s (Leftover p i) = Leftover (go s p) i-    go s (NeedInput x y) = NeedInput (go s . x) (go s . y)-    go s (HaveOutput p f o) = HaveOutput (go s p) (runM f s >> return ()) o+thread toRes runM s0 (ConduitM c0) =+    ConduitM $ \rest ->+        let go s (Done r) = rest (toRes r s)+            go s (PipeM mp) = PipeM $ do+                (p, s') <- runM mp s+                return $ go s' p+            go s (Leftover p i) = Leftover (go s p) i+            go s (NeedInput x y) = NeedInput (go s . x) (go s . y)+            go s (HaveOutput p f o) = HaveOutput (go s p) (runM f s >> return ()) o+         in go s0 (c0 Done) {-# INLINABLE thread #-}  -- | Run 'SL.StateT' in the base monad@@ -568,4 +569,3 @@ execRWSC i s p = fmap f $ runRWSC i s p   where f x = let (_, s2, w2) = x in (s2, w2) {-# INLINABLE execRWSC #-}-

@@ -1,4 +1,6 @@ {-# LANGUAGE RankNTypes #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE TupleSections #-} -- | Higher-level functions to interact with the elements of a stream. Most of -- these are based on list functions. --@@ -16,6 +18,8 @@     , unfoldM     , enumFromTo     , iterate+    , replicate+    , replicateM       -- * Sinks       -- ** Pure     , fold@@ -69,15 +73,15 @@     , (>>=)     , seq     , otherwise-    , Enum (succ), Eq+    , Enum, Eq     , maybe-    , either     , (<=)     ) 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 Control.Monad (when, (<=<), liftM, void) import Control.Monad.Trans.Class (lift) @@ -121,14 +125,54 @@ -- combining with @sourceList@ since this avoids any intermediate data -- structures. --+-- Subject to fusion+-- -- Since 0.4.2-enumFromTo :: (Enum a, Eq a, Monad m)+enumFromTo :: (Enum a, Prelude.Ord a, Monad m)            => a            -> a            -> Producer m a-enumFromTo x = CI.ConduitM . CI.enumFromTo x-{-# INLINE enumFromTo #-}+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+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 ()+  #-}+ -- | Produces an infinite stream of repeated applications of f to x. iterate :: Monad m => (a -> a) -> a -> Producer m a iterate f =@@ -136,31 +180,121 @@   where     go a = yield a >> go (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+replicateC cnt0 a =+    loop cnt0+  where+    loop i+        | 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 #-}++-- | Replicate a monadic value the given number of times.+--+-- 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+replicateMC cnt0 ma =+    loop cnt0+  where+    loop i+        | 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 #-}+ -- | A strict left fold. --+-- Subject to fusion+-- -- Since 0.3.0 fold :: Monad m      => (b -> a -> b)      -> b      -> Consumer a m b-fold f =+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+foldC f =     loop   where-    loop accum =-        await >>= maybe (return accum) go-      where-        go a =-            let accum' = f accum a-             in accum' `seq` loop accum'+    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 #-}+ -- | 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 f =+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+foldMC f =     loop   where     loop accum = do@@ -169,7 +303,56 @@         go a = do             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 #-}++-----------------------------------------------------------------+-- These are for cases where- for whatever reason- stream fusion cannot be+-- applied.+connectFold :: Monad m => Source m a -> (b -> a -> b) -> b -> m b+connectFold (CI.ConduitM src0) f =+    go (src0 CI.Done)+  where+    go (CI.Done ()) b = return b+    go (CI.HaveOutput src _ a) b = go src Prelude.$! f b a+    go (CI.NeedInput _ c) b = go (c ()) b+    go (CI.Leftover src ()) b = go src b+    go (CI.PipeM msrc) b = do+        src <- msrc+        go src b+{-# INLINE connectFold #-}+{-# RULES "$$ 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 =+    go (src0 CI.Done)+  where+    go (CI.Done ()) b = return b+    go (CI.HaveOutput src _ a) b = do+        !b' <- f b a+        go src b'+    go (CI.NeedInput _ c) b = go (c ()) b+    go (CI.Leftover src ()) b = go src b+    go (CI.PipeM msrc) b = do+        src <- msrc+        go src b+{-# INLINE connectFoldM #-}+{-# RULES "$$ foldM" forall src f b. src $$ foldM f b = connectFoldM src f b #-}+-----------------------------------------------------------------+ -- | A monoidal strict left fold. -- -- Since 0.5.3@@ -203,7 +386,7 @@  srcMapM_ :: Monad m => Source m a -> (a -> m ()) -> m () srcMapM_ (CI.ConduitM src) f =-    go src+    go (src CI.Done)   where     go (CI.Done ()) = return ()     go (CI.PipeM mp) = mp >>= go@@ -244,7 +427,7 @@ take =     loop id   where-    loop front 0 = return $ front []+    loop front count | count <= 0 = return $ front []     loop front count = await >>= maybe         (return $ front [])         (\x -> loop (front .(x:)) (count - 1))@@ -264,18 +447,40 @@  -- | Apply a transformation to all values in a stream. --+-- Subject to fusion+-- -- Since 0.3.0 map :: Monad m => (a -> b) -> Conduit a m b-map f = awaitForever $ yield . f-{-# INLINE [1] map #-}+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+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 #-}+ -- 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 "source/map fusion =$=" forall f src. src =$= map f = mapFuseRight src f #-}  mapFuseRight :: Monad m => Source m a -> (a -> b) -> Source m b-mapFuseRight (CI.ConduitM src) f = CI.ConduitM (CI.mapOutput f src)+mapFuseRight src f = CIC.mapOutput f src {-# INLINE mapFuseRight #-}+-}  {- @@ -304,10 +509,32 @@ -- If you do not need the transformed values, and instead just want the monadic -- side-effects of running the action, see 'mapM_'. --+-- Subject to fusion+-- -- Since 0.3.0 mapM :: Monad m => (a -> m b) -> Conduit a m b-mapM f = awaitForever $ yield <=< lift . f+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+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 #-}+ -- | Apply a monadic action on all values in a stream. -- -- This @Conduit@ can be used to perform a monadic side-effect for every@@ -446,13 +673,33 @@ -- will pull all values into memory. For a lazy variant, see -- "Data.Conduit.Lazy". --+-- Subject to fusion+-- -- Since 0.3.0 consume :: Monad m => Consumer a m [a]-consume =+consume = consumeC+{-# INLINE [0] consume #-}+{-# RULES "unstream consume" consume = unstream (streamConduit consumeC consumeS) #-}++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 #-}+ -- | Grouping input according to an equality function. -- -- Since 0.3.0@@ -515,7 +762,7 @@ isolate =     loop   where-    loop 0 = return ()+    loop count | count <= 0 = return ()     loop count = await >>= maybe (return ()) (\x -> yield x >> loop (count - 1))  -- | Keep only values in the stream passing a given predicate.@@ -525,20 +772,18 @@ filter f = awaitForever $ \i -> when (f i) (yield i)  filterFuseRight :: Monad m => Source m a -> (a -> Bool) -> Source m a-filterFuseRight (CI.ConduitM src) f =-    CI.ConduitM (go src)-  where-    go (CI.Done ()) = CI.Done ()+filterFuseRight (CI.ConduitM src) f = CI.ConduitM $ \rest -> let+    go (CI.Done ()) = rest ()     go (CI.PipeM mp) = CI.PipeM (liftM go mp)     go (CI.Leftover p i) = CI.Leftover (go p) i     go (CI.HaveOutput p c o)         | f o = CI.HaveOutput (go p) c o         | otherwise = go p     go (CI.NeedInput p c) = CI.NeedInput (go . p) (go . c)+    in go (src CI.Done) -- 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 "source/filter fusion =$=" forall f src. src =$= filter f = filterFuseRight src f #-} {-# INLINE filterFuseRight #-} @@ -552,7 +797,7 @@  srcSinkNull :: Monad m => Source m a -> m () srcSinkNull (CI.ConduitM src) =-    go src+    go (src CI.Done)   where     go (CI.Done ()) = return ()     go (CI.PipeM mp) = mp >>= go

@@ -0,0 +1,427 @@+{-# LANGUAGE BangPatterns              #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE FlexibleContexts          #-}+{-# LANGUAGE MultiParamTypeClasses     #-}+{-# LANGUAGE TupleSections             #-}+{-# LANGUAGE RankNTypes                #-}+-- Collection of three benchmarks: a simple integral sum, monte carlo analysis,+-- and sliding vector.+import           Control.DeepSeq+import           Control.Monad               (foldM)+import           Control.Monad               (when, liftM)+import           Control.Monad.Codensity     (lowerCodensity)+import           Control.Monad.IO.Class      (MonadIO, liftIO)+import           Control.Monad.Trans.Class   (lift)+import           Criterion.Main+import           Data.Conduit+import           Data.Conduit.Internal       (ConduitM (..), Pipe (..))+import qualified Data.Conduit.Internal       as CI+import qualified Data.Conduit.List           as CL+import qualified Data.Foldable               as F+import           Data.Functor.Identity       (runIdentity)+import           Data.IORef+import           Data.List                   (foldl')+import           Data.Monoid                 (mempty)+import qualified Data.Sequence               as Seq+import qualified Data.Vector                 as VB+import qualified Data.Vector.Generic         as V+import qualified Data.Vector.Generic.Mutable as VM+import qualified Data.Vector.Unboxed         as VU+import           System.Environment          (withArgs)+import qualified System.Random.MWC           as MWC+import           Test.Hspec++data TestBench = TBGroup String [TestBench]+               | TBBench Benchmark+               | forall a b. (Eq b, Show b) => TBPure String a b (a -> b)+               | forall a. (Eq a, Show a) => TBIO String a (IO a)+               | forall a. (Eq a, Show a) => TBIOTest String (a -> IO ()) (IO a)+               | forall a. (Eq a, Show a) => TBIOBench String a (IO a) (IO ())++toSpec :: TestBench -> Spec+toSpec (TBGroup name tbs) = describe name $ mapM_ toSpec tbs+toSpec (TBBench _) = return ()+toSpec (TBPure name a b f) = it name $ f a `shouldBe` b+toSpec (TBIO name a f) = it name $ f >>= (`shouldBe` a)+toSpec (TBIOTest name spec f) = it name $ f >>= spec+toSpec (TBIOBench name a f _) = it name $ f >>= (`shouldBe` a)++toBench :: TestBench -> Benchmark+toBench (TBGroup name tbs) = bgroup name $ map toBench tbs+toBench (TBBench b) = b+toBench (TBPure name a _ f) = bench name $ whnf f a+toBench (TBIO name _ f) = bench name $ whnfIO f+toBench (TBIOTest name _ f) = bench name $ whnfIO f+toBench (TBIOBench name _ _ f) = bench name $ whnfIO f++runTestBench :: [TestBench] -> IO ()+runTestBench tbs = do+    withArgs [] $ hspec $ mapM_ toSpec tbs+    defaultMain $ map toBench tbs++main :: IO ()+main = runTestBench =<< sequence+    [ sumTB+    , mapSumTB+    , monteCarloTB+    , fmap (TBGroup "sliding window") $ sequence+        [ slidingWindow 10+        , slidingWindow 30+        , slidingWindow 100+        , slidingWindow 1000+        ]+    ]++-----------------------------------------------------------------------++sumTB :: IO TestBench+sumTB = do+    upperRef <- newIORef upper0+    return $ TBGroup "sum"+        [ TBPure "Data.List.foldl'" upper0 expected+            $ \upper -> foldl' (+) 0 [1..upper]+        , TBIO "Control.Monad.foldM" expected $ do+            upper <- readIORef upperRef+            foldM plusM 0 [1..upper]+        , TBPure "low level" upper0 expected $ \upper ->+            let go x !t+                    | x > upper = t+                    | otherwise = go (x + 1) (t + x)+             in go 1 0+        , TBIO "boxed vectors, I/O" expected $ do+            upper <- readIORef upperRef+            VB.foldM' plusM 0 $ VB.enumFromTo 1 upper+        , TBPure "boxed vectors" upper0 expected+            $ \upper -> VB.foldl' (+) 0 (VB.enumFromTo 1 upper)+        , TBPure "unboxed vectors" upper0 expected+            $ \upper -> VU.foldl' (+) 0 (VU.enumFromTo 1 upper)+        , TBPure "conduit, pure, fold" upper0 expected+            $ \upper -> runIdentity $ CL.enumFromTo 1 upper $$ CL.fold (+) 0+        , TBPure "conduit, pure, foldM" upper0 expected+            $ \upper -> runIdentity $ CL.enumFromTo 1 upper $$ CL.foldM plusM 0+        , TBIO "conduit, IO, fold" expected $ do+            upper <- readIORef upperRef+            CL.enumFromTo 1 upper $$ CL.fold (+) 0+        , TBIO "conduit, IO, foldM" expected $ do+            upper <- readIORef upperRef+            CL.enumFromTo 1 upper $$ CL.foldM plusM 0+        ]+  where+    upper0 = 10000 :: Int+    expected = sum [1..upper0]++    plusM x y = return $! x + y++-----------------------------------------------------------------------++mapSumTB :: IO TestBench+mapSumTB = return $ TBGroup "map + sum"+    [ TBPure "boxed vectors" upper0 expected+        $ \upper -> VB.foldl' (+) 0+                  $ VB.map (+ 1)+                  $ VB.map (* 2)+                  $ VB.enumFromTo 1 upper+    , TBPure "unboxed vectors" upper0 expected+        $ \upper -> VU.foldl' (+) 0+                  $ VU.map (+ 1)+                  $ VU.map (* 2)+                  $ VU.enumFromTo 1 upper+    , TBPure "conduit, connect1" upper0 expected $ \upper -> runIdentity+        $ CL.enumFromTo 1 upper+       $$ CL.map (* 2)+      =$= CL.map (+ 1)+      =$= CL.fold (+) 0+    , TBPure "conduit, connect2" upper0 expected $ \upper -> runIdentity+        $ CL.enumFromTo 1 upper+      =$= CL.map (* 2)+       $$ CL.map (+ 1)+      =$= CL.fold (+) 0+    , TBPure "conduit, connect3" upper0 expected $ \upper -> runIdentity+        $ CL.enumFromTo 1 upper+      =$= CL.map (* 2)+      =$= CL.map (+ 1)+       $$ CL.fold (+) 0+    , TBPure "conduit, inner fuse" upper0 expected $ \upper -> runIdentity+        $ CL.enumFromTo 1 upper+      =$= (CL.map (* 2)+      =$= CL.map (+ 1))+       $$ CL.fold (+) 0+    ]+  where+    upper0 = 10000 :: Int+    expected = sum $ map (+ 1) $ map (* 2) [1..upper0]++-----------------------------------------------------------------------++monteCarloTB :: IO TestBench+monteCarloTB = return $ TBGroup "monte carlo"+    [ TBIOTest "conduit" closeEnough $ do+        gen <- MWC.createSystemRandom+        successes <- CL.replicateM count (MWC.uniform gen)+                  $$ CL.fold (\t (x, y) ->+                                if (x*x + y*(y :: Double) < 1)+                                    then t + 1+                                    else t)+                        (0 :: Int)+        return $ fromIntegral successes / fromIntegral count * 4+    , TBIOTest "low level" closeEnough $ do+        gen <- MWC.createSystemRandom+        let go 0 !t = return $! fromIntegral t / fromIntegral count * 4+            go i !t = do+                (x, y) <- MWC.uniform gen+                let t'+                        | x*x + y*(y :: Double) < 1 = t + 1+                        | otherwise = t+                go (i - 1) t'+        go count (0 :: Int)+    ]+  where+    count = 100000 :: Int++    closeEnough x+        | abs (x - 3.14159 :: Double) < 0.2 = return ()+        | otherwise = error $ "Monte carlo analysis too inaccurate: " ++ show x++-----------------------------------------------------------------------++slidingWindow :: Int -> IO TestBench+slidingWindow window = do+    upperRef <- newIORef upper0+    return $ TBGroup (show window)+        [ TBIOBench "low level, Seq" expected+            (swLowLevelSeq window upperRef id (\x y -> x . (F.toList y:)) ($ []))+            (swLowLevelSeq window upperRef () (\() y -> rnf y) id)+        , TBIOBench "conduit, Seq" expected+            (swConduitSeq window upperRef id (\x y -> x . (F.toList y:)) ($ []))+            (swConduitSeq window upperRef () (\() y -> rnf y) id)+        {- https://ghc.haskell.org/trac/ghc/ticket/9446+        , TBIOBench "low level, boxed Vector" expected+            (swLowLevelVector window upperRef id (\x y -> x . (VB.toList y:)) ($ []))+            (swLowLevelVector window upperRef () (\() y -> rnf (y :: VB.Vector Int)) id)+            -}+        , TBBench $ bench "low level, boxed Vector" $ whnfIO $+            swLowLevelVector window upperRef () (\() y -> rnf (y :: VB.Vector Int)) id++        {- https://ghc.haskell.org/trac/ghc/ticket/9446+        , TBIOBench "conduit, boxed Vector" expected+            (swConduitVector window upperRef id (\x y -> x . (VB.toList y:)) ($ []))+            (swConduitVector window upperRef () (\() y -> rnf (y :: VB.Vector Int)) id)+        -}++        , TBBench $ bench "conduit, boxed Vector" $ whnfIO $+            swConduitVector window upperRef () (\() y -> rnf (y :: VB.Vector Int)) id+++        , TBIOBench "low level, unboxed Vector" expected+            (swLowLevelVector window upperRef id (\x y -> x . (VU.toList y:)) ($ []))+            (swLowLevelVector window upperRef () (\() y -> rnf (y :: VU.Vector Int)) id)+        , TBIOBench "conduit, unboxed Vector" expected+            (swConduitVector window upperRef id (\x y -> x . (VU.toList y:)) ($ []))+            (swConduitVector window upperRef () (\() y -> rnf (y :: VU.Vector Int)) id)+        ]+  where+    upper0 = 10000+    expected =+        loop [1..upper0]+      where+        loop input+            | length x == window = x : loop y+            | otherwise = []+          where+            x = take window input+            y = drop 1 input++swLowLevelSeq :: Int -> IORef Int -> t -> (t -> Seq.Seq Int -> t) -> (t -> t') -> IO t'+swLowLevelSeq window upperRef t0 f final = do+    upper <- readIORef upperRef++    let phase1 i !s+            | i > window = phase2 i s t0+            | otherwise = phase1 (i + 1) (s Seq.|> i)++        phase2 i !s !t+            | i > upper = t'+            | otherwise = phase2 (i + 1) s' t'+          where+            t' = f t s+            s' = Seq.drop 1 s Seq.|> i++    return $! final $! phase1 1 mempty++swLowLevelVector :: V.Vector v Int+                 => Int+                 -> IORef Int+                 -> t+                 -> (t -> v Int -> t)+                 -> (t -> t')+                 -> IO t'+swLowLevelVector window upperRef t0 f final = do+    upper <- readIORef upperRef++    let go !i !t _ _ _ | i > upper = return $! final $! t+        go !i !t !end _mv mv2 | end == bufSz  = newBuf >>= go i t sz mv2+        go !i !t !end mv mv2 = do+            VM.unsafeWrite mv end i+            when (end > sz) $ VM.unsafeWrite mv2 (end - sz) i+            let end' = end + 1+            t' <-+                if end' < sz+                    then return t+                    else do+                        v <- V.unsafeFreeze $ VM.unsafeSlice (end' - sz) sz mv+                        return $! f t v+            go (i + 1) t' end' mv mv2++    mv <- newBuf+    mv2 <- newBuf+    go 1 t0 0 mv mv2+  where+    sz = window+    bufSz = 2 * window+    newBuf = VM.new bufSz++swConduitSeq :: Int+             -> IORef Int+             -> t+             -> (t -> Seq.Seq Int -> t)+             -> (t -> t')+             -> IO t'+swConduitSeq window upperRef t0 f final = do+    upper <- readIORef upperRef++    t <- CL.enumFromTo 1 upper+        $= slidingWindowC window+        $$ CL.fold f t0+    return $! final t++swConduitVector :: V.Vector v Int+                => Int+                -> IORef Int+                -> t+                -> (t -> v Int -> t)+                -> (t -> t')+                -> IO t'+swConduitVector window upperRef t0 f final = do+    upper <- readIORef upperRef++    t <- CL.enumFromTo 1 upper+        $= slidingVectorC window+        $$ CL.fold f t0+    return $! final t++slidingWindowC :: Monad m => Int -> Conduit a m (Seq.Seq a)+slidingWindowC = slidingWindowCC+{-# INLINE [0] slidingWindowC #-}+{-# RULES "unstream slidingWindowC"+    forall i. slidingWindowC i = CI.unstream (CI.streamConduit (slidingWindowCC i) (slidingWindowS i))+  #-}++slidingWindowCC :: Monad m => Int -> Conduit a m (Seq.Seq a)+slidingWindowCC sz =+    go sz mempty+  where+    goContinue st = await >>=+                    maybe (return ())+                          (\x -> do+                             let st' = st Seq.|> x+                             yield st' >> goContinue (Seq.drop 1 st')+                          )+    go 0 st = yield st >> goContinue (Seq.drop 1 st)+    go !n st = CL.head >>= \m ->+               case m of+                 Nothing | n < sz -> yield st+                         | otherwise -> return ()+                 Just x -> go (n-1) (st Seq.|> x)+{-# INLINE slidingWindowCC #-}++slidingWindowS :: Monad m => Int -> CI.Stream m a () -> CI.Stream m (Seq.Seq a) ()+slidingWindowS sz (CI.Stream step ms0) =+    CI.Stream step' $ liftM (\s -> Left (s, sz, mempty)) ms0+  where+    step' (Left (s, 0, st)) = return $ CI.Emit (Right (s, st)) st+    step' (Left (s, i, st)) = do+        res <- step s+        return $ case res of+            CI.Stop () -> CI.Stop ()+            CI.Skip s' -> CI.Skip $ Left (s', i, st)+            CI.Emit s' a -> CI.Skip $ Left (s', i - 1, st Seq.|> a)+    step' (Right (s, st)) = do+        res <- step s+        return $ case res of+            CI.Stop () -> CI.Stop ()+            CI.Skip s' -> CI.Skip $ Right (s', st)+            CI.Emit s' a ->+                let st' = Seq.drop 1 st Seq.|> a+                 in CI.Emit (Right (s', st')) st'+{-# INLINE slidingWindowS #-}++slidingVectorC :: V.Vector v a => Int -> Conduit a IO (v a)+slidingVectorC = slidingVectorCC+{-# INLINE [0] slidingVectorC #-}+{-# RULES "unstream slidingVectorC"+    forall i. slidingVectorC i = CI.unstream (CI.streamConduit (slidingVectorCC i) (slidingVectorS i))+  #-}++slidingVectorCC :: V.Vector v a => Int -> Conduit a IO (v a)+slidingVectorCC sz = do+    mv <- newBuf+    mv2 <- newBuf+    go 0 mv mv2+  where+    bufSz = 2 * sz+    newBuf = liftIO (VM.new bufSz)++    go !end _mv mv2 | end == bufSz  = newBuf >>= go sz mv2+    go !end mv mv2 = do+      mx <- await+      case mx of+        Nothing -> when (end > 0 && end < sz) $ do+          v <- liftIO $ V.unsafeFreeze $ VM.take end mv+          yield v+        Just x -> do+          liftIO $ do+            VM.unsafeWrite mv end x+            when (end > sz) $ VM.unsafeWrite mv2 (end - sz) x+          let end' = end + 1+          when (end' >= sz) $ do+            v <- liftIO $ V.unsafeFreeze $ VM.unsafeSlice (end' - sz) sz mv+            yield v+          go end' mv mv2++slidingVectorS :: V.Vector v a => Int -> CI.Stream IO a () -> CI.Stream IO (v a) ()+slidingVectorS sz (CI.Stream step ms0) =+    CI.Stream step' ms1+  where+    bufSz = 2 * sz+    newBuf = liftIO (VM.new bufSz)++    ms1 = do+        s <- ms0+        mv <- newBuf+        mv2 <- newBuf+        return (s, 0, mv, mv2)++    step' (_, -1, _, _) = return $ CI.Stop ()+    step' (s, end, _mv, mv2) | end == bufSz = do+        mv3 <- newBuf+        return $ CI.Skip (s, sz, mv2, mv3)+    step' (s, end, mv, mv2) = do+        res <- step s+        case res of+            CI.Stop ()+                | end > 0 && end < sz -> do+                    v <- liftIO $ V.unsafeFreeze $ VM.take end mv+                    return $ CI.Emit (s, -1, mv, mv2) v+                | otherwise -> return $ CI.Stop ()+            CI.Skip s' -> return $ CI.Skip (s', end, mv, mv2)+            CI.Emit s' x -> liftIO $ do+                VM.unsafeWrite mv end x+                when (end > sz) $ VM.unsafeWrite mv2 (end - sz) x+                let end' = end + 1+                    state = (s', end', mv, mv2)+                if end' >= sz+                    then do+                        v <- V.unsafeFreeze $ VM.unsafeSlice (end' - sz) sz mv+                        return $ CI.Emit state v+                    else return $ CI.Skip state+{-# INLINE slidingVectorS #-}

@@ -0,0 +1,73 @@+{-# LANGUAGE RankNTypes, BangPatterns #-}+-- Compare low-level, fused, unfused, and partially fused+import Data.Conduit+import qualified Data.Conduit.List as CL+import Data.Conduit.Internal (Step (..), Stream (..), unstream, StreamConduit (..))+import Criterion.Main+import Data.Functor.Identity (runIdentity)++-- | unfused+enumFromToC :: (Eq a, Monad m, Enum a) => a -> a -> Producer m a+enumFromToC x0 y =+    loop x0+  where+    loop x+        | x == y = yield x+        | otherwise = yield x >> loop (succ x)+{-# INLINE enumFromToC #-}++-- | unfused+mapC :: Monad m => (a -> b) -> Conduit a m b+mapC f = awaitForever $ yield . f+{-# INLINE mapC #-}++-- | unfused+foldC :: Monad m => (b -> a -> b) -> b -> Consumer a m b+foldC f =+    loop+  where+    loop !b = await >>= maybe (return b) (loop . f b)+{-# INLINE foldC #-}++main :: IO ()+main = defaultMain+    [ bench "low level" $ flip whnf upper0 $ \upper ->+        let loop x t+                | x > upper = t+                | otherwise = loop (x + 1) (t + ((x * 2) + 1))+         in loop 1 0+    , bench "completely fused" $ flip whnf upper0 $ \upper ->+        runIdentity+            $ CL.enumFromTo 1 upper+           $$ CL.map (* 2)+           =$ CL.map (+ 1)+           =$ CL.fold (+) 0+    , bench "completely unfused" $ flip whnf upper0 $ \upper ->+        runIdentity+            $ enumFromToC 1 upper+           $$ mapC (* 2)+           =$ mapC (+ 1)+           =$ foldC (+) 0+    , bench "beginning fusion" $ flip whnf upper0 $ \upper ->+        runIdentity+            $ (CL.enumFromTo 1 upper $= CL.map (* 2))+           $$ mapC (+ 1)+           =$ foldC (+) 0+    , bench "middle fusion" $ flip whnf upper0 $ \upper ->+        runIdentity+            $ enumFromToC 1 upper+           $$ (CL.map (* 2) =$= CL.map (+ 1))+           =$ foldC (+) 0+    , bench "ending fusion" $ flip whnf upper0 $ \upper ->+        runIdentity+            $ enumFromToC 1 upper+           $= mapC (* 2)+           $$ (CL.map (+ 1) =$ CL.fold (+) 0)+    , bench "performance of CL.enumFromTo without fusion" $ flip whnf upper0 $ \upper ->+        runIdentity+            $ CL.enumFromTo 1 upper+           $= mapC (* 2)+           $$ (CL.map (+ 1) =$ CL.fold (+) 0)+    ]+  where+    upper0 = 100000 :: Int

@@ -1,12 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}-import           Control.Monad     (replicateM_)-import           Data.ByteString   (ByteString)-import           Data.Conduit-import qualified Data.Conduit.List as CL-import           Data.Conduit.Text (decode, utf8)--src :: Source IO ByteString-src = replicateM_ 1000000 $ yield "Hello World!\n"--main :: IO ()-main = src $$ decode utf8 =$ CL.sinkNull

@@ -1,3 +1,5 @@+__1.2__ Two performance optimizations added. (1) A stream fusion framework. This is a non-breaking change. (2) Codensity transform applied to the `ConduitM` datatype. This only affects users importing the `.Internal` module. Both changes are thoroughly described in the following to blog posts: [Speeding up conduit](https://www.fpcomplete.com/blog/2014/08/iap-speeding-up-conduit), and [conduit stream fusion](https://www.fpcomplete.com/blog/2014/08/conduit-stream-fusion).+ __1.1__ Refactoring into conduit and conduit-extra packages. Core functionality is now in conduit, whereas most common helper modules (including Text, Binary, Zlib, etc) are in conduit-extra. To upgrade to this version, there should only be import list and conduit file changes necessary.  __1.0__ Simplified the user-facing interface back to the Source, Sink, and Conduit types, with Producer and Consumer for generic code. Error messages have been simplified, and optional leftovers and upstream terminators have been removed from the external API. Some long-deprecated functions were finally removed.

@@ -1,5 +1,5 @@ Name:                conduit-Version:             1.1.7+Version:             1.2.0 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>.@@ -19,9 +19,12 @@                        Data.Conduit.List                        Data.Conduit.Internal                        Data.Conduit.Lift+  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+                     , exceptions               >= 0.6                      , lifted-base              >= 0.1                      , transformers-base        >= 0.4.1        && < 0.5                      , monad-control            >= 0.3.1        && < 0.4@@ -57,16 +60,42 @@ --    ghc-options: -threaded --    build-depends: base, directory, doctest >= 0.8 -benchmark utf8-memory-usage+-- benchmark utf8-memory-usage+--     type: exitcode-stdio-1.0+--     hs-source-dirs: benchmarks+--     build-depends:  base+--                   , text-stream-decode+--                   , bytestring+--                   , text+--                   , conduit+--     main-is:        utf8-memory-usage.hs+--     ghc-options:    -Wall -O2 -with-rtsopts=-s++benchmark optimize-201408     type: exitcode-stdio-1.0     hs-source-dirs: benchmarks     build-depends:  base-                  , text-stream-decode-                  , bytestring-                  , text                   , conduit-    main-is:        utf8-memory-usage.hs-    ghc-options:    -Wall -O2 -with-rtsopts=-s+                  , vector+                  , deepseq+                  , containers+                  , transformers+                  , hspec+                  , mwc-random+                  , criterion+                  , kan-extensions+    main-is:        optimize-201408.hs+    ghc-options:    -Wall -O2 -rtsopts++benchmark unfused+    type: exitcode-stdio-1.0+    hs-source-dirs: benchmarks+    build-depends:  base+                  , conduit+                  , criterion+                  , transformers+    main-is:        unfused.hs+    ghc-options:    -Wall -O2 -rtsopts  source-repository head   type:     git

@@ -219,8 +219,8 @@         it "map, left >+>" $ do             x <- runResourceT $                 CI.ConduitM-                    (CI.unConduitM (CL.sourceList [1..10])-                    CI.>+> CI.injectLeftovers (CI.unConduitM $ CL.map (* 2)))+                    ((CI.unConduitM (CL.sourceList [1..10]) CI.Done+                    CI.>+> CI.injectLeftovers (flip CI.unConduitM CI.Done $ CL.map (* 2))) >>=)                     C.$$ CL.fold (+) 0             x `shouldBe` 2 * sum [1..10 :: Int] @@ -459,8 +459,8 @@             ref <- I.newIORef []             let add x = I.modifyIORef ref (x:)                 adder' = CI.NeedInput (\a -> liftIO (add a) >> adder') return-                adder = CI.ConduitM adder'-                residue x = CI.ConduitM $ CI.Leftover (CI.Done ()) x+                adder = CI.ConduitM (adder' >>=)+                residue x = CI.ConduitM $ \rest -> CI.Leftover (rest ()) x              _ <- C.yield 1 C.$$ adder             x <- I.readIORef ref@@ -524,12 +524,12 @@      describe "left/right identity" $ do         it' "left identity" $ do-            x <- CL.sourceList [1..10 :: Int] C.$$ CI.ConduitM CI.idP C.=$ CL.fold (+) 0+            x <- CL.sourceList [1..10 :: Int] C.$$ CI.ConduitM (CI.idP >>=) C.=$ CL.fold (+) 0             y <- CL.sourceList [1..10 :: Int] C.$$ CL.fold (+) 0             x `shouldBe` y         it' "right identity" $ do-            x <- CI.runPipe $ mapM_ CI.yield [1..10 :: Int] CI.>+> (CI.injectLeftovers $ CI.unConduitM $ CL.fold (+) 0) CI.>+> CI.idP-            y <- CI.runPipe $ mapM_ CI.yield [1..10 :: Int] CI.>+> (CI.injectLeftovers $ CI.unConduitM $ CL.fold (+) 0)+            x <- CI.runPipe $ mapM_ CI.yield [1..10 :: Int] CI.>+> (CI.injectLeftovers $ flip CI.unConduitM CI.Done $ CL.fold (+) 0) CI.>+> CI.idP+            y <- CI.runPipe $ mapM_ CI.yield [1..10 :: Int] CI.>+> (CI.injectLeftovers $ flip CI.unConduitM CI.Done $ CL.fold (+) 0)             x `shouldBe` y      describe "generalizing" $ do@@ -561,6 +561,20 @@             res <- CL.iterate (+ 1) (1 :: Int) C.$$ CL.isolate 10 C.=$ CL.fold (+) 0             res `shouldBe` sum [1..10] +    prop "replicate" $ \cnt' -> do+        let cnt = min cnt' 100+        res <- CL.replicate cnt () C.$$ CL.consume+        res `shouldBe` replicate cnt ()++    prop "replicateM" $ \cnt' -> do+        ref <- I.newIORef 0+        let cnt = min cnt' 100+        res <- CL.replicateM cnt (I.modifyIORef ref (+ 1)) C.$$ CL.consume+        res `shouldBe` replicate cnt ()++        ref' <- I.readIORef ref+        ref' `shouldBe` (if cnt' <= 0 then 0 else cnt)+     describe "unwrapResumable" $ do         it' "works" $ do             ref <- I.newIORef (0 :: Int)@@ -635,11 +649,11 @@     describe "injectLeftovers" $ do         it "works" $ do             let src = mapM_ CI.yield [1..10 :: Int]-                conduit = CI.injectLeftovers $ CI.unConduitM $ C.awaitForever $ \i -> do+                conduit = CI.injectLeftovers $ flip CI.unConduitM CI.Done $ C.awaitForever $ \i -> do                     js <- CL.take 2                     mapM_ C.leftover $ reverse js                     C.yield i-            res <- CI.ConduitM (src CI.>+> CI.injectLeftovers conduit) C.$$ CL.consume+            res <- CI.ConduitM ((src CI.>+> CI.injectLeftovers conduit) >>=) C.$$ CL.consume             res `shouldBe` [1..10]     describe "up-upstream finalizers" $ do         it "pipe" $ do