conduit 0.1.1.1 → 0.2.0

raw patch · 15 files changed

+936/−951 lines, 15 filesdep ~hspecPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: hspec

API changes (from Hackage documentation)

Files

• Data/Conduit.hs +302/−287
• Data/Conduit/Binary.hs +107/−68
• Data/Conduit/Lazy.hs +5/−6
• Data/Conduit/List.hs +102/−68
• Data/Conduit/Text.hs +12/−12
• Data/Conduit/Types/Conduit.hs +22/−25
• Data/Conduit/Types/Sink.hs +84/−134
• Data/Conduit/Types/Source.hs +35/−92
• Data/Conduit/Util/Conduit.hs +108/−101
• Data/Conduit/Util/Sink.hs +73/−71
• Data/Conduit/Util/Source.hs +59/−70
• System/PosixFile.hsc +4/−4
• System/Win32File.hsc +4/−4
• conduit.cabal +6/−4
• test/main.hs +13/−5

@@ -7,6 +7,20 @@ -- operators. module Data.Conduit     ( -- * Types+      -- | The three core types to this package are 'Source' (the data+      -- producer), 'Sink' (the data consumer), and 'Conduit' (the data+      -- transformer). For all three types, a result will provide the next+      -- value to be used. For example, the @Open@ constructor includes a new+      -- @Source@ in it. This leads to the main invariant for all conduit code:+      -- these three types may /never/ be reused.  While some specific values+      -- may work fine with reuse, the result is generally unpredictable and+      -- should no be relied upon.+      --+      -- The user-facing API provided by the connect and fuse operators+      -- automatically addresses the low level details of pulling, pushing, and+      -- closing, and there should rarely be need to perform these actions in+      -- user code.+       -- ** Source       module Data.Conduit.Types.Source       -- *** Buffering@@ -32,6 +46,8 @@     , module Data.Conduit.Util.Sink       -- ** Conduit     , module Data.Conduit.Util.Conduit+      -- * Flushing+    , Flush (..)       -- * Convenience re-exports     , ResourceT     , Resource (..)@@ -41,6 +57,8 @@     , ResourceThrow (..)     ) where +import Control.Applicative ((<$>))+import Control.Monad (liftM) import Control.Monad.Trans.Resource import Data.Conduit.Types.Source import Data.Conduit.Util.Source@@ -49,6 +67,8 @@ import Data.Conduit.Types.Conduit import Data.Conduit.Util.Conduit +-- $typeOverview+ infixr 0 $$  -- | The connect operator, which pulls data from a source and pushes to a sink.@@ -57,21 +77,24 @@ -- 1. In the case of a @SinkNoData@ constructor, the source is not opened at -- all, and the output value is returned immediately. ----- 2. The sink returns @Done@, in which case any leftover input is returned via--- @bsourceUnpull@ the source is closed.+-- 2. The sink returns @Done@. If the input was a @BufferedSource@, any+-- leftover input is put in the buffer. For a normal @Source@, the leftover+-- value is discarded, and the source is closed. -- -- 3. The source return @Closed@, in which case the sink is closed. ----- Note that this function will automatically close any 'Source's, but will not--- close any 'BufferedSource's, allowing them to be reused.+-- Note that this function will automatically close any @Source@s, but will not+-- close any @BufferedSource@s, allowing them to be reused. ----- Since 0.0.0+-- Since 0.2.0 ($$) :: (IsSource src, Resource m) => src m a -> Sink a m b -> ResourceT m b ($$) = connect {-# INLINE ($$) #-}  -- | A typeclass allowing us to unify operators for 'Source' and -- 'BufferedSource'.+--+-- Since 0.2.0 class IsSource src where     connect :: Resource m => src m a -> Sink a m b -> ResourceT m b     fuseLeft :: Resource m => src m a -> Conduit a m b -> Source m b@@ -89,38 +112,37 @@     {-# INLINE fuseLeft #-}  normalConnect :: Resource m => Source m a -> Sink a m b -> ResourceT m b-normalConnect (Source msrc) (Sink msink) = do-    sinkI <- msink-    case sinkI of-        SinkNoData output -> return output-        SinkData push close -> do-            src <- msrc-            connect' src push close+normalConnect _ (SinkNoData output) = return output+normalConnect src0 (SinkLift msink) = msink >>= normalConnect src0+normalConnect src0 (SinkData push0 close0) =+    connect' src0 push0 close0   where-    connect' src push close =-        loop-      where-        loop = do-            res <- sourcePull src-            case res of-                Closed -> do-                    res' <- close-                    return res'-                Open a -> do-                    mres <- push a-                    case mres of-                        Done _leftover res' -> do-                            sourceClose src-                            return res'-                        Processing -> loop+    connect' src push close = do+        res <- sourcePull src+        case res of+            Closed -> do+                res' <- close+                return res'+            Open src' a -> do+                mres <- push a+                case mres of+                    Done _leftover res' -> do+                        sourceClose src'+                        return res'+                    Processing push' close' -> connect' src' push' close' -data FuseLeftState a = FLClosed [a] | FLOpen [a]+data FuseLeftState src conduit output =+    FLClosed [output]+  | FLOpen src conduit [output]  infixl 1 $=  -- | Left fuse, combining a source and a conduit together into a new source. ----- Since 0.0.0+-- Note that any @Source@ passed in will be automatically closed, while a+-- @BufferedSource@ will be left open.+--+-- Since 0.2.0 ($=) :: (IsSource src, Resource m)      => src m a      -> Conduit a m b@@ -129,185 +151,182 @@ {-# INLINE ($=) #-}  normalFuseLeft :: Resource m => Source m a -> Conduit a m b -> Source m b-normalFuseLeft (Source msrc) (Conduit mc) = Source $ do-    istate <- newRef $ FLOpen [] -- still open, no buffer-    src <- msrc-    c <- mc-    return $ PreparedSource-        (pull istate src c)-        (close istate src c)+normalFuseLeft src0 conduit0 = Source+    { sourcePull = pull $ FLOpen src0 conduit0 []+    , sourceClose = return ()+    }   where-    pull istate src c = do-        state' <- readRef istate+    mkSrc state = Source (pull state) (close state)+    pull state' =         case state' of             FLClosed [] -> return Closed-            FLClosed (x:xs) -> do-                writeRef istate $ FLClosed xs-                return $ Open x-            FLOpen (x:xs) -> do-                writeRef istate $ FLOpen xs-                return $ Open x-            FLOpen [] -> do+            FLClosed (x:xs) -> return $ Open+                (mkSrc (FLClosed xs))+                x+            FLOpen src conduit (x:xs) -> return $ Open+                (mkSrc (FLOpen src conduit xs))+                x+            FLOpen src conduit [] -> do                 mres <- sourcePull src                 case mres of                     Closed -> do-                        res <- conduitClose c+                        res <- conduitClose conduit                         case res of-                            [] -> do-                                writeRef istate $ FLClosed []-                                return Closed-                            x:xs -> do-                                writeRef istate $ FLClosed xs-                                return $ Open x-                    Open input -> do-                        res' <- conduitPush c input+                            [] -> return Closed+                            x:xs -> return $ Open+                                (mkSrc (FLClosed xs))+                                x+                    Open src'' input -> do+                        res' <- conduitPush conduit input                         case res' of-                            Producing [] -> pull istate src c-                            Producing (x:xs) -> do-                                writeRef istate $ FLOpen xs-                                return $ Open x+                            Producing conduit' [] ->+                                pull $ FLOpen src'' conduit' []+                            Producing conduit' (x:xs) -> return $ Open+                                (mkSrc (FLOpen src'' conduit' xs))+                                x                             Finished _leftover output -> do-                                sourceClose src+                                sourceClose src''                                 case output of-                                    [] -> do-                                        writeRef istate $ FLClosed []-                                        return Closed-                                    x:xs -> do-                                        writeRef istate $ FLClosed xs-                                        return $ Open x-    close istate src c = do+                                    [] -> return Closed+                                    x:xs -> return $ Open+                                        (mkSrc (FLClosed xs))+                                        x+    close state = do         -- See comment on bufferedFuseLeft for why we need to have the         -- following check-        state <- readRef istate         case state of             FLClosed _ -> return ()-            FLOpen _ -> do-                _ignored <- conduitClose c-                sourceClose src+            FLOpen src' (Conduit _ closeC) _ -> do+                _ignored <- closeC+                sourceClose src'  infixr 0 =$  -- | Right fuse, combining a conduit and a sink together into a new sink. ----- Since 0.0.0+-- Since 0.2.0 (=$) :: Resource m => Conduit a m b -> Sink b m c -> Sink a m c-Conduit mc =$ Sink ms = Sink $ do-    s <- ms-    case s of-        SinkData pushI closeI -> mc >>= go pushI closeI-        SinkNoData mres -> return $ SinkNoData mres+_ =$ SinkNoData res = SinkNoData res+conduit =$ SinkLift msink = SinkLift (liftM (conduit =$) msink)+conduitOrig =$ SinkData pushI0 closeI0 = SinkData+    { sinkPush = push pushI0 closeI0 conduitOrig+    , sinkClose = close pushI0 closeI0 conduitOrig+    }   where-    go pushI closeI c = do-        return SinkData-            { sinkPush = \cinput -> do-                res <- conduitPush c cinput-                case res of-                    Producing sinput -> do-                        let push [] = return Processing-                            push (i:is) = do-                                mres <- pushI i-                                case mres of-                                    Processing -> push is-                                    Done _sleftover res' -> do-                                        _ <- conduitClose c-                                        return $ Done Nothing res'-                        push sinput-                    Finished cleftover sinput -> do-                        let push [] = closeI-                            push (i:is) = do-                                mres <- pushI i-                                case mres of-                                    Processing -> push is-                                    Done _sleftover res' -> return res'-                        res' <- push sinput-                        return $ Done cleftover res'-            , sinkClose = do-                sinput <- conduitClose c-                let push [] = closeI-                    push (i:is) = do-                        mres <- pushI i+    push pushI closeI conduit0 cinput = do+        res <- conduitPush conduit0 cinput+        case res of+            Producing conduit' sinput -> do+                let loop p c [] = return (Processing (push p c conduit') (close p c conduit'))+                    loop p _ (i:is) = do+                        mres <- p i                         case mres of-                            Processing -> push is+                            Processing p' c' -> loop p' c' is+                            Done _sleftover res' -> do+                                _ <- conduitClose conduit'+                                return $ Done Nothing res'+                loop pushI closeI sinput+            Finished cleftover sinput -> do+                let loop _ c [] = c+                    loop p _ (i:is) = do+                        mres <- p i+                        case mres of+                            Processing p' c' -> loop p' c' is                             Done _sleftover res' -> return res'-                push sinput-            }+                res' <- loop pushI closeI sinput+                return $ Done cleftover res'+    close pushI closeI conduit = do+        sinput <- conduitClose conduit+        let loop _ c [] = c+            loop p _ (i:is) = do+                mres <- p i+                case mres of+                    Processing p' c' -> loop p' c' is+                    Done _sleftover res' -> return res'+        loop pushI closeI sinput  infixr 0 =$=  -- | Middle fuse, combining two conduits together into a new conduit. ----- Since 0.0.0+-- Since 0.2.0 (=$=) :: Resource m => Conduit a m b -> Conduit b m c -> Conduit a m c-Conduit outerM =$= Conduit innerM = Conduit $ do-    outer <- outerM-    inner <- innerM-    return PreparedConduit-        { conduitPush = \inputO -> do-            res <- conduitPush outer inputO-            case res of-                Producing inputI -> do-                    let push [] front = return $ Producing $ front []-                        push (i:is) front = do-                            resI <- conduitPush inner i-                            case resI of-                                Producing c -> push is (front . (c ++))-                                Finished _leftover c -> do-                                    _ <- conduitClose outer-                                    return $ Finished Nothing $ front c-                    push inputI id-                Finished leftoverO inputI -> do-                    c <- conduitPushClose inner inputI-                    return $ Finished leftoverO c-        , conduitClose = do-            b <- conduitClose outer-            c <- conduitPushClose inner b-            return c-        }+outerOrig =$= innerOrig = Conduit+    (pushF outerOrig innerOrig)+    (closeF outerOrig innerOrig)+  where+    pushF outer0 inner0 inputO = do+        res <- conduitPush outer0 inputO+        case res of+            Producing outer inputI -> do+                let loop inner [] front = return $ Producing+                        (Conduit (pushF outer inner) (closeF outer inner))+                        (front [])+                    loop inner (i:is) front = do+                        resI <- conduitPush inner i+                        case resI of+                            Producing conduit c -> loop+                                conduit+                                is+                                (front . (c ++))+                            Finished _leftover c -> do+                                _ <- conduitClose outer+                                return $ Finished Nothing $ front c+                loop inner0 inputI id+            Finished leftoverO inputI -> do+                c <- conduitPushClose inner0 inputI+                return $ Finished leftoverO c+    closeF outer inner = do+        b <- conduitClose outer+        c <- conduitPushClose inner b+        return c  -- | Push some data to a conduit, then close it if necessary.-conduitPushClose :: Monad m => PreparedConduit a m b -> [a] -> ResourceT m [b]+conduitPushClose :: Monad m => Conduit a m b -> [a] -> ResourceT m [b] conduitPushClose c [] = conduitClose c conduitPushClose c (input:rest) = do     res <- conduitPush c input     case res of         Finished _ b -> return b-        Producing b -> do-            b' <- conduitPushClose c rest+        Producing conduit b -> do+            b' <- conduitPushClose conduit rest             return $ b ++ b' --- | When actually interacting with 'Source's, we usually want to be able to+-- | When actually interacting with @Source@s, we sometimes want to be able to -- buffer the output, in case any intermediate steps return leftover data. A--- 'BufferedSource' allows for such buffering.+-- @BufferedSource@ allows for such buffering. ----- A 'BufferedSource', unlike a 'Source', is resumable, meaning it can be passed to--- multiple 'Sink's without restarting.+-- A @BufferedSource@, unlike a @Source@, is resumable, meaning it can be+-- passed to multiple @Sink@s without restarting. Therefore, a @BufferedSource@+-- relaxes the main invariant of this package: the same value may be used+-- multiple times. ----- Finally, a 'BufferedSource' relaxes one of the invariants of a 'Source':--- pulling after an the source is closed is allowed.+-- The intention of a @BufferedSource@ is to be used internally by an+-- application or library, not to be part of its user-facing API. For example,+-- the Warp webserver uses a @BufferedSource@ internally for parsing the+-- request headers, but then passes a normal @Source@ to the web application+-- for reading the request body. ----- A @BufferedSource@ is also known as a /resumable source/, in that it can be--- called multiple times, and each time will provide new data. One caveat:--- while the types will allow you to use the buffered source in multiple--- threads, there is no guarantee that all @BufferedSource@s will handle this--- correctly.+-- One caveat: while the types will allow you to use the buffered source in+-- multiple threads, there is no guarantee that all @BufferedSource@s will+-- handle this correctly. ----- Since 0.0.0-data BufferedSource m a = BufferedSource-    { bsSource :: PreparedSource m a-    , bsBuffer :: Ref (Base m) (BSState a)-    }+-- Since 0.2.0+data BufferedSource m a = BufferedSource (Ref (Base m) (BSState m a)) -data BSState a = ClosedEmpty | OpenEmpty | ClosedFull a | OpenFull a+data BSState m a =+    ClosedEmpty+  | OpenEmpty (Source m a)+  | ClosedFull a+  | OpenFull (Source m a) a --- | Prepare a 'Source' and initialize a buffer. Note that you should manually--- call 'bsourceClose' when the 'BufferedSource' is no longer in use.+-- | Places the given @Source@ and a buffer into a mutable variable. Note that+-- you should manually call 'bsourceClose' when the 'BufferedSource' is no+-- longer in use. ----- Since 0.0.0+-- Since 0.2.0 bufferSource :: Resource m => Source m a -> ResourceT m (BufferedSource m a)-bufferSource (Source msrc) = do-    src <- msrc-    buf <- newRef OpenEmpty-    return $ BufferedSource src buf+bufferSource src = BufferedSource <$> newRef (OpenEmpty src)  -- | Turn a 'BufferedSource' into a 'Source'. Note that in general this will -- mean your original 'BufferedSource' will be closed. Additionally, all@@ -316,180 +335,176 @@ -- -- Note: @bufferSource@ . @unbufferSource@ is /not/ the identity function. ----- Since 0.0.1+-- Since 0.2.0 unbufferSource :: Resource m                => BufferedSource m a                -> Source m a-unbufferSource (BufferedSource src bufRef) = Source $ do-    buf <- readRef bufRef-    case buf of-        OpenEmpty -> return src-        OpenFull a -> do-            isUsedRef <- newRef False-            return PreparedSource-                { sourcePull = do-                    isUsed <- readRef isUsedRef-                    if isUsed-                        then sourcePull src-                        else do-                            writeRef isUsedRef True-                            return $ Open a+unbufferSource (BufferedSource bs) = Source+    { sourcePull = msrc >>= sourcePull+    , sourceClose = msrc >>= sourceClose+    }+  where+    msrc = do+        buf <- readRef bs+        case buf of+            OpenEmpty src -> return src+            OpenFull src a -> return Source+                { sourcePull = return $ Open src a                 , sourceClose = sourceClose src                 }-        ClosedEmpty -> return PreparedSource-            -- Note: we could put some invariant checking in here if we wanted-            { sourcePull = return Closed-            , sourceClose = return ()-            }-        ClosedFull a -> do-            isUsedRef <- newRef False-            return PreparedSource-                { sourcePull = do-                    isUsed <- readRef isUsedRef-                    if isUsed-                        then return Closed-                        else do-                            writeRef isUsedRef True-                            return $ Open a-                , sourceClose = sourceClose src+            ClosedEmpty -> return Source+                -- Note: we could put some invariant checking in here if we wanted+                { sourcePull = return Closed+                , sourceClose = return ()                 }+            ClosedFull a -> return Source+                { sourcePull = return $ Open+                    (Source (return Closed) (return ()))+                    a+                , sourceClose = return ()+                }  bufferedConnect :: Resource m => BufferedSource m a -> Sink a m b -> ResourceT m b-bufferedConnect bs (Sink msink) = do-    sinkI <- msink-    case sinkI of-        SinkNoData output -> return output-        SinkData push close -> do-            bsState <- readRef $ bsBuffer bs-            case bsState of-                ClosedEmpty -> close-                OpenEmpty -> connect' push close-                ClosedFull a -> do-                    res <- push a-                    case res of-                        Done mleftover res' -> do-                            writeRef (bsBuffer bs) $ maybe ClosedEmpty ClosedFull mleftover-                            return res'-                        Processing -> do-                            writeRef (bsBuffer bs) ClosedEmpty-                            close-                OpenFull a -> push a >>= onRes (connect' push close)-  where-    connect' push close =-        loop-      where-        loop = do-            res <- sourcePull $ bsSource bs+bufferedConnect _ (SinkNoData output) = return output+bufferedConnect bsrc (SinkLift msink) = msink >>= bufferedConnect bsrc+bufferedConnect (BufferedSource bs) (SinkData push0 close0) = do+    bsState <- readRef bs+    case bsState of+        ClosedEmpty -> close0+        OpenEmpty src -> connect' src push0 close0+        ClosedFull a -> do+            res <- push0 a             case res of-                Closed -> do-                    writeRef (bsBuffer bs) ClosedEmpty-                    res' <- close+                Done mleftover res' -> do+                    writeRef bs $ maybe ClosedEmpty ClosedFull mleftover                     return res'-                Open a -> push a >>= onRes loop-    onRes _ (Done mleftover res) = do-        writeRef (bsBuffer bs) (maybe OpenEmpty OpenFull mleftover)+                Processing _ close' -> do+                    writeRef bs ClosedEmpty+                    close'+        OpenFull src a -> push0 a >>= onRes src+  where+    connect' src push close = do+        res <- sourcePull src+        case res of+            Closed -> do+                writeRef bs ClosedEmpty+                res' <- close+                return res'+            Open src' a -> push a >>= onRes src'+    onRes src (Done mleftover res) = do+        writeRef bs $ maybe (OpenEmpty src) (OpenFull src) mleftover         return res-    onRes loop Processing = loop+    onRes src (Processing push close) = connect' src push close  bufferedFuseLeft     :: Resource m     => BufferedSource m a     -> Conduit a m b     -> Source m b-bufferedFuseLeft bsrc (Conduit mc) = Source $ do-    istate <- newRef $ FLOpen [] -- still open, no buffer-    c <- mc-    return $ PreparedSource-        (pull istate c)-        (close istate c)+bufferedFuseLeft bsrc conduit0 = Source+    { sourcePull = pullF $ FLOpen () conduit0 [] -- still open, no buffer+    , sourceClose = return ()+    }   where-    pull istate c = do-        state' <- readRef istate+    mkSrc state = Source+        (pullF state)+        (closeF state)+    pullF state' =         case state' of             FLClosed [] -> return Closed-            FLClosed (x:xs) -> do-                writeRef istate $ FLClosed xs-                return $ Open x-            FLOpen (x:xs) -> do-                writeRef istate $ FLOpen xs-                return $ Open x-            FLOpen [] -> do+            FLClosed (x:xs) -> return $ Open+                (mkSrc (FLClosed xs))+                x+            FLOpen () conduit (x:xs) -> return $ Open+                (mkSrc (FLOpen () conduit xs))+                x+            FLOpen () conduit [] -> do                 mres <- bsourcePull bsrc                 case mres of-                    Closed -> do-                        res <- conduitClose c+                    Nothing -> do+                        res <- conduitClose conduit                         case res of-                            [] -> do-                                writeRef istate $ FLClosed []-                                return Closed-                            x:xs -> do-                                writeRef istate $ FLClosed xs-                                return $ Open x-                    Open input -> do-                        res' <- conduitPush c input+                            [] -> return Closed+                            x:xs -> return $ Open+                                (mkSrc (FLClosed xs))+                                x+                    Just input -> do+                        res' <- conduitPush conduit input                         case res' of-                            Producing [] -> pull istate c-                            Producing (x:xs) -> do-                                writeRef istate $ FLOpen xs-                                return $ Open x+                            Producing conduit' [] ->+                                pullF (FLOpen () conduit' [])+                            Producing conduit' (x:xs) -> return $ Open+                                (mkSrc (FLOpen () conduit' xs))+                                x                             Finished leftover output -> do                                 bsourceUnpull bsrc leftover                                 case output of-                                    [] -> do-                                        writeRef istate $ FLClosed []-                                        return Closed-                                    x:xs -> do-                                        writeRef istate $ FLClosed xs-                                        return $ Open x-    close istate c = do+                                    [] -> return Closed+                                    x:xs -> return $ Open+                                        (mkSrc (FLClosed xs))+                                        x+    closeF state = do         -- Normally we don't have to worry about double closing, as the         -- invariant of a source is that close is never called twice. However,         -- here, if the Conduit returned Finished with some data, the overall         -- Source will return an Open while the Conduit will be Closed.         -- Therefore, we have to do a check.-        state <- readRef istate         case state of             FLClosed _ -> return ()-            FLOpen _ -> do-                _ignored <- conduitClose c+            FLOpen () (Conduit _ close) _ -> do+                _ignored <- close                 return () -bsourcePull :: Resource m => BufferedSource m a -> ResourceT m (SourceResult a)-bsourcePull (BufferedSource src bufRef) = do-    buf <- readRef bufRef+bsourcePull :: Resource m => BufferedSource m a -> ResourceT m (Maybe a)+bsourcePull (BufferedSource bs) = do+    buf <- readRef bs     case buf of-        OpenEmpty -> do+        OpenEmpty src -> do             res <- sourcePull src             case res of-                Open _ -> return res-                Closed -> writeRef bufRef ClosedEmpty >> return Closed-        ClosedEmpty -> return Closed-        OpenFull a -> do-            writeRef bufRef OpenEmpty-            return $ Open a+                Open src' a -> do+                    writeRef bs $ OpenEmpty src'+                    return $ Just a+                Closed -> writeRef bs ClosedEmpty >> return Nothing+        ClosedEmpty -> return Nothing+        OpenFull src a -> do+            writeRef bs (OpenEmpty src)+            return $ Just a         ClosedFull a -> do-            writeRef bufRef ClosedEmpty-            return $ Open a+            writeRef bs ClosedEmpty+            return $ Just a  bsourceUnpull :: Resource m => BufferedSource m a -> Maybe a -> ResourceT m () bsourceUnpull _ Nothing = return ()-bsourceUnpull (BufferedSource _ bufRef) (Just a) = do-    buf <- readRef bufRef+bsourceUnpull (BufferedSource ref) (Just a) = do+    buf <- readRef ref     case buf of-        OpenEmpty -> writeRef bufRef $ OpenFull a-        ClosedEmpty -> writeRef bufRef $ ClosedFull a+        OpenEmpty src -> writeRef ref (OpenFull src a)+        ClosedEmpty -> writeRef ref (ClosedFull a)         _ -> error $ "Invariant violated: bsourceUnpull called on full data" --- | Close the underlying 'PreparedSource' for the given 'BufferedSource'. Note+-- | Close the underlying 'Source' for the given 'BufferedSource'. Note -- that this function can safely be called multiple times, as it will first--- check if the 'PreparedSource' was previously closed.+-- check if the 'Source' was previously closed. ----- Since 0.0.0+-- Since 0.2.0 bsourceClose :: Resource m => BufferedSource m a -> ResourceT m ()-bsourceClose (BufferedSource src bufRef) = do-    buf <- readRef bufRef+bsourceClose (BufferedSource ref) = do+    buf <- readRef ref     case buf of-        OpenEmpty -> sourceClose src-        OpenFull _ -> sourceClose src+        OpenEmpty src -> sourceClose src+        OpenFull src _ -> sourceClose src         ClosedEmpty -> return ()         ClosedFull _ -> 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.2.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)

@@ -16,6 +16,7 @@     , takeWhile     , dropWhile     , take+    , Data.Conduit.Binary.lines     ) where  import Prelude hiding (head, take, takeWhile, dropWhile)@@ -27,9 +28,7 @@ import Control.Monad (liftM) import Control.Monad.IO.Class (liftIO) import qualified System.IO as IO-import Control.Monad.Trans.Resource-    ( withIO, release, newRef, readRef, writeRef-    )+import Control.Monad.Trans.Resource (withIO, release) import Data.Word (Word8) #if CABAL_OS_WINDOWS import qualified System.Win32File as F@@ -43,7 +42,7 @@ -- While you are not required to call @hClose@ on the resulting handle, you -- should do so as early as possible to free scarce resources. ----- Since 0.0.2+-- Since 0.2.0 openFile :: ResourceIO m          => FilePath          -> IO.IOMode@@ -52,7 +51,7 @@  -- | Stream the contents of a file as binary data. ----- Since 0.0.0+-- Since 0.2.0 sourceFile :: ResourceIO m            => FilePath            -> Source m S.ByteString@@ -69,25 +68,29 @@ -- function will /not/ automatically close the @Handle@ when processing -- completes, since it did not acquire the @Handle@ in the first place. ----- Since 0.0.2.+-- Since 0.2.0 sourceHandle :: ResourceIO m              => IO.Handle              -> Source m S.ByteString-sourceHandle h = Source $ return $ PreparedSource-    { sourcePull = do+sourceHandle h =+    src+  where+    src = Source pull close++    pull = do         bs <- liftIO (S.hGetSome h 4096)         if S.null bs             then return Closed-            else return (Open bs)-    , sourceClose = return ()-    }+            else return $ Open src bs +    close = return ()+ -- | An alternative to 'sourceHandle'. -- Instead of taking a pre-opened 'IO.Handle', it takes an action that opens -- a 'IO.Handle' (in read mode), so that it can open it only when needed -- and close it as soon as possible. ----- Since 0.1.1+-- Since 0.2.0 sourceIOHandle :: ResourceIO m                => IO IO.Handle                -> Source m S.ByteString@@ -95,58 +98,55 @@     (\handle -> do         bs <- liftIO (S.hGetSome handle 4096)         if S.null bs-            then return Closed-            else return $ Open bs)+            then return IOClosed+            else return $ IOOpen bs)  -- | Stream all incoming data to the given 'IO.Handle'. Note that this function -- will /not/ automatically close the @Handle@ when processing completes. ----- Since 0.0.2.+-- Since 0.2.0 sinkHandle :: ResourceIO m            => IO.Handle            -> Sink S.ByteString m ()-sinkHandle h = Sink $ return $ SinkData-    { sinkPush = \input -> liftIO (S.hPut h input) >> return Processing-    , sinkClose = return ()-    }+sinkHandle h =+    SinkData push close+  where+    push input = liftIO (S.hPut h input) >> return (Processing push close)+    close = return ()  -- | An alternative to 'sinkHandle'. -- Instead of taking a pre-opened 'IO.Handle', it takes an action that opens -- a 'IO.Handle' (in write mode), so that it can open it only when needed -- and close it as soon as possible. ----- Since 0.1.1+-- Since 0.2.0 sinkIOHandle :: ResourceIO m              => IO IO.Handle              -> Sink S.ByteString m () sinkIOHandle alloc = sinkIO alloc IO.hClose-    (\handle bs -> liftIO (S.hPut handle bs) >> return Processing)+    (\handle bs -> liftIO (S.hPut handle bs) >> return IOProcessing)     (const $ return ())  -- | Stream the contents of a file as binary data, starting from a certain -- offset and only consuming up to a certain number of bytes. ----- Since 0.0.0+-- Since 0.2.0 sourceFileRange :: ResourceIO m                 => FilePath                 -> Maybe Integer -- ^ Offset                 -> Maybe Integer -- ^ Maximum count                 -> Source m S.ByteString-sourceFileRange fp offset count = Source $ do-    (key, handle) <- withIO (IO.openBinaryFile fp IO.ReadMode) IO.hClose-    case offset of-        Nothing -> return ()-        Just off -> liftIO $ IO.hSeek handle IO.AbsoluteSeek off-    pull <-+sourceFileRange fp offset count = Source+    { sourcePull = do+        (key, handle) <- withIO (IO.openBinaryFile fp IO.ReadMode) IO.hClose+        case offset of+            Nothing -> return ()+            Just off -> liftIO $ IO.hSeek handle IO.AbsoluteSeek off         case count of-            Nothing -> return $ pullUnlimited handle key-            Just c -> do-                ic <- newRef c-                return $ pullLimited ic handle key-    return PreparedSource-        { sourcePull = pull-        , sourceClose = release key-        }+            Nothing -> pullUnlimited handle key+            Just c -> pullLimited c handle key+    , sourceClose = return ()+    }   where     pullUnlimited handle key = do         bs <- liftIO $ S.hGetSome handle 4096@@ -154,9 +154,15 @@             then do                 release key                 return Closed-            else return $ Open bs-    pullLimited ic handle key = do-        c <- fmap fromInteger $ readRef ic+            else do+                let src = Source+                        { sourcePull = pullUnlimited handle key+                        , sourceClose = release key+                        }+                return $ Open src bs++    pullLimited c0 handle key = do+        let c = fromInteger c0         bs <- liftIO $ S.hGetSome handle (min c 4096)         let c' = c - S.length bs         assert (c' >= 0) $@@ -165,12 +171,15 @@                     release key                     return Closed                 else do-                    writeRef ic $ toInteger c'-                    return $ Open bs+                    let src = Source+                            { sourcePull = pullLimited (toInteger c') handle key+                            , sourceClose = release key+                            }+                    return $ Open src bs  -- | Stream all incoming data to the given file. ----- Since 0.0.0+-- Since 0.2.0 sinkFile :: ResourceIO m          => FilePath          -> Sink S.ByteString m ()@@ -179,7 +188,7 @@ -- | Stream the contents of the input to a file, and also send it along the -- pipeline. Similar in concept to the Unix command @tee@. ----- Since 0.0.0+-- Since 0.2.0 conduitFile :: ResourceIO m             => FilePath             -> Conduit S.ByteString m S.ByteString@@ -188,14 +197,14 @@     IO.hClose     (\handle bs -> do         liftIO $ S.hPut handle bs-        return $ Producing [bs])+        return $ IOProducing [bs])     (const $ return [])  -- | Ensure that only up to the given number of bytes are consume by the inner -- sink. Note that this does /not/ ensure that all of those bytes are in fact -- consumed. ----- Since 0.0.0+-- Since 0.2.0 isolate :: Resource m         => Int         -> Conduit S.ByteString m S.ByteString@@ -204,60 +213,90 @@     push     close   where-    push 0 bs = return (0, Finished (Just bs) [])+    push 0 bs = return $ StateFinished (Just bs) []     push count bs = do         let (a, b) = S.splitAt count bs         let count' = count - S.length a-        return (count',+        return $             if count' == 0-                then Finished (if S.null b then Nothing else Just b) (if S.null a then [] else [a])-                else assert (S.null b) $ Producing [a])+                then StateFinished (if S.null b then Nothing else Just b) (if S.null a then [] else [a])+                else assert (S.null b) $ StateProducing count' [a]     close _ = return []  -- | Return the next byte from the stream, if available. ----- Since 0.0.2+-- Since 0.2.0 head :: Resource m => Sink S.ByteString m (Maybe Word8)-head = Sink $ return $ SinkData-    { sinkPush = \bs ->+head =+    SinkData push close+  where+    push bs =         case S.uncons bs of-            Nothing -> return Processing+            Nothing -> return $ Processing push close             Just (w, bs') -> do                 let lo = if S.null bs' then Nothing else Just bs'                 return $ Done lo (Just w)-    , sinkClose = return Nothing-    }+    close = return Nothing  -- | Return all bytes while the predicate returns @True@. ----- Since 0.0.2+-- Since 0.2.0 takeWhile :: Resource m => (Word8 -> Bool) -> Conduit S.ByteString m S.ByteString-takeWhile p = Conduit $ return $ PreparedConduit-    { conduitPush = \bs -> do+takeWhile p =+    conduit+  where+    conduit = Conduit push close+    push bs = do         let (x, y) = S.span p bs         return $             if S.null y-                then Producing [x]+                then Producing conduit [x]                 else Finished (Just y) (if S.null x then [] else [x])-    , conduitClose = return []-    }+    close = return []  -- | Ignore all bytes while the predicate returns @True@. ----- Since 0.0.2+-- Since 0.2.0 dropWhile :: Resource m => (Word8 -> Bool) -> Sink S.ByteString m ()-dropWhile p = Sink $ return $ SinkData-    { sinkPush = \bs -> do+dropWhile p =+    SinkData push close+  where+    push bs = do         let bs' = S.dropWhile p bs         return $             if S.null bs'-                then Processing+                then Processing push close                 else Done (Just bs') ()-    , sinkClose = return ()-    }+    close = return ()  -- | Take the given number of bytes, if available. ----- Since 0.0.3+-- Since 0.2.0 take :: Resource m => Int -> Sink S.ByteString m L.ByteString take n = L.fromChunks `liftM` (isolate n =$ CL.consume)++-- | Split the input bytes into lines. In other words, split on the LF byte+-- (10), and strip it from the output.+--+-- Since 0.2.0+lines :: Resource m => Conduit S.ByteString m S.ByteString+lines =+    conduitState id push close+  where+    push front bs' = return $ StateProducing leftover ls+      where+        bs = front bs'+        (leftover, ls) = getLines id bs++    getLines front bs+        | S.null bs = (id, front [])+        | S.null y = (S.append x, front [])+        | otherwise = getLines (front . (x:)) (S.drop 1 y)+      where+        (x, y) = S.breakByte 10 bs++    close front+        | S.null bs = return []+        | otherwise = return [bs]+      where+        bs = front S.empty

@@ -12,17 +12,16 @@  -- | Use lazy I\/O to consume all elements from a @Source@. ----- Since 0.0.0+-- Since 0.2.0 lazyConsume :: MonadBaseControl IO m => Source m a -> ResourceT m [a]-lazyConsume (Source msrc) = do-    src <- msrc-    go src+lazyConsume src0 = do+    go src0   where      go src = liftBaseOp_ unsafeInterleaveIO $ do         res <- sourcePull src         case res of             Closed -> return []-            Open x -> do-                y <- go src+            Open src' x -> do+                y <- go src'                 return $ x : y

@@ -28,12 +28,14 @@       -- ** Pure     , map     , concatMap+    , concatMapAccum     , groupBy     , isolate     , filter       -- ** Monadic     , mapM     , concatMapM+    , concatMapAccumM     ) where  import Prelude@@ -49,19 +51,19 @@  -- | A strict left fold. ----- Since 0.0.0+-- Since 0.2.0 fold :: Resource m      => (b -> a -> b)      -> b      -> Sink a m b fold f accum0 = sinkState     accum0-    (\accum input -> return (f accum input, Processing))+    (\accum input -> return (StateProcessing $ f accum input))     return  -- | A monadic strict left fold. ----- Since 0.0.0+-- Since 0.2.0 foldM :: Resource m       => (b -> a -> m b)       -> b@@ -70,29 +72,31 @@     accum0     (\accum input -> do         accum' <- lift $ f accum input-        return (accum', Processing)+        return $ StateProcessing accum'     )     return  -- | Apply the action to all values in the stream. ----- Since 0.0.0+-- Since 0.2.0 mapM_ :: Resource m       => (a -> m ())       -> Sink a m ()-mapM_ f = Sink $ return $ SinkData-    (\input -> lift (f input) >> return Processing)-    (return ())+mapM_ f =+    SinkData push close+  where+    push input = lift (f input) >> return (Processing push close)+    close = return ()  -- | Convert a list into a source. ----- Since 0.0.0+-- Since 0.2.0 sourceList :: Resource m => [a] -> Source m a sourceList l0 =     sourceState l0 go   where-    go [] = return ([], Closed)-    go (x:xs) = return (xs, Open x)+    go [] = return StateClosed+    go (x:xs) = return $ StateOpen xs x  -- | Ignore a certain number of values in the stream. This function is -- semantically equivalent to:@@ -102,7 +106,7 @@ -- However, @drop@ is more efficient as it does not need to hold values in -- memory. ----- Since 0.0.0+-- Since 0.2.0 drop :: Resource m      => Int      -> Sink a m ()@@ -111,12 +115,12 @@     push     close   where-    push 0 x = return (0, Done (Just x) ())+    push 0 x = return $ StateDone (Just x) ()     push count _ = do         let count' = count - 1-        return (count', if count' == 0-                            then Done Nothing ()-                            else Processing)+        return $ if count' == 0+            then StateDone Nothing ()+            else StateProcessing count'     close _ = return ()  -- | Take some values from the stream and return as a list. If you want to@@ -125,7 +129,7 @@ -- -- > take i = isolate i =$ consume ----- Since 0.0.0+-- Since 0.2.0 take :: Resource m      => Int      -> Sink a m [a]@@ -134,22 +138,21 @@     push     close   where-    push (0, front) x = return ((0, front), Done (Just x) (front []))+    push (0, front) x = return (StateDone (Just x) (front []))     push (count, front) x = do         let count' = count - 1             front' = front . (x:)-            res = if count' == 0-                    then Done Nothing (front' [])-                    else Processing-        return ((count', front'), res)+        return $ if count' == 0+                    then StateDone Nothing (front' [])+                    else StateProcessing (count', front')     close (_, front) = return $ front []  -- | Take a single value from the stream, if available. ----- Since 0.0.0+-- Since 0.2.0 head :: Resource m => Sink a m (Maybe a) head =-    Sink $ return $ SinkData push close+    SinkData push close   where     push x = return $ Done Nothing (Just x)     close = return Nothing@@ -157,81 +160,109 @@ -- | Look at the next value in the stream, if available. This function will not -- change the state of the stream. ----- Since 0.0.0+-- Since 0.2.0 peek :: Resource m => Sink a m (Maybe a) peek =-    Sink $ return $ SinkData push close+    SinkData push close   where     push x = return $ Done (Just x) (Just x)     close = return Nothing  -- | Apply a transformation to all values in a stream. ----- Since 0.0.0+-- Since 0.2.0 map :: Monad m => (a -> b) -> Conduit a m b-map f = Conduit $ return $ PreparedConduit-    { conduitPush = return . Producing . return . f-    , conduitClose = return []-    }+map f =+    conduit+  where+    conduit = Conduit push close+    push = return . Producing conduit . return . f+    close = return []  -- | Apply a monadic transformation to all values in a stream. -- -- If you do not need the transformed values, and instead just want the monadic -- side-effects of running the action, see 'mapM_'. ----- Since 0.0.0+-- Since 0.2.0 mapM :: Monad m => (a -> m b) -> Conduit a m b-mapM f = Conduit $ return $ PreparedConduit-    { conduitPush = fmap (Producing . return) . lift . f-    , conduitClose = return []-    }+mapM f =+    conduit+  where+    conduit = Conduit push close+    push = fmap (Producing conduit . return) . lift . f+    close = return []  -- | Apply a transformation to all values in a stream, concatenating the output -- values. ----- Since 0.0.0+-- Since 0.2.0 concatMap :: Monad m => (a -> [b]) -> Conduit a m b-concatMap f = Conduit $ return $ PreparedConduit-    { conduitPush = return . Producing . f-    , conduitClose = return []-    }+concatMap f =+    conduit+  where+    conduit = Conduit push close+    push = return . Producing conduit . f+    close = return []  -- | Apply a monadic transformation to all values in a stream, concatenating -- the output values. ----- Since 0.0.0+-- Since 0.2.0 concatMapM :: Monad m => (a -> m [b]) -> Conduit a m b-concatMapM f = Conduit $ return $ PreparedConduit-    { conduitPush = fmap Producing . lift . f-    , conduitClose = return []-    }+concatMapM f =+    conduit+  where+    conduit = Conduit push close+    push = fmap (Producing conduit) . lift . f+    close = return [] +-- | 'concatMap' with accumerator.+--+-- Since 0.2.0+concatMapAccum :: Resource m => (a -> accum -> (accum, [b])) -> accum -> Conduit a m b+concatMapAccum f accum = conduitState accum push close+  where+    push state input = let (state', result) = f input state+                       in return $ StateProducing state' result+    close _ = return []++-- | 'concatMapM' with accumerator.+--+-- Since 0.2.0+concatMapAccumM :: Resource m => (a -> accum -> m (accum, [b])) -> accum -> Conduit a m b+concatMapAccumM f accum = conduitState accum push close+  where+    push state input = do (state', result) <- lift (f input state)+                          return $ StateProducing state' result+    close _ = return []+ -- | Consume all values from the stream and return as a list. Note that this -- will pull all values into memory. For a lazy variant, see -- "Data.Conduit.Lazy". ----- Since 0.0.0+-- Since 0.2.0 consume :: Resource m => Sink a m [a] consume = sinkState     id-    (\front input -> return (front . (input :), Processing))+    (\front input -> return (StateProcessing $ front . (input :)))     (\front -> return $ front [])  -- | Grouping input according to an equality function. ----- Since 0.0.2+-- Since 0.2.0 groupBy :: Resource m => (a -> a -> Bool) -> Conduit a m [a] groupBy f = conduitState     []     push     close   where-    push []      v = return ([v], Producing [])+    push []      v = return $ StateProducing [v] []     push s@(x:_) v =       if f x v then-        return (v:s, Producing [])+        return $ StateProducing (v:s) []       else-        return ([v], Producing [s])+        return $ StateProducing [v] [s]     close s = return [s]  -- | Ensure that the inner sink consumes no more than the given number of@@ -246,7 +277,7 @@ -- >     someOtherSink -- >     ... ----- Since 0.0.0+-- Since 0.2.0 isolate :: Resource m => Int -> Conduit a m a isolate count0 = conduitState     count0@@ -256,35 +287,38 @@     close _ = return []     push count x = do         if count == 0-            then return (count, Finished (Just x) [])+            then return $ StateFinished (Just x) []             else do                 let count' = count - 1-                return (count',-                    if count' == 0-                        then Finished Nothing [x]-                        else Producing [x])+                return $ if count' == 0+                    then StateFinished Nothing [x]+                    else StateProducing count' [x]  -- | Keep only values in the stream passing a given predicate. ----- Since 0.0.0+-- Since 0.2.0 filter :: Resource m => (a -> Bool) -> Conduit a m a-filter f = Conduit $ return $ PreparedConduit-    { conduitPush = return . Producing . Prelude.filter f . return-    , conduitClose = return []-    }+filter f =+    conduit+  where+    conduit = Conduit push close+    push = return . Producing conduit . Prelude.filter f . return+    close = return []  -- | Ignore the remainder of values in the source. Particularly useful when -- combined with 'isolate'. ----- Since 0.0.0+-- Since 0.2.0 sinkNull :: Resource m => Sink a m ()-sinkNull = Sink $ return $ SinkData-    (\_ -> return Processing)-    (return ())+sinkNull =+    SinkData push close+  where+    push _ = return $ Processing push close+    close = return ()  -- | A source that returns nothing. Note that this is just a type-restricted -- synonym for 'mempty'. ----- Since 0.0.4+-- Since 0.2.0 sourceNull :: Resource m => Source m a sourceNull = mempty

@@ -47,7 +47,7 @@  -- | A specific character encoding. ----- Since 0.0.0+-- Since 0.2.0 data Codec = Codec     { codecName :: T.Text     , codecEncode@@ -67,7 +67,7 @@ -- | Convert text into bytes, using the provided codec. If the codec is -- not capable of representing an input character, an exception will be thrown. ----- Since 0.0.0+-- Since 0.2.0 encode :: ResourceThrow m => Codec -> C.Conduit T.Text m B.ByteString encode codec = CL.mapM $ \t -> do     let (bs, mexc) = codecEncode codec t@@ -77,7 +77,7 @@ -- | Convert bytes into text, using the provided codec. If the codec is -- not capable of decoding an input byte sequence, an exception will be thrown. ----- Since 0.0.0+-- Since 0.2.0 decode :: ResourceThrow m => Codec -> C.Conduit B.ByteString m T.Text decode codec = C.conduitState     Nothing@@ -86,7 +86,7 @@   where     push mb input = do         (mb', ts) <- go' mb input-        return $ (mb', C.Producing ts)+        return $ C.StateProducing mb' ts     close mb =         case mb of             Nothing -> return []@@ -115,7 +115,7 @@         front' = front . (text:)  -- |--- Since 0.0.0+-- Since 0.2.0 data TextException = DecodeException Codec Word8                    | EncodeException Codec Char     deriving (Show, Typeable)@@ -148,7 +148,7 @@             Right _ -> Right B.empty)  -- |--- Since 0.0.0+-- Since 0.2.0 utf8 :: Codec utf8 = Codec name enc dec where     name = T.pack "UTF-8"@@ -181,7 +181,7 @@                     else decodeMore  -- |--- Since 0.0.0+-- Since 0.2.0 utf16_le :: Codec utf16_le = Codec name enc dec where     name = T.pack "UTF-16-LE"@@ -208,7 +208,7 @@         decodeAll = (TE.decodeUtf16LE bytes, B.empty)  -- |--- Since 0.0.0+-- Since 0.2.0 utf16_be :: Codec utf16_be = Codec name enc dec where     name = T.pack "UTF-16-BE"@@ -243,7 +243,7 @@     x = (fromIntegral x1 `shiftL` 8) .|. fromIntegral x0  -- |--- Since 0.0.0+-- Since 0.2.0 utf32_le :: Codec utf32_le = Codec name enc dec where     name = T.pack "UTF-32-LE"@@ -253,7 +253,7 @@         Nothing -> splitSlowly TE.decodeUtf32LE bs  -- |--- Since 0.0.0+-- Since 0.2.0 utf32_be :: Codec utf32_be = Codec name enc dec where     name = T.pack "UTF-32-BE"@@ -276,7 +276,7 @@         else B.splitAt lenToDecode bytes  -- |--- Since 0.0.0+-- Since 0.2.0 ascii :: Codec ascii = Codec name enc dec where     name = T.pack "ASCII"@@ -295,7 +295,7 @@             else Left (DecodeException ascii (B.head unsafe), unsafe)  -- |--- Since 0.0.0+-- Since 0.2.0 iso8859_1 :: Codec iso8859_1 = Codec name enc dec where     name = T.pack "ISO-8859-1"

@@ -2,51 +2,48 @@ -- is almost always connected either left (to a source) or right (to a sink). module Data.Conduit.Types.Conduit     ( ConduitResult (..)-    , PreparedConduit (..)     , Conduit (..)+    , ConduitPush+    , ConduitClose     ) where  import Control.Monad.Trans.Resource (ResourceT) import Control.Monad (liftM) +-- | The value of the @conduitPush@ record.+type ConduitPush input m output = input -> ResourceT m (ConduitResult input m output)++-- | The value of the @conduitClose@ record.+type ConduitClose m output = ResourceT m [output]+ -- | When data is pushed to a @Conduit@, it may either indicate that it is -- still producing output and provide some, or indicate that it is finished -- producing output, in which case it returns optional leftover input and some -- final output. ----- Since 0.0.0-data ConduitResult input output = Producing [output] | Finished (Maybe input) [output]+-- The @Producing@ constructor provides a new @Conduit@ to be used in place of+-- the previous one.+--+-- Since 0.2.0+data ConduitResult input m output =+    Producing (Conduit input m output) [output]+  | Finished (Maybe input) [output] -instance Functor (ConduitResult input) where-    fmap f (Producing o) = Producing (fmap f o)+instance Monad m => Functor (ConduitResult input m) where+    fmap f (Producing c o) = Producing (fmap f c) (fmap f o)     fmap f (Finished i o) = Finished i (fmap f o)  -- | A conduit has two operations: it can receive new input (a push), and can -- be closed. ----- Invariants:------ * Neither a push nor close may be performed after a conduit returns a--- 'Finished' from a push, or after a close is performed.------ Since 0.0.0-data PreparedConduit input m output = PreparedConduit-    { conduitPush :: input -> ResourceT m (ConduitResult input output)-    , conduitClose :: ResourceT m [output]+-- Since 0.2.0+data Conduit input m output = Conduit+    { conduitPush :: ConduitPush input m output+    , conduitClose :: ConduitClose m output     } -instance Monad m => Functor (PreparedConduit input m) where+instance Monad m => Functor (Conduit input m) where     fmap f c = c         { conduitPush = liftM (fmap f) . conduitPush c         , conduitClose = liftM (fmap f) (conduitClose c)         }---- | A monadic action generating a 'PreparedConduit'. See @Source@ and @Sink@--- for more motivation.------ Since 0.0.0-newtype Conduit input m output =-    Conduit { prepareConduit :: ResourceT m (PreparedConduit input m output) }--instance Monad m => Functor (Conduit input m) where-    fmap f (Conduit mc) = Conduit (liftM (fmap f) mc)

@@ -6,28 +6,66 @@ -- | Defines the types for a sink, which is a consumer of data. module Data.Conduit.Types.Sink     ( SinkResult (..)-    , PreparedSink (..)     , Sink (..)+    , SinkPush+    , SinkClose     ) where  import Control.Monad.Trans.Resource import Control.Monad.Trans.Class (MonadTrans (lift)) import Control.Monad.IO.Class (MonadIO (liftIO))-import Control.Monad (liftM)+import Control.Monad (liftM, ap) import Control.Applicative (Applicative (..)) import Control.Monad.Base (MonadBase (liftBase)) +-- | The value of the @sinkPush@ record.+type SinkPush input m output = input -> ResourceT m (SinkResult input m output)++-- | The value of the @sinkClose@ record.+type SinkClose m output = ResourceT m output+ -- | A @Sink@ ultimately returns a single output value. Each time data is -- pushed to it, a @Sink@ may indicate that it is still processing data, or -- that it is done, in which case it returns some optional leftover input and -- an output value. ----- Since 0.0.0-data SinkResult input output = Processing | Done (Maybe input) output-instance Functor (SinkResult input) where-    fmap _ Processing = Processing+-- The @Processing@ constructors provides updated push and close functions to+-- be used in place of the original @Sink@.+--+-- Since 0.2.0+data SinkResult input m output =+    Processing (SinkPush input m output) (SinkClose m output)+  | Done (Maybe input) output+instance Monad m => Functor (SinkResult input m) where+    fmap f (Processing push close) = Processing ((fmap . fmap . fmap) f push) (fmap f close)     fmap f (Done input output) = Done input (f output) +{-+Note to my future self, and anyone else who reads my code: It's tempting to+change `Sink` to look like:++    newtype Sink input m output = Sink { runSink :: ResourceT m (SinkResult input m output) }++If you start implementing this, eventually you'll realize that you will have to+enforce an invariant to make it all work: a `SinkResult` can't return leftovers+unless data was pushed to it.++The idea is that, with the actual definition of `Sink`, it's impossible to get+a `SinkResult` without first pushing in some input. Therefore, it's always+valid at the type level to return leftovers. In this simplified `Sink`, it+would be possible to have code that looks like:++    sink1 = Sink $ return $ Done (Just "foo") ()+    fsink2 () = Sink $ return $ Done (Just "bar") ()+    sink1 >>= fsink2++Now we'd have to coalesce "foo" and "bar" together (e.g., require `Monoid`),+throw away data, or throw an exception.++So the current three-constructor approach to `Sink` may not be as pretty, but+it enforce the invariants much better.+-}+ -- | In general, a sink will consume data and eventually produce an output when -- it has consumed \"enough\" data. There are two caveats to that statement: --@@ -44,155 +82,67 @@ -- cannot always produce output, this should be indicated in its return value, -- using something like a 'Maybe' or 'Either'. ----- Invariants:------ * After a 'PreparedSink' produces a result (either via 'sinkPush' or--- 'sinkClose'), neither of those two functions may be called on the @Sink@--- again.------ * If a @Sink@ needs to clean up any resources (e.g., close a file handle),--- it must do so whenever it returns a result, either via @sinkPush@ or--- @sinkClose@. Note that, due to usage of @ResourceT@, this is merely an--- optimization.+-- A @Sink@ should clean up any resources it has allocated when it returns a+-- value, whether that be via @sinkPush@ or @sinkClose@. ----- Since 0.0.0-data PreparedSink input m output =+-- Since 0.2.0+data Sink input m output =     SinkNoData output   | SinkData-        { sinkPush :: input -> ResourceT m (SinkResult input output)-        , sinkClose :: ResourceT m output+        { sinkPush :: SinkPush input m output+        , sinkClose :: SinkClose m output         }+  -- | This constructor is provided to allow us to create an efficient+  -- @MonadTrans@ instance.+  | SinkLift (ResourceT m (Sink input m output)) -instance Monad m => Functor (PreparedSink input m) where+instance Monad m => Functor (Sink input m) where     fmap f (SinkNoData x) = SinkNoData (f x)     fmap f (SinkData p c) = SinkData         { sinkPush = liftM (fmap f) . p         , sinkClose = liftM f c         }---- | Most 'PreparedSink's require some type of state, similar to--- 'PreparedSource's. Like a @Source@ for a @PreparedSource@, a @Sink@ is a--- simple monadic wrapper around a @PreparedSink@ which allows initialization--- of such state. See @Source@ for further caveats.------ Note that this type provides a 'Monad' instance, allowing you to easily--- compose @Sink@s together.------ Since 0.0.0-newtype Sink input m output = Sink { prepareSink :: ResourceT m (PreparedSink input m output) }--instance Monad m => Functor (Sink input m) where-    fmap f (Sink msink) = Sink (liftM (fmap f) msink)+    fmap f (SinkLift msink) = SinkLift (liftM (fmap f) msink)  instance Resource m => Applicative (Sink input m) where-    pure x = Sink (return (SinkNoData x))-    Sink mf <*> Sink ma = Sink $ do-        f <- mf-        a <- ma-        case (f, a) of-            (SinkNoData f', SinkNoData a') -> return (SinkNoData (f' a'))-            _ -> do-                istate <- newRef (toEither f, toEither a)-                return $ appHelper istate--toEither :: PreparedSink input m output -> SinkEither input m output-toEither (SinkData x y) = SinkPair x y-toEither (SinkNoData x) = SinkOutput x--type SinkPush input m output = input -> ResourceT m (SinkResult input output)-type SinkClose input m output = ResourceT m output-data SinkEither input m output-    = SinkPair (SinkPush input m output) (SinkClose input m output)-    | SinkOutput output-type SinkState input m a b = Ref (Base m) (SinkEither input m (a -> b), SinkEither input m a)+    pure = return+    (<*>) = ap -appHelper :: Resource m => SinkState input m a b -> PreparedSink input m b-appHelper istate = SinkData (pushHelper istate) (closeHelper istate)+instance Resource m => Monad (Sink input m) where+    return = SinkNoData+    SinkNoData x >>= f = f x+    SinkLift mx >>= f = SinkLift $ do+        x <- mx+        return $ x >>= f+    SinkData push0 close0 >>= f =+        SinkData (push push0) (close close0)+      where+        push push' input = do+            res <- push' input+            case res of+                Done lo output -> pushHelper lo (f output)+                Processing push'' close'' ->+                    return $ Processing (push push'') (close close'') -pushHelper :: Resource m-           => SinkState input m a b-           -> input-           -> ResourceT m (SinkResult input b)-pushHelper istate stream0 = do-    state <- readRef istate-    go state stream0-  where-    go (SinkPair f _, eb) stream = do-        mres <- f stream-        case mres of-            Processing -> return Processing-            Done leftover res -> do-                let state' = (SinkOutput res, eb)-                writeRef istate state'-                maybe (return Processing) (go state') leftover-    go (f@SinkOutput{}, SinkPair b _) stream = do-        mres <- b stream-        case mres of-            Processing -> return Processing-            Done leftover res -> do-                let state' = (f, SinkOutput res)-                writeRef istate state'-                maybe (return Processing) (go state') leftover-    go (SinkOutput f, SinkOutput b) leftover = return $ Done (Just leftover) $ f b+        pushHelper lo (SinkNoData y) = return $ Done lo y+        pushHelper (Just l) (SinkData pushF _) = pushF l+        pushHelper Nothing (SinkData pushF closeF) =+            return (Processing pushF closeF)+        pushHelper lo (SinkLift msink) = msink >>= pushHelper lo -closeHelper :: Resource m-            => SinkState input m a b-            -> ResourceT m b-closeHelper istate = do-    (sf, sa) <- readRef istate-    case sf of-        SinkOutput f -> go' f sa-        SinkPair _ close -> do-            f <- close-            go' f sa-  where-    go' f (SinkPair _ close) = do-        a <- close-        return (f a)-    go' f (SinkOutput a) = return (f a)+        close close' = do+            output <- close'+            closeHelper (f output) -instance Resource m => Monad (Sink input m) where-    return = pure-    mx >>= f = Sink $ do-        x <- prepareSink mx-        case x of-            SinkNoData x' -> prepareSink $ f x'-            SinkData push' close' -> do-                istate <- newRef $ Left (push', close')-                return $ SinkData (push istate) (close istate)-      where-        push istate input = do-            state <- readRef istate-            case state of-                Left (push', _) -> do-                    res <- push' input-                    case res of-                        Done leftover output -> do-                            f' <- prepareSink $ f output-                            case f' of-                                SinkNoData y ->-                                    return $ Done leftover y-                                SinkData pushF closeF -> do-                                    writeRef istate $ Right (pushF, closeF)-                                    maybe (return Processing) (push istate) leftover-                        Processing -> return Processing-                Right (push', _) -> push' input-        close istate = do-            state <- readRef istate-            case state of-                Left (_, close') -> do-                    output <- close'-                    f' <- prepareSink $ f output-                    case f' of-                        SinkNoData y -> return y-                        SinkData _ closeF -> closeF-                Right (_, close') -> close'+        closeHelper (SinkNoData y) = return y+        closeHelper (SinkData _ closeF) = closeF+        closeHelper (SinkLift msink) = msink >>= closeHelper  instance (Resource m, Base m ~ base, Applicative base) => MonadBase base (Sink input m) where     liftBase = lift . resourceLiftBase  instance MonadTrans (Sink input) where-    lift f = Sink (lift (liftM SinkNoData f))+    lift = SinkLift . liftM SinkNoData . lift  instance (Resource m, MonadIO m) => MonadIO (Sink input m) where     liftIO = lift . liftIO

@@ -1,128 +1,71 @@ {-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE DeriveDataTypeable #-} -- | Defines the types for a source, which is a producer of data. module Data.Conduit.Types.Source     ( SourceResult (..)-    , PreparedSource (..)     , Source (..)-    , SourceInvariantException (..)     ) where  import Control.Monad.Trans.Resource import Data.Monoid (Monoid (..)) import Control.Monad (liftM)-import Data.Typeable (Typeable)-import Control.Exception (Exception, throw)  -- | Result of pulling from a source. Either a new piece of data (@Open@), or -- indicates that the source is now @Closed@. ----- Since 0.0.0-data SourceResult a = Open a | Closed-    deriving (Show, Eq, Ord)+-- The @Open@ constructor returns both a new value, as well as a new @Source@,+-- which should be used in place of the previous @Source@.+--+-- Since 0.2.0+data SourceResult m a = Open (Source m a) a | Closed -instance Functor SourceResult where-    fmap f (Open a) = Open (f a)+instance Monad m => Functor (SourceResult m) where+    fmap f (Open p a) = Open (fmap f p) (f a)     fmap _ Closed = Closed --- | A 'PreparedSource' has two operations on it: pull some data, and close the--- 'PreparedSource'. Since 'PreparedSource' is built on top of 'ResourceT', all--- acquired resources should be automatically released anyway. Closing a--- 'PreparedSource' early+-- | A @Source@ has two operations on it: pull some data, and close the+-- @Source@. Since @Source@ is built on top of 'ResourceT', all acquired+-- resources should be automatically released anyway. Closing a @Source@ early -- is merely an optimization to free scarce resources as soon as possible. ----- A 'PreparedSource' has three invariants:------ * It is illegal to call 'sourcePull' after a previous call returns 'Closed', or after a call to 'sourceClose'.------ * It is illegal to call 'sourceClose' multiple times, or after a previous--- 'sourcePull' returns a 'Closed'.------ * A 'PreparedSource' is responsible to free any resources when either 'sourceClose'--- is called or a 'Closed' is returned. However, based on the usage of--- 'ResourceT', this is simply an optimization.+-- A @Source@ is should free any resources it allocated when either+-- @sourceClose@ is called or a @Closed@ is returned. However, based on the+-- usage of @ResourceT@, this is simply an optimization. ----- Since 0.0.0-data PreparedSource m a = PreparedSource-    { sourcePull :: ResourceT m (SourceResult a)+-- Since 0.2.0+data Source m a = Source+    { sourcePull :: ResourceT m (SourceResult m a)     , sourceClose :: ResourceT m ()     } -instance Monad m => Functor (PreparedSource m) where+instance Monad m => Functor (Source m) where     fmap f src = src         { sourcePull = liftM (fmap f) (sourcePull src)         } --- | All but the simplest of 'PreparedSource's (e.g., @repeat@) require some--- type of state to track their current status. This may be in the form of a--- mutable variable (e.g., @IORef@), or via opening a resource like a @Handle@.--- While a 'PreparedSource' is given no opportunity to acquire such resources,--- this type is.------ A 'Source' is simply a monadic action that returns a 'PreparedSource'. One--- nice consequence of this is the possibility of creating an efficient--- 'Monoid' instance, which will only acquire one resource at a time, instead--- of bulk acquiring all resources at the beginning of running the 'Source'.------ Note that each time you \"call\" a @Source@, it is started from scratch. If--- you want a resumable source (e.g., one which can be passed to multiple--- @Sink@s), you likely want to use a 'BufferedSource'.------ Since 0.0.0-newtype Source m a = Source { prepareSource :: ResourceT m (PreparedSource m a) }--instance Monad m => Functor (Source m) where-    fmap f (Source msrc) = Source (liftM (fmap f) msrc)- instance Resource m => Monoid (Source m a) where-    mempty = Source (return PreparedSource+    mempty = Source         { sourcePull = return Closed         , sourceClose = return ()-        })+        }     mappend a b = mconcat [a, b]     mconcat [] = mempty-    mconcat (Source mnext:rest0) = Source $ do-        -- open up the first Source...-        next0 <- mnext-        -- and place it in a mutable reference along with all of the upcoming-        -- Sources-        istate <- newRef (next0, rest0)-        return PreparedSource-            { sourcePull = pull istate-            , sourceClose = close istate-            }+    mconcat (next0:rest0) =+        src next0 rest0       where-        pull istate =-            readRef istate >>= pull'-          where-            pull' (current, rest) = do-                res <- sourcePull current-                case res of-                    -- end of the current Source-                    Closed -> do-                        case rest of-                            -- ... and open the next one-                            Source ma:as -> do-                                a <- ma-                                writeRef istate (a, as)-                                -- continue pulling base on this new state-                                pull istate-                            -- no more source, return an EOF-                            [] -> do-                                -- give an error message if the first Source-                                -- invariant is violated (read data after EOF)-                                writeRef istate $-                                    throw $ PullAfterEOF "Source:mconcat"-                                return Closed-                    Open _ -> return res-        close istate = do+        src next rest = Source (pull next rest) (close next rest)++        pull current rest = do+            res <- sourcePull current+            case res of+                -- end of the current Source+                Closed -> do+                    case rest of+                        -- ... and open the next one+                        a:as -> pull a as+                        -- no more source, return an EOF+                        [] -> return Closed+                Open current' val -> return (Open (src current' rest) val)+        close current _rest = do             -- we only need to close the current Source, since they are opened             -- one at a time-            (current, _) <- readRef istate             sourceClose current---- |--- Since 0.0.0-data SourceInvariantException = PullAfterEOF String-    deriving (Show, Typeable)-instance Exception SourceInvariantException

@@ -6,7 +6,9 @@ -- "Data.Conduit.Types.Conduit" for more information on the base types. module Data.Conduit.Util.Conduit     ( conduitState+    , ConduitStateResult (..)     , conduitIO+    , ConduitIOResult (..)     , transConduit       -- *** Sequencing     , SequencedSink@@ -18,101 +20,111 @@ import Control.Monad.Trans.Class import Data.Conduit.Types.Conduit import Data.Conduit.Types.Sink-import Control.Monad (liftM) --- | Construct a 'Conduit' with some stateful functions. This function address--- all mutable state for you.+-- | A helper type for @conduitState@, indicating the result of being pushed+-- to.  It can either indicate that processing is done, or to continue with the+-- updated state. ----- Since 0.0.0+-- Since 0.2.0+data ConduitStateResult state input output =+    StateFinished (Maybe input) [output]+  | StateProducing state [output]++instance Functor (ConduitStateResult state input) where+    fmap f (StateFinished a b) = StateFinished a (map f b)+    fmap f (StateProducing a b) = StateProducing a (map f b)++-- | Construct a 'Conduit' with some stateful functions. This function addresses+-- threading the state value for you.+--+-- Since 0.2.0 conduitState     :: Resource m     => state -- ^ initial state-    -> (state -> input -> ResourceT m (state, ConduitResult input output)) -- ^ Push function.+    -> (state -> input -> ResourceT m (ConduitStateResult state input output)) -- ^ Push function.     -> (state -> ResourceT m [output]) -- ^ Close function. The state need not be returned, since it will not be used again.     -> Conduit input m output-conduitState state0 push close = Conduit $ do-#if DEBUG-    iclosed <- newRef False-#endif-    istate <- newRef state0-    return PreparedConduit-        { conduitPush = \input -> do-#if DEBUG-            False <- readRef iclosed-#endif-            state <- readRef istate-            (state', res) <- state `seq` push state input-            writeRef istate state'-#if DEBUG-            case res of-                Finished _ _ -> writeRef iclosed True-                Producing _ -> return ()-#endif-            return res-        , conduitClose = do-#if DEBUG-            False <- readRef iclosed-            writeRef iclosed True-#endif-            readRef istate >>= close-        }+conduitState state0 push0 close0 =+    Conduit (push state0) (close0 state0)+  where+    push state input = do+        res <- state `seq` push0 state input+        return $ case res of+            StateFinished a b -> Finished a b+            StateProducing state' output -> Producing+                (Conduit (push state') (close0 state'))+                output +-- | A helper type for @conduitIO@, indicating the result of being pushed to.+-- It can either indicate that processing is done, or to continue.+--+-- Since 0.2.0+data ConduitIOResult input output =+    IOFinished (Maybe input) [output]+  | IOProducing [output]++instance Functor (ConduitIOResult input) where+    fmap f (IOFinished a b) = IOFinished a (map f b)+    fmap f (IOProducing b) = IOProducing (map f b)+ -- | Construct a 'Conduit'. ----- Since 0.0.0+-- Since 0.2.0 conduitIO :: ResourceIO m            => IO state -- ^ resource and/or state allocation            -> (state -> IO ()) -- ^ resource and/or state cleanup-           -> (state -> input -> m (ConduitResult input output)) -- ^ Push function. Note that this need not explicitly perform any cleanup.+           -> (state -> input -> m (ConduitIOResult input output)) -- ^ Push function. Note that this need not explicitly perform any cleanup.            -> (state -> m [output]) -- ^ Close function. Note that this need not explicitly perform any cleanup.            -> Conduit input m output-conduitIO alloc cleanup push close = Conduit $ do-#if DEBUG-    iclosed <- newRef False-#endif-    (key, state) <- withIO alloc cleanup-    return PreparedConduit-        { conduitPush = \input -> do-#if DEBUG-            False <- readRef iclosed-#endif-            res <- lift $ push state input-            case res of-                Producing{} -> return ()-                Finished{} -> do-#if DEBUG-                    writeRef iclosed True-#endif-                    release key-            return res-        , conduitClose = do-#if DEBUG-            False <- readRef iclosed-            writeRef iclosed True-#endif-            output <- lift $ close state-            release key-            return output-        }+conduitIO alloc cleanup push0 close0 = Conduit+    { conduitPush = \input -> do+        (key, state) <- withIO alloc cleanup+        push key state input+    , conduitClose = do+        (key, state) <- withIO alloc cleanup+        close key state+    }+  where+    push key state input = do+        res <- lift $ push0 state input+        case res of+            IOProducing output -> return $ Producing+                (Conduit (push key state) (close key state))+                output+            IOFinished a b -> do+                release key+                return $ Finished a b+    close key state = do+        output <- lift $ close0 state+        release key+        return output  -- | Transform the monad a 'Conduit' lives in. ----- Since 0.0.0+-- See @transSource@ for more information.+--+-- Since 0.2.0 transConduit :: (Monad m, Base m ~ Base n)+             => (forall a. m a -> n a)+             -> Conduit input m output+             -> Conduit input n output+transConduit f c = c+    { conduitPush = transResourceT f . fmap (transConduitPush f) . conduitPush c+    , conduitClose = transResourceT f (conduitClose c)+    }++transConduitPush :: (Base m ~ Base n, Monad m)               => (forall a. m a -> n a)-              -> Conduit input m output-              -> Conduit input n output-transConduit f (Conduit mc) =-    Conduit (transResourceT f (liftM go mc))-  where-    go c = c-        { conduitPush = transResourceT f . conduitPush c-        , conduitClose = transResourceT f (conduitClose c)-        }+              -> ConduitResult input m output+              -> ConduitResult input n output+transConduitPush _ (Finished a b) = Finished a b+transConduitPush f (Producing conduit output) = Producing+    (transConduit f conduit)+    output  -- | Return value from a 'SequencedSink'. ----- Since 0.0.0+-- Since 0.2.0 data SequencedSinkResponse state input m output =     Emit state [output] -- ^ Set a new state, and emit some new output.   | Stop -- ^ End the conduit.@@ -122,19 +134,19 @@ -- to write higher-level code that takes advantage of existing conduits and -- sinks, and leverages a sink's monadic interface. ----- Since 0.0.0+-- Since 0.2.0 type SequencedSink state input m output =     state -> Sink input m (SequencedSinkResponse state input m output)  data SCState state input m output =     SCNewState state-  | SCConduit (PreparedConduit input m output)-  | SCSink (input -> ResourceT m (SinkResult input (SequencedSinkResponse state input m output)))+  | SCConduit (Conduit input m output)+  | SCSink (input -> ResourceT m (SinkResult input m (SequencedSinkResponse state input m output)))            (ResourceT m (SequencedSinkResponse state input m output))  -- | Convert a 'SequencedSink' into a 'Conduit'. ----- Since 0.0.0+-- Since 0.2.0 sequenceSink     :: Resource m     => state -- ^ initial state@@ -150,43 +162,40 @@       -> Maybe input       -> ([output] -> [output])       -> SequencedSink state input m output-      -> ResourceT m (SCState state input m output, ConduitResult input output)+      -> ResourceT m (ConduitStateResult (SCState state input m output) input output) goRes (Emit state output) (Just input) front fsink =     scPush (front . (output++)) fsink (SCNewState state) input goRes (Emit state output) Nothing front _ =-    return (SCNewState state, Producing $ front output)+    return $ StateProducing (SCNewState state) $ front output goRes Stop minput front _ =-    return (error "sequenceSink", Finished minput $ front [])-goRes (StartConduit c) Nothing front _ = do-    pc <- prepareConduit c-    return (SCConduit pc, Producing $ front [])-goRes (StartConduit c) (Just input) front fsink = do-    pc <- prepareConduit c-    scPush front fsink (SCConduit pc) input+    return $ StateFinished minput $ front []+goRes (StartConduit c) Nothing front _ =+    return $ StateProducing (SCConduit c) $ front []+goRes (StartConduit c) (Just input) front fsink =+    scPush front fsink (SCConduit c) input  scPush :: Resource m      => ([output] -> [output])      -> SequencedSink state input m output      -> SCState state input m output      -> input-     -> ResourceT m (SCState state input m output, ConduitResult input output)-scPush front fsink (SCNewState state) input = do-    sink <- prepareSink $ fsink state-    case sink of-        SinkData push' close' -> scPush front fsink (SCSink push' close') input-        SinkNoData res -> goRes res (Just input) front fsink+     -> ResourceT m (ConduitStateResult (SCState state input m output) input output)+scPush front fsink (SCNewState state) input =+    go (fsink state)+  where+    go (SinkData push' close') = scPush front fsink (SCSink push' close') input+    go (SinkNoData res) = goRes res (Just input) front fsink+    go (SinkLift msink) = msink >>= go scPush front _ (SCConduit conduit) input = do     res <- conduitPush conduit input-    let res' =-            case res of-                Producing x -> Producing $ front x-                Finished x y -> Finished x $ front y-    return (SCConduit conduit, res')-scPush front fsink (SCSink push close) input = do+    return $ case res of+        Producing c x -> StateProducing (SCConduit c) $ front x+        Finished x y -> StateFinished x $ front y+scPush front fsink (SCSink push _) input = do     mres <- push input     case mres of         Done minput res -> goRes res minput front fsink-        Processing -> return (SCSink push close, Producing $ front [])+        Processing push' close' -> return (StateProducing (SCSink push' close') $ front [])  scClose :: Monad m => SCState state inptu m output -> ResourceT m [output] scClose (SCNewState _) = return []@@ -196,6 +205,4 @@     case res of         Emit _ os -> return os         Stop -> return []-        StartConduit c -> do-            pc <- prepareConduit c-            conduitClose pc+        StartConduit c -> conduitClose c

@@ -6,7 +6,9 @@ -- for more information on the base types. module Data.Conduit.Util.Sink     ( sinkState+    , SinkStateResult (..)     , sinkIO+    , SinkIOResult (..)     , transSink     ) where @@ -15,93 +17,93 @@ import Data.Conduit.Types.Sink import Control.Monad (liftM) --- | Construct a 'Sink' with some stateful functions. This function address--- all mutable state for you.+-- | A helper type for @sinkState@, indicating the result of being pushed to.+-- It can either indicate that processing is done, or to continue with the+-- updated state. ----- Since 0.0.0+-- Since 0.2.0+data SinkStateResult state input output =+    StateDone (Maybe input) output+  | StateProcessing state++-- | Construct a 'Sink' with some stateful functions. This function addresses+-- threading the state value for you.+--+-- Since 0.2.0 sinkState     :: Resource m     => state -- ^ initial state-    -> (state -> input -> ResourceT m (state, SinkResult input output)) -- ^ push+    -> (state -> input -> ResourceT m (SinkStateResult state input output)) -- ^ push     -> (state -> ResourceT m output) -- ^ Close. Note that the state is not returned, as it is not needed.     -> Sink input m output-sinkState state0 push close = Sink $ do-    istate <- newRef state0-#if DEBUG-    iclosed <- newRef False-#endif-    return SinkData-        { sinkPush = \input -> do-#if DEBUG-            False <- readRef iclosed-#endif-            state <- readRef istate-            (state', res) <- state `seq` push state input-            writeRef istate state'-#if DEBUG-            case res of-                Done{} -> writeRef iclosed True-                Processing -> return ()-#endif-            return res-        , sinkClose = do-#if DEBUG-            False <- readRef iclosed-            writeRef iclosed True-#endif-            readRef istate >>= close-        }+sinkState state0 push0 close0 =+    SinkData (push state0) (close0 state0)+  where+    push state input = do+        res <- state `seq` push0 state input+        case res of+            StateProcessing state' -> return $ Processing (push state') (close0 state')+            StateDone mleftover output -> return $ Done mleftover output +-- | A helper type for @sinkIO@, indicating the result of being pushed to. It+-- can either indicate that processing is done, or to continue.+--+-- Since 0.2.0+data SinkIOResult input output = IODone (Maybe input) output | IOProcessing+ -- | Construct a 'Sink'. Note that your push and close functions need not -- explicitly perform any cleanup. ----- Since 0.0.0+-- Since 0.2.0 sinkIO :: ResourceIO m         => IO state -- ^ resource and/or state allocation         -> (state -> IO ()) -- ^ resource and/or state cleanup-        -> (state -> input -> m (SinkResult input output)) -- ^ push+        -> (state -> input -> m (SinkIOResult input output)) -- ^ push         -> (state -> m output) -- ^ close         -> Sink input m output-sinkIO alloc cleanup push close = Sink $ do-    (key, state) <- withIO alloc cleanup-#if DEBUG-    iclosed <- newRef False-#endif-    return SinkData-        { sinkPush = \input -> do-#if DEBUG-            False <- readRef iclosed-#endif-            res <- lift $ push state input-            case res of-                Done{} -> do-                    release key-#if DEBUG-                    writeRef iclosed True-#endif-                Processing -> return ()-            return res-        , sinkClose = do-#if DEBUG-            False <- readRef iclosed-            writeRef iclosed True-#endif-            res <- lift $ close state-            release key-            return res-        }+sinkIO alloc cleanup push0 close0 = SinkData+    { sinkPush = \input -> do+        (key, state) <- withIO alloc cleanup+        push key state input+    , sinkClose = do+        (key, state) <- withIO alloc cleanup+        close key state+    }+  where+    push key state input = do+        res <- lift $ push0 state input+        case res of+            IODone a b -> do+                release key+                return $ Done a b+            IOProcessing -> return $ Processing+                (push key state)+                (close key state)+    close key state = do+        res <- lift $ close0 state+        release key+        return res  -- | Transform the monad a 'Sink' lives in. ----- Since 0.0.0+-- See @transSource@ for more information.+--+-- Since 0.2.0 transSink :: (Base m ~ Base n, Monad m)-           => (forall a. m a -> n a)-           -> Sink input m output-           -> Sink input n output-transSink f (Sink mc) =-    Sink (transResourceT f (liftM go mc))-  where-    go c = c-        { sinkPush = transResourceT f . sinkPush c-        , sinkClose = transResourceT f (sinkClose c)-        }+          => (forall a. m a -> n a)+          -> Sink input m output+          -> Sink input n output+transSink _ (SinkNoData x) = SinkNoData x+transSink f (SinkLift msink) = SinkLift (transResourceT f (liftM (transSink f) msink))+transSink f (SinkData push close) = SinkData+    (transResourceT f . fmap (transSinkPush f) . push)+    (transResourceT f close)++transSinkPush :: (Base m ~ Base n, Monad m)+              => (forall a. m a -> n a)+              -> SinkResult input m output+              -> SinkResult input n output+transSinkPush _ (Done a b) = Done a b+transSinkPush f (Processing push close) = Processing+    (transResourceT f . fmap (transSinkPush f) . push)+    (transResourceT f close)

@@ -7,100 +7,89 @@ -- on the base types. module Data.Conduit.Util.Source     ( sourceState+    , SourceStateResult (..)     , sourceIO+    , SourceIOResult (..)     , transSource     ) where  import Control.Monad.Trans.Resource import Control.Monad.Trans.Class (lift) import Data.Conduit.Types.Source-import Control.Monad (liftM) --- | Construct a 'Source' with some stateful functions. This function address--- all mutable state for you.+-- | The return value when pulling in the @sourceState@ function. Either+-- indicates no more data, or the next value and an updated state. ----- Since 0.0.0+-- Since 0.2.0+data SourceStateResult state output = StateOpen state output | StateClosed++-- | Construct a 'Source' with some stateful functions. This function addresses+-- threading the state value for you.+--+-- Since 0.2.0 sourceState     :: Resource m     => state -- ^ Initial state-    -> (state -> ResourceT m (state, SourceResult output)) -- ^ Pull function+    -> (state -> ResourceT m (SourceStateResult state output)) -- ^ Pull function     -> Source m output-sourceState state0 pull = Source $ do-    istate <- newRef state0-#if DEBUG-    iclosed <- newRef False-#endif-    return PreparedSource-        { sourcePull = do-#if DEBUG-            False <- readRef iclosed-#endif-            state <- readRef istate-            (state', res) <- pull state-#if DEBUG-            let isClosed =-                    case res of-                        Closed -> True-                        Open _ -> False-            writeRef iclosed isClosed-#endif-            writeRef istate state'-            return res-        , sourceClose = do-#if DEBUG-            False <- readRef iclosed-            writeRef iclosed True-#else-            return ()-#endif-        }+sourceState state0 pull0 =+    src state0+  where+    src state = Source (pull state) close +    pull state = do+        res <- pull0 state+        return $ case res of+            StateOpen state' val -> Open (src state') val+            StateClosed -> Closed++    close = return ()++-- | The return value when pulling in the @sourceIO@ function. Either indicates+-- no more data, or the next value.+data SourceIOResult output = IOOpen output | IOClosed+ -- | Construct a 'Source' based on some IO actions for alloc/release. ----- Since 0.0.0+-- Since 0.2.0 sourceIO :: ResourceIO m           => IO state -- ^ resource and/or state allocation           -> (state -> IO ()) -- ^ resource and/or state cleanup-          -> (state -> m (SourceResult output)) -- ^ Pull function. Note that this need not explicitly perform any cleanup.+          -> (state -> m (SourceIOResult output)) -- ^ Pull function. Note that this need not explicitly perform any cleanup.           -> Source m output-sourceIO alloc cleanup pull = Source $ do-    (key, state) <- withIO alloc cleanup-#if DEBUG-    iclosed <- newRef False-#endif-    return PreparedSource+sourceIO alloc cleanup pull0 =+    Source         { sourcePull = do-#if DEBUG-            False <- readRef iclosed-#endif-            res <- lift $ pull state-            case res of-                Closed -> do-#if DEBUG-                    writeRef iclosed True-#endif-                    release key-                _ -> return ()-            return res-        , sourceClose = do-#if DEBUG-            False <- readRef iclosed-            writeRef iclosed True-#endif-            release key+            (key, state) <- withIO alloc cleanup+            pull key state+        , sourceClose = return ()         }+  where+    src key state = Source (pull key state) (release key) +    pull key state = do+        res <- lift $ pull0 state+        case res of+            IOClosed -> do+                release key+                return Closed+            IOOpen val -> return $ Open (src key state) val+ -- | Transform the monad a 'Source' lives in. ----- Since 0.0.0+-- Note that this will /not/ thread the individual monads together, meaning+-- side effects will be lost. This function is most useful for transformers+-- only providing context and not producing side-effects, such as @ReaderT@.+--+-- Since 0.2.0 transSource :: (Base m ~ Base n, Monad m)-             => (forall a. m a -> n a)-             -> Source m output-             -> Source n output-transSource f (Source mc) =-    Source (transResourceT f (liftM go mc))+            => (forall a. m a -> n a)+            -> Source m output+            -> Source n output+transSource f c = c+    { sourcePull = transResourceT f (fmap go2 $ sourcePull c)+    , sourceClose = transResourceT f (sourceClose c)+    }   where-    go c = c-        { sourcePull = transResourceT f (sourcePull c)-        , sourceClose = transResourceT f (sourceClose c)-        }+    go2 (Open p a) = Open (transSource f p) a+    go2 Closed = Closed

@@ -16,7 +16,7 @@ import qualified Data.ByteString as S import qualified Data.ByteString.Unsafe as BU import Prelude hiding (read)-import Data.Conduit.Types.Source (SourceResult (..))+import Data.Conduit.Util.Source (SourceIOResult (..))  #include <fcntl.h> @@ -45,13 +45,13 @@         then throwErrno $ "Could not open file: " ++ fp         else return $ FD h -read :: FD -> IO (SourceResult S.ByteString)+read :: FD -> IO (SourceIOResult S.ByteString) read fd = do     cstr <- mallocBytes 4096     len <- c_read fd cstr 4096     if len == 0-        then free cstr >> return Closed-        else fmap Open $ BU.unsafePackCStringFinalizer+        then free cstr >> return IOClosed+        else fmap IOOpen $ BU.unsafePackCStringFinalizer                 cstr                 (fromIntegral len)                 (free cstr)

@@ -18,7 +18,7 @@ import Data.Text.Encoding (encodeUtf16LE) import Data.Word (Word8) import Prelude hiding (read)-import Data.Conduit (SourceResult (..))+import Data.Conduit (SourceIOResult (..))  #include <fcntl.h> #include <Share.h>@@ -74,16 +74,16 @@         then throwErrno $ "Could not open file: " ++ fp         else return $ FD h -read :: FD -> IO (SourceResult S.ByteString)+read :: FD -> IO (SourceIOResult S.ByteString) read fd = do     cstr <- mallocBytes 4096     len <- c_read fd cstr 4096     if len == 0         then do             free cstr-            return Closed+            return IOClosed         else do-            fmap Open $ BU.unsafePackCStringFinalizer+            fmap IOOpen $ BU.unsafePackCStringFinalizer                 cstr                 (fromIntegral len)                 (free cstr)

@@ -1,14 +1,16 @@ Name:                conduit-Version:             0.1.1.1+Version:             0.2.0 Synopsis:            Streaming data processing library. Description: 	Conduits are an approach to the streaming data problem. It is meant as an alternative to enumerators\/iterators, hoping to address the same issues with different trade-offs based on real-world experience with enumerators. For more information, see <http://www.yesodweb.com/book/conduit>. 	. 	Release history: 	.-	* 0.1: @BufferedSource@ is now an abstract type, and has a much more efficient internal representation. The result was a 41% speedup on microbenchmarks (note: do not expect speedups anywhere near that in real usage). In general, we are moving towards @BufferedSource@ being a specific tool used internally as needed, but using @Source@ for all external APIs.+    [0.2] Instead of storing state in mutable variables, we now use CPS. A @Source@ returns the next @Source@, and likewise for @Sink@s and @Conduit@s. Not only does this take better advantage of GHC\'s optimizations (about a 20% speedup), but it allows some operations to have a reduction in algorithmic complexity from exponential to linear. This also allowed us to remove the @Prepared@ set of types. Also, the @State@ functions (e.g., @sinkState@) use better constructors for return types, avoiding the need for a dummy state on completion.+    .+	[0.1] @BufferedSource@ is now an abstract type, and has a much more efficient internal representation. The result was a 41% speedup on microbenchmarks (note: do not expect speedups anywhere near that in real usage). In general, we are moving towards @BufferedSource@ being a specific tool used internally as needed, but using @Source@ for all external APIs. 	.-	* 0.0: Initial release.+	[0.0] Initial release. License:             BSD3 License-file:        LICENSE Author:              Michael Snoyman@@ -64,7 +66,7 @@     cpp-options:   -DTEST     build-depends:   conduit                    , base-                   , hspec+                   , hspec >= 0.9.1                    , HUnit                    , QuickCheck                    , bytestring

@@ -15,10 +15,10 @@ import Control.Monad.ST (runST) import Data.Monoid import qualified Data.ByteString as S+import qualified Data.ByteString.Char8 as S8 import qualified Data.IORef as I import qualified Data.ByteString.Lazy as L import Data.ByteString.Lazy.Char8 ()-import Control.Monad.Trans.Writer (Writer) import qualified Data.Text as T import qualified Data.Text.Lazy as TL import qualified Data.Text.Lazy.Encoding as TLE@@ -226,13 +226,13 @@         it' "works inside a ResourceT" $ runResourceT $ do             counter <- liftIO $ I.newIORef 0             let incr i = C.sourceIO-                    (liftIO $ I.newIORef $ C.Open (i :: Int))+                    (liftIO $ I.newIORef $ C.IOOpen (i :: Int))                     (const $ return ())                     (\istate -> do                         res <- liftIO $ I.atomicModifyIORef istate-                            (\state -> (C.Closed, state))+                            (\state -> (C.IOClosed, state))                         case res of-                            C.Closed -> return ()+                            C.IOClosed -> return ()                             _ -> do                                 count <- liftIO $ I.atomicModifyIORef counter                                     (\j -> (j + 1, j + 1))@@ -400,5 +400,13 @@                 bsrc C.$= CB.isolate 10 C.$$ CL.head             x @?= Just "foobarbazb" -it' :: String -> IO () -> Writer [Spec] ()+    describe "binary" $ do+        prop "lines" $ \bss' -> runST $ runResourceT $ do+            let bss = map S.pack bss'+                bs = S.concat bss+                src = CL.sourceList bss+            res <- src C.$$ CB.lines C.=$ CL.consume+            return $ S8.lines bs == res++it' :: String -> IO () -> Specs it' = it