conduit 0.2.2 → 0.3.0

raw patch · 14 files changed

+824/−1504 lines, 14 filesdep +resourcetPVP ok

version bump matches the API change (PVP)

Dependencies added: resourcet

API changes (from Hackage documentation)

Files

• Control/Monad/Trans/Resource.hs +0/−558
• Data/Conduit.hs +238/−296
• Data/Conduit/Binary.hs +76/−73
• Data/Conduit/Lazy.hs +15/−18
• Data/Conduit/List.hs +111/−118
• Data/Conduit/Text.hs +17/−18
• Data/Conduit/Types/Conduit.hs +46/−32
• Data/Conduit/Types/Sink.hs +45/−102
• Data/Conduit/Types/Source.hs +20/−58
• Data/Conduit/Util/Conduit.hs +160/−134
• Data/Conduit/Util/Sink.hs +38/−42
• Data/Conduit/Util/Source.hs +40/−41
• conduit.cabal +5/−2
• test/main.hs +13/−12

@@ -1,558 +0,0 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE UndecidableInstances #-}-{-# LANGUAGE TypeFamilies #-}-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE CPP #-}-{-# LANGUAGE DeriveDataTypeable #-}--- | Allocate resources which are guaranteed to be released.------ For more information, see <http://www.yesodweb.com/blog/2011/12/resourcet>.------ One point to note: all register cleanup actions live in the base monad, not--- the main monad. This allows both more efficient code, and for monads to be--- transformed.-module Control.Monad.Trans.Resource-    ( -- * Data types-      ResourceT-    , ReleaseKey-      -- * Unwrap-    , runResourceT-      -- * Resource allocation-    , with-    , withIO-    , register-    , release-      -- * Use references-    , readRef-    , writeRef-    , newRef-      -- * Special actions-    , resourceForkIO-      -- * Monad transformation-    , transResourceT-      -- * A specific Exception transformer-    , ExceptionT (..)-    , runExceptionT_-      -- * Type class/associated types-    , Resource (..)-    , ResourceUnsafeIO (..)-    , ResourceIO-    , ResourceBaseIO (..)-    , ResourceThrow (..)-      -- ** Low-level-    , HasRef (..)-    , InvalidAccess (..)-    , resourceActive-    ) where--import Data.Typeable-import Data.IntMap (IntMap)-import qualified Data.IntMap as IntMap-import Control.Exception (SomeException, throw, Exception)-import Control.Monad.Trans.Control-    ( MonadTransControl (..), MonadBaseControl (..)-    , ComposeSt, defaultLiftBaseWith, defaultRestoreM-    , liftBaseDiscard-    )-import qualified Data.IORef as I-import Control.Monad.Base (MonadBase, liftBase)-import Control.Applicative (Applicative (..))-import Control.Monad.Trans.Class (MonadTrans (..))-import Control.Monad.IO.Class (MonadIO (..))-import Control.Monad (liftM)-import qualified Control.Exception as E-import Control.Monad.ST (ST, unsafeIOToST)-import qualified Control.Monad.ST.Lazy as Lazy-import qualified Data.STRef as S-import qualified Data.STRef.Lazy as SL-import Data.Monoid (Monoid)-import qualified Control.Exception.Lifted as L--import Control.Monad.Trans.Identity ( IdentityT)-import Control.Monad.Trans.List     ( ListT    )-import Control.Monad.Trans.Maybe    ( MaybeT   )-import Control.Monad.Trans.Error    ( ErrorT, Error)-import Control.Monad.Trans.Reader   ( ReaderT  )-import Control.Monad.Trans.State    ( StateT   )-import Control.Monad.Trans.Writer   ( WriterT  )-import Control.Monad.Trans.RWS      ( RWST     )--import Data.Word (Word)--import qualified Control.Monad.Trans.RWS.Strict    as Strict ( RWST   )-import qualified Control.Monad.Trans.State.Strict  as Strict ( StateT )-import qualified Control.Monad.Trans.Writer.Strict as Strict ( WriterT )-import Control.Concurrent (ThreadId, forkIO)---- | Create a new reference.-newRef :: Resource m => a -> ResourceT m (Ref (Base m) a)-newRef = lift . resourceLiftBase . newRef'-{-# INLINE newRef #-}---- | Read a value from a reference.-readRef :: Resource m => Ref (Base m) a -> ResourceT m a-readRef = lift . resourceLiftBase . readRef'-{-# INLINE readRef #-}---- | Write a value to a reference.-writeRef :: Resource m => Ref (Base m) a -> a -> ResourceT m ()-writeRef r = lift . resourceLiftBase . writeRef' r-{-# INLINE writeRef #-}---- | A base monad which provides mutable references and some exception-safe way--- of interacting with them. For monads which cannot handle exceptions (e.g.,--- 'ST'), exceptions may be ignored. However, in such cases, scarce resources--- should /not/ be allocated in those monads, as exceptions may cause the--- cleanup functions to not run.------ The instance for 'IO', however, is fully exception-safe.------ Minimal complete definition: @Ref@, @newRef'@, @readRef'@ and @writeRef'@.-class Monad m => HasRef m where-    type Ref m :: * -> *-    newRef' :: a -> m (Ref m a)-    readRef' :: Ref m a -> m a-    writeRef' :: Ref m a -> a -> m ()--    -- | For monads supporting multi-threaded access (e.g., @IO@), this much be-    -- an atomic modification.-    atomicModifyRef' :: Ref m a -> (a -> (a, b)) -> m b-    atomicModifyRef' sa f = do-        a0 <- readRef' sa-        let (a, b) = f a0-        writeRef' sa a-        return b--    mask :: ((forall a. m a -> m a) -> m b) -> m b-    mask f = f id--    mask_ :: m a -> m a-    mask_ = mask . const--    try :: m a -> m (Either SomeException a)-    try = liftM Right--instance HasRef IO where-    type Ref IO = I.IORef-    newRef' = I.newIORef-    {-# INLINE newRef' #-}-    atomicModifyRef' = I.atomicModifyIORef-    {-# INLINE atomicModifyRef' #-}-    readRef' = I.readIORef-    {-# INLINE readRef' #-}-    writeRef' = I.writeIORef-    {-# INLINE writeRef' #-}-    mask = E.mask-    {-# INLINE mask #-}-    mask_ = E.mask_-    {-# INLINE mask_ #-}-    try = E.try-    {-# INLINE try #-}--instance HasRef (ST s) where-    type Ref (ST s) = S.STRef s-    newRef' = S.newSTRef-    readRef' = S.readSTRef-    writeRef' = S.writeSTRef--instance HasRef (Lazy.ST s) where-    type Ref (Lazy.ST s) = SL.STRef s-    newRef' = SL.newSTRef-    readRef' = SL.readSTRef-    writeRef' = SL.writeSTRef---- | A 'Monad' with a base that has mutable references, and allows some way to--- run base actions and clean up properly.-class (HasRef (Base m), Monad m) => Resource m where-    -- | The base monad for the current monad stack. This will usually be @IO@-    -- or @ST@.-    type Base m :: * -> *--    -- | Run some action in the @Base@ monad. This function corresponds to-    -- 'liftBase', but due to various type issues, we need to have our own-    -- version here.-    resourceLiftBase :: Base m a -> m a--    -- | Guarantee that some initialization and cleanup code is called before-    -- and after some action. Note that the initialization and cleanup lives in-    -- the base monad, while the body is in the top monad.-    resourceBracket_ :: Base m () -- ^ init-                     -> Base m () -- ^ cleanup-                     -> m c       -- ^ body-                     -> m c--instance Resource IO where-    type Base IO = IO-    resourceLiftBase = id-    resourceBracket_ = E.bracket_--instance Resource (ST s) where-    type Base (ST s) = ST s-    resourceLiftBase = id-    resourceBracket_ ma mb mc = do-        ma-        c <- mc-        mb-        return c--instance Resource (Lazy.ST s) where-    type Base (Lazy.ST s) = Lazy.ST s-    resourceLiftBase = id-    resourceBracket_ ma mb mc = do-        ma-        c <- mc-        mb-        return c--instance (MonadTransControl t, Resource m, Monad (t m))-        => Resource (t m) where-    type Base (t m) = Base m--    resourceLiftBase = lift . resourceLiftBase-    resourceBracket_ a b c =-        control' $ \run -> resourceBracket_ a b (run c)-      where-        control' f = liftWith f >>= restoreT . return---- | A 'Resource' based on some monad which allows running of some 'IO'--- actions, via unsafe calls. This applies to 'IO' and 'ST', for instance.-class Resource m => ResourceUnsafeIO m where-    unsafeFromIO :: IO a -> m a--instance ResourceUnsafeIO IO where-    unsafeFromIO = id--instance ResourceUnsafeIO (ST s) where-    unsafeFromIO = unsafeIOToST--instance ResourceUnsafeIO (Lazy.ST s) where-    unsafeFromIO = Lazy.unsafeIOToST--instance (MonadTransControl t, ResourceUnsafeIO m, Monad (t m)) => ResourceUnsafeIO (t m) where-    unsafeFromIO = lift . unsafeFromIO---- | A helper class for 'ResourceIO', stating that the base monad provides @IO@--- actions.-class ResourceBaseIO m where-    safeFromIOBase :: IO a -> m a--instance ResourceBaseIO IO where-    safeFromIOBase = id---- | A 'Resource' which can safely run 'IO' calls.-class (ResourceBaseIO (Base m), ResourceUnsafeIO m, ResourceThrow m,-       MonadIO m, MonadBaseControl IO m)-        => ResourceIO m--instance ResourceIO IO--instance (MonadTransControl t, ResourceIO m, Monad (t m), ResourceThrow (t m),-          MonadBaseControl IO (t m), MonadIO (t m))-        => ResourceIO (t m)---- | A lookup key for a specific release action. This value is returned by--- 'register', 'with' and 'withIO', and is passed to 'release'.-newtype ReleaseKey = ReleaseKey Int-    deriving Typeable--type RefCount = Word-type NextKey = Int--data ReleaseMap base =-    ReleaseMap !NextKey !RefCount !(IntMap (base ()))-  | ReleaseMapClosed---- | The Resource transformer. This transformer keeps track of all registered--- actions, and calls them upon exit (via 'runResourceT'). Actions may be--- registered via 'register', or resources may be allocated atomically via--- 'with' or 'withIO'. The with functions correspond closely to @bracket@.------ Releasing may be performed before exit via the 'release' function. This is a--- highly recommended optimization, as it will ensure that scarce resources are--- freed early. Note that calling @release@ will deregister the action, so that--- a release action will only ever be called once.-newtype ResourceT m a =-    ResourceT (Ref (Base m) (ReleaseMap (Base m)) -> m a)--instance Typeable1 m => Typeable1 (ResourceT m) where-    typeOf1 = goType undefined-      where-        goType :: Typeable1 m => m a -> ResourceT m a -> TypeRep-        goType m _ =-            mkTyConApp-                (mkTyCon "Control.Monad.Trans.Resource.ResourceT")-                [ typeOf1 m-                ]---- | Perform some allocation, and automatically register a cleanup action.------ If you are performing an @IO@ action, it will likely be easier to use the--- 'withIO' function, which handles types more cleanly.-with :: Resource m-     => Base m a -- ^ allocate-     -> (a -> Base m ()) -- ^ free resource-     -> ResourceT m (ReleaseKey, a)-with acquire rel = ResourceT $ \istate -> resourceLiftBase $ mask $ \restore -> do-    a <- restore acquire-    key <- register' istate $ rel a-    return (key, a)---- | Same as 'with', but explicitly uses @IO@ as a base.-withIO :: ResourceIO m-       => IO a -- ^ allocate-       -> (a -> IO ()) -- ^ free resource-       -> ResourceT m (ReleaseKey, a)-withIO acquire rel = ResourceT $ \istate -> resourceLiftBase $ mask $ \restore -> do-    a <- restore $ safeFromIOBase acquire-    key <- register' istate $ safeFromIOBase $ safeFromIOBase $ rel a-    return (key, a)---- | Register some action that will be called precisely once, either when--- 'runResourceT' is called, or when the 'ReleaseKey' is passed to 'release'.-register :: Resource m-         => Base m ()-         -> ResourceT m ReleaseKey-register rel = ResourceT $ \istate -> resourceLiftBase $ register' istate rel--register' :: HasRef base-          => Ref base (ReleaseMap base)-          -> base ()-          -> base ReleaseKey-register' istate rel = atomicModifyRef' istate $ \rm ->-    case rm of-        ReleaseMap key rf m ->-            ( ReleaseMap (key + 1) rf (IntMap.insert key rel m)-            , ReleaseKey key-            )-        ReleaseMapClosed -> throw $ InvalidAccess "register'"--data InvalidAccess = InvalidAccess { functionName :: String }-    deriving Typeable--instance Show InvalidAccess where-    show (InvalidAccess f) = concat-        [ "Control.Monad.Trans.Resource."-        , f-        , ": The mutable state is being accessed after cleanup. Please contact the maintainers."-        ]--instance Exception InvalidAccess---- | Call a release action early, and deregister it from the list of cleanup--- actions to be performed.-release :: Resource m-        => ReleaseKey-        -> ResourceT m ()-release rk = ResourceT $ \istate -> resourceLiftBase $ release' istate rk--release' :: HasRef base-         => Ref base (ReleaseMap base)-         -> ReleaseKey-         -> base ()-release' istate (ReleaseKey key) = mask $ \restore -> do-    maction <- atomicModifyRef' istate lookupAction-    maybe (return ()) restore maction-  where-    lookupAction rm@(ReleaseMap next rf m) =-        case IntMap.lookup key m of-            Nothing -> (rm, Nothing)-            Just action ->-                ( ReleaseMap next rf $ IntMap.delete key m-                , Just action-                )-    lookupAction ReleaseMapClosed = throw $ InvalidAccess "release'"--stateAlloc :: HasRef m => Ref m (ReleaseMap m) -> m ()-stateAlloc istate = do-    atomicModifyRef' istate $ \rm ->-        case rm of-            ReleaseMap nk rf m ->-                (ReleaseMap nk (rf + 1) m, ())-            ReleaseMapClosed -> throw $ InvalidAccess "stateAlloc"--stateCleanup :: HasRef m => Ref m (ReleaseMap m) -> m ()-stateCleanup istate = mask_ $ do-    mm <- atomicModifyRef' istate $ \rm ->-        case rm of-            ReleaseMap nk rf m ->-                let rf' = rf - 1-                 in if rf' == minBound-                        then (ReleaseMapClosed, Just m)-                        else (ReleaseMap nk rf' m, Nothing)-            ReleaseMapClosed -> throw $ InvalidAccess "stateCleanup"-    case mm of-        Just m ->-            mapM_ (\x -> try x >> return ()) $ IntMap.elems m-        Nothing -> return ()---- | Unwrap a 'ResourceT' transformer, and call all registered release actions.------ Note that there is some reference counting involved due to 'resourceForkIO'.--- If multiple threads are sharing the same collection of resources, only the--- last call to @runResourceT@ will deallocate the resources.-runResourceT :: Resource m => ResourceT m a -> m a-runResourceT (ResourceT r) = do-    istate <- resourceLiftBase $ newRef'-        $ ReleaseMap minBound minBound IntMap.empty-    resourceBracket_-        (stateAlloc istate)-        (stateCleanup istate)-        (r istate)---- | Transform the monad a @ResourceT@ lives in. This is most often used to--- strip or add new transformers to a stack, e.g. to run a @ReaderT@. Note that--- the original and new monad must both have the same 'Base' monad.-transResourceT :: (Base m ~ Base n)-               => (m a -> n b)-               -> ResourceT m a-               -> ResourceT n b-transResourceT f (ResourceT mx) = ResourceT (\r -> f (mx r))---------- All of our monad et al instances-instance Monad m => Functor (ResourceT m) where-    fmap f (ResourceT m) = ResourceT $ \r -> liftM f (m r)--instance Monad m => Applicative (ResourceT m) where-    pure = ResourceT . const . return-    ResourceT mf <*> ResourceT ma = ResourceT $ \r -> do-        f <- mf r-        a <- ma r-        return $ f a--instance Monad m => Monad (ResourceT m) where-    return = pure-    ResourceT ma >>= f =-        ResourceT $ \r -> ma r >>= flip un r . f-      where-        un (ResourceT x) = x--instance MonadTrans ResourceT where-    lift = ResourceT . const--instance MonadIO m => MonadIO (ResourceT m) where-    liftIO = lift . liftIO--instance MonadBase b m => MonadBase b (ResourceT m) where-    liftBase = lift . liftBase--{--instance MonadTransControl ResourceT where-    newtype StT ResourceT a = StReader {unStReader :: a}-    liftWith f = ResourceT $ \r -> f $ \(ResourceT t) -> liftM StReader $ t r-    restoreT = ResourceT . const . liftM unStReader-    {-# INLINE liftWith #-}-    {-# INLINE restoreT #-}--}--instance MonadBaseControl b m => MonadBaseControl b (ResourceT m) where-     newtype StM (ResourceT m) a = StMT (StM m a)-     liftBaseWith f = ResourceT $ \reader ->-         liftBaseWith $ \runInBase ->-             f $ liftM StMT . runInBase . (\(ResourceT r) -> r reader)-     restoreM (StMT base) = ResourceT $ const $ restoreM base---- | The express purpose of this transformer is to allow the 'ST' monad to--- catch exceptions via the 'ResourceThrow' typeclass.-newtype ExceptionT m a = ExceptionT { runExceptionT :: m (Either SomeException a) }---- | Same as 'runExceptionT', but immediately 'E.throw' any exception returned.-runExceptionT_ :: Monad m => ExceptionT m a -> m a-runExceptionT_ = liftM (either E.throw id) . runExceptionT--instance Monad m => Functor (ExceptionT m) where-    fmap f = ExceptionT . (liftM . fmap) f . runExceptionT-instance Monad m => Applicative (ExceptionT m) where-    pure = ExceptionT . return . Right-    ExceptionT mf <*> ExceptionT ma = ExceptionT $ do-        ef <- mf-        case ef of-            Left e -> return (Left e)-            Right f -> do-                ea <- ma-                case ea of-                    Left e -> return (Left e)-                    Right x -> return (Right (f x))-instance Monad m => Monad (ExceptionT m) where-    return = pure-    ExceptionT ma >>= f = ExceptionT $ do-        ea <- ma-        case ea of-            Left e -> return (Left e)-            Right a -> runExceptionT (f a)-instance MonadBase b m => MonadBase b (ExceptionT m) where-    liftBase = lift . liftBase-instance MonadTrans ExceptionT where-    lift = ExceptionT . liftM Right-instance MonadTransControl ExceptionT where-    newtype StT ExceptionT a = StExc { unStExc :: Either SomeException a }-    liftWith f = ExceptionT $ liftM return $ f $ liftM StExc . runExceptionT-    restoreT = ExceptionT . liftM unStExc-instance MonadBaseControl b m => MonadBaseControl b (ExceptionT m) where-    newtype StM (ExceptionT m) a = StE { unStE :: ComposeSt ExceptionT m a }-    liftBaseWith = defaultLiftBaseWith StE-    restoreM = defaultRestoreM unStE-instance (Resource m, MonadBaseControl (Base m) m)-        => ResourceThrow (ExceptionT m) where-    resourceThrow = ExceptionT . return . Left . E.toException---- | A 'Resource' which can throw exceptions. Note that this does not work in a--- vanilla @ST@ monad. Instead, you should use the 'ExceptionT' transformer on--- top of @ST@.-class Resource m => ResourceThrow m where-    resourceThrow :: E.Exception e => e -> m a--instance ResourceThrow IO where-    resourceThrow = E.throwIO--#define GO(T) instance (ResourceThrow m) => ResourceThrow (T m) where resourceThrow = lift . resourceThrow-#define GOX(X, T) instance (X, ResourceThrow m) => ResourceThrow (T m) where resourceThrow = lift . resourceThrow-GO(IdentityT)-GO(ListT)-GO(MaybeT)-GOX(Error e, ErrorT e)-GO(ReaderT r)-GO(StateT s)-GOX(Monoid w, WriterT w)-GOX(Monoid w, RWST r w s)-GOX(Monoid w, Strict.RWST r w s)-GO(Strict.StateT s)-GOX(Monoid w, Strict.WriterT w)-#undef GO-#undef GOX---- | Introduce a reference-counting scheme to allow a resource context to be--- shared by multiple threads. Once the last thread exits, all remaining--- resources will be released.------ Note that abuse of this function will greatly delay the deallocation of--- registered resources. This function should be used with care. A general--- guideline:------ If you are allocating a resource that should be shared by multiple threads,--- and will be held for a long time, you should allocate it at the beginning of--- a new @ResourceT@ block and then call @resourceForkIO@ from there.-resourceForkIO :: ResourceIO m => ResourceT m () -> ResourceT m ThreadId-resourceForkIO (ResourceT f) = ResourceT $ \r -> L.mask $ \restore ->-    -- We need to make sure the counter is incremented before this call-    -- returns. Otherwise, the parent thread may call runResourceT before-    -- the child thread increments, and all resources will be freed-    -- before the child gets called.-    resourceBracket_-        (stateAlloc r)-        (return ())-        (liftBaseDiscard forkIO $ resourceBracket_-            (return ())-            (stateCleanup r)-            (restore $ f r))---- | Determine if the current @ResourceT@ is still active. This is necessary--- for such cases as lazy I\/O, where an unevaluated thunk may still refer to a--- closed @ResourceT@.-resourceActive :: Resource m => ResourceT m Bool-resourceActive = ResourceT $ \rmMap -> do-    rm <- resourceLiftBase $ readRef' rmMap-    case rm of-        ReleaseMapClosed -> return False-        _ -> return True

@@ -50,16 +50,16 @@     , Flush (..)       -- * Convenience re-exports     , ResourceT-    , Resource (..)-    , ResourceIO-    , ResourceUnsafeIO+    , MonadResource+    , MonadThrow (..)+    , MonadUnsafeIO (..)     , runResourceT-    , ResourceThrow (..)     ) where -import Control.Applicative ((<$>)) import Control.Monad (liftM) import Control.Monad.Trans.Resource+import Control.Monad.IO.Class (MonadIO (liftIO))+import qualified Data.IORef as I import Data.Conduit.Types.Source import Data.Conduit.Util.Source import Data.Conduit.Types.Sink@@ -72,226 +72,188 @@ infixr 0 $$  -- | The connect operator, which pulls data from a source and pushes to a sink.--- There are three ways this process can terminate:------ 1. In the case of a @SinkNoData@ constructor, the source is not opened at--- all, and the output value is returned immediately.+-- There are two ways this process can terminate: ----- 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.+-- 1. If the @Sink@ is a @Done@ constructor, the @Source@ is closed. ----- 3. The source return @Closed@, in which case the sink is closed.+-- 2. If the @Source@ is a @Closed@ constructor, 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.+-- This function will automatically close any @Source@s, but will not close any+-- @BufferedSource@s, allowing them to be reused. Also, leftover data will be+-- discarded when connecting a @Source@, but will be buffered when using a+-- @BufferedSource@. ----- Since 0.2.0-($$) :: (IsSource src, Resource m) => src m a -> Sink a m b -> ResourceT m b+-- Since 0.3.0+($$) :: IsSource src m => src m a -> Sink a m b -> 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+-- Since 0.3.0+class IsSource src m where+    connect :: src m a -> Sink a m b -> m b+    fuseLeft :: src m a -> Conduit a m b -> Source m b -instance IsSource Source where+instance Monad m => IsSource Source m where     connect = normalConnect     {-# INLINE connect #-}     fuseLeft = normalFuseLeft     {-# INLINE fuseLeft #-} -instance IsSource BufferedSource where+instance MonadIO m => IsSource BufferedSource m where     connect = bufferedConnect     {-# INLINE connect #-}     fuseLeft = bufferedFuseLeft     {-# INLINE fuseLeft #-} -normalConnect :: Resource m => Source m a -> Sink a m b -> ResourceT m b-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 = 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'+normalConnect :: Monad m => Source m a -> Sink a m b -> m b -data FuseLeftState src conduit output =-    FLClosed [output]-  | FLOpen src conduit [output]+-- @Sink@ cannot handle any more input, close the @Source@ (regardless of its+-- state) and discard leftovers.+normalConnect src (Done _leftover output) = sourceClose src >> return output +-- Run the @Sink@'s monadic action and try again.+normalConnect src (SinkM msink) = msink >>= normalConnect src++-- Run the @Source@'s monadic action and try again.+normalConnect (SourceM msrc _) sink@Processing{} = msrc >>= flip normalConnect sink++-- No more input available from @Source@, close the @Sink@.+normalConnect Closed (Processing _ close) = close++-- More input available, and the @Sink@ wants it: plug it in and keep going.+normalConnect (Open src _ a) (Processing push _) = normalConnect src $ push a++data FuseLeftState srcState input m output =+    FLClosed+  | FLOpen srcState (ConduitPush input m output) (ConduitClose m output)+  | FLHaveOutput srcState (Conduit input m output) (m ())+ infixl 1 $=  -- | Left fuse, combining a source and a conduit together into a new source. ----- Note that any @Source@ passed in will be automatically closed, while a--- @BufferedSource@ will be left open.+-- Any @Source@ passed in will be automatically closed, while a+-- @BufferedSource@ will be left open. Leftover input will be discarded for a+-- @Source@, and buffered for a @BufferedSource@. ----- Since 0.2.0-($=) :: (IsSource src, Resource m)+-- Since 0.3.0+($=) :: IsSource src m      => src m a      -> Conduit a m b      -> Source m b ($=) = fuseLeft {-# INLINE ($=) #-} -normalFuseLeft :: Resource m => Source m a -> Conduit a m b -> Source m b-normalFuseLeft src0 conduit0 = Source-    { sourcePull = pull $ FLOpen src0 conduit0 []-    , sourceClose = return ()-    }-  where-    mkSrc state = Source (pull state) (close state)-    pull state' =-        case state' of-            FLClosed [] -> return Closed-            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 conduit-                        case res of-                            [] -> return Closed-                            x:xs -> return $ Open-                                (mkSrc (FLClosed xs))-                                x-                    Open src'' input -> do-                        res' <- conduitPush conduit input-                        case res' of-                            Producing conduit' [] ->-                                pull $ FLOpen src'' conduit' []-                            Producing conduit' (x:xs) -> return $ Open-                                (mkSrc (FLOpen src'' conduit' xs))-                                x-                            Finished _leftover output -> do-                                sourceClose src''-                                case output of-                                    [] -> 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-        case state of-            FLClosed _ -> return ()-            FLOpen src' (Conduit _ closeC) _ -> do-                _ignored <- closeC-                sourceClose src'+normalFuseLeft :: Monad m => Source m a -> Conduit a m b -> Source m b +-- No more input, close the @Conduit@.+normalFuseLeft Closed (NeedInput _ close) = close+normalFuseLeft Closed (Finished _) = Closed++-- @Conduit@ is done, discard leftovers and close the @Source@.+normalFuseLeft src (Finished _) = SourceM+    (sourceClose src >> return Closed)+    (sourceClose src)++-- @Conduit@ has some output, return it and keep going.+normalFuseLeft src (HaveOutput p c x) = Open+    (normalFuseLeft src p)+    (sourceClose src >> c)+    x++-- @Source@ provided input, and @Conduit@ wants it. Pipe it through and keep going.+normalFuseLeft (Open src _ a) (NeedInput push _) = normalFuseLeft src $ push a++-- Need to perform a monadic action to get the next @Conduit@.+normalFuseLeft src (ConduitM mcon conclose) = SourceM+    (liftM (normalFuseLeft src) mcon)+    (conclose >> sourceClose src)++-- Need to perform a monadic action to get the next @Source@.+normalFuseLeft (SourceM msrc closeS) conduit@(NeedInput _ closeC) =+    SourceM (liftM (flip normalFuseLeft conduit) msrc) $ do+        closeS+        sourceClose closeC+ infixr 0 =$  -- | Right fuse, combining a conduit and a sink together into a new sink. ----- Since 0.2.0-(=$) :: Resource m => Conduit a m b -> Sink b m c -> Sink a m c-_ =$ 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-    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 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'-                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+-- Any leftover data returns from the @Sink@ will be discarded.+--+-- Since 0.3.0+(=$) :: Monad m => Conduit a m b -> Sink b m c -> Sink a m c +-- @Sink@ is complete, discard leftovers, close the @Conduit@ and return the+-- output.+conduit =$ Done _leftover output = SinkM $ conduitClose conduit >> return (Done Nothing output)++-- Need to perform a monadic action to get the next @Sink@.+conduit =$ SinkM msink = SinkM (liftM (conduit =$) msink)++-- Need more input for the @Conduit@, so ask for it.+NeedInput pushO closeO =$ sink = Processing+    (\input -> pushO input =$ sink)+    (closeO $$ sink)++-- @Conduit@ can provide no more input to @Sink@. Close the @Sink@ and return+-- the leftovers from the @Conduit@.+Finished mleftover =$ Processing _ close = SinkM $ liftM (Done mleftover) close++-- Perform a monadic action to get the next @Conduit@.+ConduitM mcon _ =$ sink@Processing{} = SinkM $ liftM (=$ sink) mcon++-- @Conduit@ is providing input for the @Sink@, so use it and keep going.+HaveOutput con _ input =$ Processing pushI _ = con =$ pushI input+ infixr 0 =$=  -- | Middle fuse, combining two conduits together into a new conduit. ----- Since 0.2.0-(=$=) :: Resource m => Conduit a m b -> Conduit b m c -> Conduit a m c-outerOrig =$= innerOrig = Conduit-    (pushF outerOrig innerOrig)-    (closeF outerOrig innerOrig)+-- Any leftovers provided by the inner @Conduit@ will be discarded.+--+-- Since 0.3.0+(=$=) :: Monad m => Conduit a m b -> Conduit b m c -> Conduit a m c++-- No more input from outer conduit, and inner wants more input. Close the+-- inner conduit and convert its source into a conduit.+Finished mleftover =$= NeedInput _ closeI =+    go closeI   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+    go Closed = Finished mleftover+    go (Open src close x) = HaveOutput (go src) close x+    go (SourceM msrc close) = ConduitM (liftM go msrc) close --- | Push some data to a conduit, then close it if necessary.-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 conduit b -> do-            b' <- conduitPushClose conduit rest-            return $ b ++ b'+-- No more input from outer conduit, and inner wants no more input anyway.+-- Discard the leftovers from the inner.+Finished mleftover =$= Finished _ = Finished mleftover +-- Provide more output from the inner conduit.+conO =$= HaveOutput con close x = HaveOutput (conO =$= con) close x++-- Perform a monadic action to get the next outer conduit.+ConduitM mcon close =$= conI = ConduitM (liftM (=$= conI) mcon) (close >> conduitClose conI)++-- Ask for more data for the outer conduit. Note that this clause comes before+-- the inner conduit ConduitM clause, so that we only run that action as+-- necessary.+NeedInput pushO closeO =$= conI = NeedInput+    (\input -> pushO input =$= conI)+    (closeO $= conI)++-- Perform a monadic action to get the next inner conduit.+conO =$= ConduitM mconI close = ConduitM (liftM (conO =$=) mconI) (close >> conduitClose conO)++-- Pipe output from outer conduit to inner conduit and keep going.+HaveOutput conO _ inputI =$= NeedInput pushI _ = conO =$= pushI inputI++-- Discard output from outer conduit, and discard leftovers from inner conduit.+-- Close the outer conduit.+HaveOutput _ close _ =$= Finished _ = ConduitM (close >> return (Finished Nothing)) close+ -- | 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.@@ -311,8 +273,8 @@ -- multiple threads, there is no guarantee that all @BufferedSource@s will -- handle this correctly. ----- Since 0.2.0-data BufferedSource m a = BufferedSource (Ref (Base m) (BSState m a))+-- Since 0.3.0+data BufferedSource m a = BufferedSource (I.IORef (BSState m a))  data BSState m a =     ClosedEmpty@@ -324,9 +286,9 @@ -- you should manually call 'bsourceClose' when the 'BufferedSource' is no -- longer in use. ----- Since 0.2.0-bufferSource :: Resource m => Source m a -> ResourceT m (BufferedSource m a)-bufferSource src = BufferedSource <$> newRef (OpenEmpty src)+-- Since 0.3.0+bufferSource :: MonadIO m => Source m a -> m (BufferedSource m a)+bufferSource src = liftM BufferedSource $ liftIO $ I.newIORef $ OpenEmpty src  -- | Turn a 'BufferedSource' into a 'Source'. Note that in general this will -- mean your original 'BufferedSource' will be closed. Additionally, all@@ -335,114 +297,94 @@ -- -- Note: @bufferSource@ . @unbufferSource@ is /not/ the identity function. ----- Since 0.2.0-unbufferSource :: Resource m+-- Since 0.3.0+unbufferSource :: MonadIO m                => BufferedSource m a                -> Source m a-unbufferSource (BufferedSource bs) = Source-    { sourcePull = msrc >>= sourcePull-    , sourceClose = msrc >>= sourceClose-    }+unbufferSource (BufferedSource bs) =+    SourceM msrc (msrc >>= sourceClose)   where     msrc = do-        buf <- readRef bs+        buf <- liftIO $ I.readIORef bs         case buf of             OpenEmpty src -> return src-            OpenFull src a -> return Source-                { sourcePull = return $ Open src 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 ()-                }+            OpenFull src a -> return $ Open src (sourceClose src) a+            ClosedEmpty -> return Closed+            ClosedFull a -> return $ Open Closed (return ()) a -bufferedConnect :: Resource m => BufferedSource m a -> Sink a m b -> ResourceT m b-bufferedConnect _ (SinkNoData output) = return output-bufferedConnect bsrc (SinkLift msink) = msink >>= bufferedConnect bsrc-bufferedConnect (BufferedSource bs) (SinkData push0 close0) = do-    bsState <- readRef bs+bufferedConnect :: MonadIO m => BufferedSource m a -> Sink a m b -> m b+bufferedConnect _ (Done Nothing output) = return output+bufferedConnect _ (Done Just{} _) = error "Invariant violated: sink returned leftover without input"+bufferedConnect bsrc (SinkM msink) = msink >>= bufferedConnect bsrc+bufferedConnect (BufferedSource bs) (Processing push0 close0) = do+    bsState <- liftIO $ I.readIORef bs     case bsState of         ClosedEmpty -> close0         OpenEmpty src -> connect' src push0 close0-        ClosedFull a -> do-            res <- push0 a-            case res of-                Done mleftover res' -> do-                    writeRef bs $ maybe ClosedEmpty ClosedFull mleftover-                    return res'-                Processing _ close' -> do-                    writeRef bs ClosedEmpty-                    close'-        OpenFull src a -> push0 a >>= onRes src+        ClosedFull a -> onRes Nothing $ push0 a+        OpenFull src a -> onRes (Just src) $ push0 a   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+    connect' Closed _ close = do+        liftIO $ I.writeIORef bs ClosedEmpty+        close+    connect' (Open src _ x) push _ = onRes (Just src) $ push x+    connect' (SourceM msrc _) push close = msrc >>= \src -> connect' src push close++    onRes msrc (Done mleftover res) = do+        let state =+                case (msrc, mleftover) of+                    (Nothing, Nothing) -> ClosedEmpty+                    (Just src, Nothing) -> OpenEmpty src+                    (Nothing, Just leftover) -> ClosedFull leftover+                    (Just src, Just leftover) -> OpenFull src leftover+        liftIO $ I.writeIORef bs state         return res-    onRes src (Processing push close) = connect' src push close+    onRes Nothing (Processing _ close) = do+        liftIO $ I.writeIORef bs ClosedEmpty+        close+    onRes (Just src) (Processing push close) = connect' src push close+    onRes msrc (SinkM msink) = msink >>= onRes msrc  bufferedFuseLeft-    :: Resource m+    :: MonadIO m     => BufferedSource m a     -> Conduit a m b     -> Source m b-bufferedFuseLeft bsrc conduit0 = Source-    { sourcePull = pullF $ FLOpen () conduit0 [] -- still open, no buffer-    , sourceClose = return ()-    }+bufferedFuseLeft bsrc (ConduitM mcon close) = SourceM+    (liftM (bufferedFuseLeft bsrc) mcon)+    close+bufferedFuseLeft _ (Finished _) = Closed+bufferedFuseLeft bsrc (HaveOutput next close x) = Open+    (bufferedFuseLeft bsrc next)+    close+    x+bufferedFuseLeft bsrc (NeedInput push0 close0) = SourceM+    (pullF $ FLOpen () push0 close0)+    (sourceClose close0)   where-    mkSrc state = Source+    mkSrc state = SourceM         (pullF state)         (closeF state)+     pullF state' =         case state' of-            FLClosed [] -> return Closed-            FLClosed (x:xs) -> return $ Open-                (mkSrc (FLClosed xs))-                x-            FLOpen () conduit (x:xs) -> return $ Open-                (mkSrc (FLOpen () conduit xs))-                x-            FLOpen () conduit [] -> do+            FLClosed -> return Closed+            FLHaveOutput () pull _ -> goRes pull+            FLOpen () push close -> do                 mres <- bsourcePull bsrc                 case mres of-                    Nothing -> do-                        res <- conduitClose conduit-                        case res of-                            [] -> return Closed-                            x:xs -> return $ Open-                                (mkSrc (FLClosed xs))-                                x-                    Just input -> do-                        res' <- conduitPush conduit input-                        case res' of-                            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-                                    [] -> return Closed-                                    x:xs -> return $ Open-                                        (mkSrc (FLClosed xs))-                                        x+                    Nothing -> return close+                    Just input -> goRes $ push input++    goRes (Finished leftover) = do+        bsourceUnpull bsrc leftover+        return Closed+    goRes (HaveOutput pull close' x) =+        let state = FLHaveOutput () pull close'+         in return $ Open (mkSrc state) (closeF state) x+    goRes (NeedInput pushI closeI) = pullF (FLOpen () pushI closeI)+    goRes (ConduitM mcon _) = mcon >>= goRes+     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,@@ -450,59 +392,59 @@         -- Source will return an Open while the Conduit will be Closed.         -- Therefore, we have to do a check.         case state of-            FLClosed _ -> return ()-            FLOpen () (Conduit _ close) _ -> do-                _ignored <- close+            FLClosed -> return ()+            FLOpen () _ close -> do+                () <- sourceClose close                 return ()+            FLHaveOutput () _ close -> close -bsourcePull :: Resource m => BufferedSource m a -> ResourceT m (Maybe a)-bsourcePull (BufferedSource bs) = do-    buf <- readRef bs-    case buf of-        OpenEmpty src -> do-            res <- sourcePull src-            case res of-                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 bs ClosedEmpty-            return $ Just a+bsourcePull :: MonadIO m => BufferedSource m a -> m (Maybe a)+bsourcePull (BufferedSource bs) =+    liftIO (I.readIORef bs) >>= goBuf+  where+    goBuf (OpenEmpty Closed) = liftIO $ I.writeIORef bs ClosedEmpty >> return Nothing+    goBuf (OpenEmpty (Open src _ a)) = do+        liftIO $ I.writeIORef bs $ OpenEmpty src+        return $ Just a+    goBuf (OpenEmpty (SourceM msrc _)) = msrc >>= goBuf . OpenEmpty+    goBuf ClosedEmpty = return Nothing+    goBuf (OpenFull src a) = do+        liftIO $ I.writeIORef bs (OpenEmpty src)+        return $ Just a+    goBuf (ClosedFull a) = do+        liftIO $ I.writeIORef bs ClosedEmpty+        return $ Just a -bsourceUnpull :: Resource m => BufferedSource m a -> Maybe a -> ResourceT m ()+bsourceUnpull :: MonadIO m => BufferedSource m a -> Maybe a -> m () bsourceUnpull _ Nothing = return () bsourceUnpull (BufferedSource ref) (Just a) = do-    buf <- readRef ref+    buf <- liftIO $ I.readIORef ref     case buf of-        OpenEmpty src -> writeRef ref (OpenFull src a)-        ClosedEmpty -> writeRef ref (ClosedFull a)+        OpenEmpty src -> liftIO $ I.writeIORef ref (OpenFull src a)+        ClosedEmpty -> liftIO $ I.writeIORef ref (ClosedFull a)         _ -> error $ "Invariant violated: bsourceUnpull called on full data"  -- | 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 'Source' was previously closed. ----- Since 0.2.0-bsourceClose :: Resource m => BufferedSource m a -> ResourceT m ()+-- Since 0.3.0+bsourceClose :: MonadIO m => BufferedSource m a -> m () bsourceClose (BufferedSource ref) = do-    buf <- readRef ref+    buf <- liftIO $ I.readIORef ref     case buf of         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+-- Since 0.3.0 data Flush a = Chunk a | Flush     deriving (Show, Eq, Ord) instance Functor Flush where

@@ -28,8 +28,9 @@ import Control.Monad (liftM) import Control.Monad.IO.Class (liftIO) import qualified System.IO as IO-import Control.Monad.Trans.Resource (withIO, release)+import Control.Monad.Trans.Resource (allocate, release) import Data.Word (Word8)+import Data.Monoid (mempty) #if CABAL_OS_WINDOWS import qualified System.Win32File as F #elif NO_HANDLES@@ -42,17 +43,17 @@ -- 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.2.0-openFile :: ResourceIO m+-- Since 0.3.0+openFile :: MonadResource m          => FilePath          -> IO.IOMode-         -> ResourceT m IO.Handle-openFile fp mode = fmap snd $ withIO (IO.openBinaryFile fp mode) IO.hClose+         -> m IO.Handle+openFile fp mode = liftM snd $ allocate (IO.openBinaryFile fp mode) IO.hClose  -- | Stream the contents of a file as binary data. ----- Since 0.2.0-sourceFile :: ResourceIO m+-- Since 0.3.0+sourceFile :: MonadResource m            => FilePath            -> Source m S.ByteString sourceFile fp =@@ -68,20 +69,20 @@ -- function will /not/ automatically close the @Handle@ when processing -- completes, since it did not acquire the @Handle@ in the first place. ----- Since 0.2.0-sourceHandle :: ResourceIO m+-- Since 0.3.0+sourceHandle :: MonadResource m              => IO.Handle              -> Source m S.ByteString sourceHandle h =     src   where-    src = Source pull close+    src = SourceM pull close      pull = do         bs <- liftIO (S.hGetSome h 4096)         if S.null bs             then return Closed-            else return $ Open src bs+            else return $ Open src close bs      close = return () @@ -90,8 +91,8 @@ -- a 'IO.Handle' (in read mode), so that it can open it only when needed -- and close it as soon as possible. ----- Since 0.2.0-sourceIOHandle :: ResourceIO m+-- Since 0.3.0+sourceIOHandle :: MonadResource m                => IO IO.Handle                -> Source m S.ByteString sourceIOHandle alloc = sourceIO alloc IO.hClose@@ -104,14 +105,14 @@ -- | Stream all incoming data to the given 'IO.Handle'. Note that this function -- will /not/ automatically close the @Handle@ when processing completes. ----- Since 0.2.0-sinkHandle :: ResourceIO m+-- Since 0.3.0+sinkHandle :: MonadResource m            => IO.Handle            -> Sink S.ByteString m () sinkHandle h =-    SinkData push close+    Processing push close   where-    push input = liftIO (S.hPut h input) >> return (Processing push close)+    push input = SinkM $ liftIO (S.hPut h input) >> return (Processing push close)     close = return ()  -- | An alternative to 'sinkHandle'.@@ -119,8 +120,8 @@ -- a 'IO.Handle' (in write mode), so that it can open it only when needed -- and close it as soon as possible. ----- Since 0.2.0-sinkIOHandle :: ResourceIO m+-- Since 0.3.0+sinkIOHandle :: MonadResource m              => IO IO.Handle              -> Sink S.ByteString m () sinkIOHandle alloc = sinkIO alloc IO.hClose@@ -130,23 +131,22 @@ -- | 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.2.0-sourceFileRange :: ResourceIO m+-- Since 0.3.0+sourceFileRange :: MonadResource m                 => FilePath                 -> Maybe Integer -- ^ Offset                 -> Maybe Integer -- ^ Maximum count                 -> Source m S.ByteString-sourceFileRange fp offset count = Source-    { sourcePull = do-        (key, handle) <- withIO (IO.openBinaryFile fp IO.ReadMode) IO.hClose+sourceFileRange fp offset count = SourceM+    (do+        (key, handle) <- allocate (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 -> pullUnlimited handle key-            Just c -> pullLimited c handle key-    , sourceClose = return ()-    }+            Just c -> pullLimited c handle key)+    (return ())   where     pullUnlimited handle key = do         bs <- liftIO $ S.hGetSome handle 4096@@ -155,11 +155,10 @@                 release key                 return Closed             else do-                let src = Source-                        { sourcePull = pullUnlimited handle key-                        , sourceClose = release key-                        }-                return $ Open src bs+                let src = SourceM+                        (pullUnlimited handle key)+                        (release key)+                return $ Open src (release key) bs      pullLimited c0 handle key = do         let c = fromInteger c0@@ -171,16 +170,15 @@                     release key                     return Closed                 else do-                    let src = Source-                            { sourcePull = pullLimited (toInteger c') handle key-                            , sourceClose = release key-                            }-                    return $ Open src bs+                    let src = SourceM+                            (pullLimited (toInteger c') handle key)+                            (release key)+                    return $ Open src (release key) bs  -- | Stream all incoming data to the given file. ----- Since 0.2.0-sinkFile :: ResourceIO m+-- Since 0.3.0+sinkFile :: MonadResource m          => FilePath          -> Sink S.ByteString m () sinkFile fp = sinkIOHandle (IO.openBinaryFile fp IO.WriteMode)@@ -188,8 +186,8 @@ -- | 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.2.0-conduitFile :: ResourceIO m+-- Since 0.3.0+conduitFile :: MonadResource m             => FilePath             -> Conduit S.ByteString m S.ByteString conduitFile fp = conduitIO@@ -204,8 +202,8 @@ -- sink. Note that this does /not/ ensure that all of those bytes are in fact -- consumed. ----- Since 0.2.0-isolate :: Resource m+-- Since 0.3.0+isolate :: Monad m         => Int         -> Conduit S.ByteString m S.ByteString isolate count0 = conduitState@@ -225,61 +223,66 @@  -- | Return the next byte from the stream, if available. ----- Since 0.2.0-head :: Resource m => Sink S.ByteString m (Maybe Word8)+-- Since 0.3.0+head :: Monad m => Sink S.ByteString m (Maybe Word8) head =-    SinkData push close+    Processing push close   where     push bs =         case S.uncons bs of-            Nothing -> return $ Processing push close-            Just (w, bs') -> do+            Nothing -> Processing push close+            Just (w, bs') ->                 let lo = if S.null bs' then Nothing else Just bs'-                return $ Done lo (Just w)+                 in Done lo (Just w)     close = return Nothing  -- | Return all bytes while the predicate returns @True@. ----- Since 0.2.0-takeWhile :: Resource m => (Word8 -> Bool) -> Conduit S.ByteString m S.ByteString+-- Since 0.3.0+takeWhile :: Monad m => (Word8 -> Bool) -> Conduit S.ByteString m S.ByteString takeWhile p =-    conduit+    NeedInput push close   where-    conduit = Conduit push close-    push bs = do-        let (x, y) = S.span p bs-        return $-            if S.null y-                then Producing conduit [x]-                else Finished (Just y) (if S.null x then [] else [x])-    close = return []+    push bs+        | S.null y =+            let r = NeedInput push close+             in if S.null x+                    then r+                    else HaveOutput r (return ()) x+        | otherwise =+            let f = Finished $ Just y+             in if S.null x+                    then f+                    else HaveOutput f (return ()) x+      where+        (x, y) = S.span p bs+    close = mempty  -- | Ignore all bytes while the predicate returns @True@. ----- Since 0.2.0-dropWhile :: Resource m => (Word8 -> Bool) -> Sink S.ByteString m ()+-- Since 0.3.0+dropWhile :: Monad m => (Word8 -> Bool) -> Sink S.ByteString m () dropWhile p =-    SinkData push close+    Processing push close   where-    push bs = do-        let bs' = S.dropWhile p bs-        return $-            if S.null bs'-                then Processing push close-                else Done (Just bs') ()+    push bs+        | S.null bs' = Processing push close+        | otherwise  = Done (Just bs') ()+      where+        bs' = S.dropWhile p bs     close = return ()  -- | Take the given number of bytes, if available. ----- Since 0.2.0-take :: Resource m => Int -> Sink S.ByteString m L.ByteString+-- Since 0.3.0+take :: Monad 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+-- Since 0.3.0+lines :: Monad m => Conduit S.ByteString m S.ByteString lines =     conduitState id push close   where

@@ -9,24 +9,21 @@ import Data.Conduit import System.IO.Unsafe (unsafeInterleaveIO) import Control.Monad.Trans.Control (MonadBaseControl, liftBaseOp_)-import Control.Monad.Trans.Resource (resourceActive)+import Control.Monad.Trans.Resource (MonadActive (monadActive))  -- | Use lazy I\/O to consume all elements from a @Source@. ----- Since 0.2.0-lazyConsume :: (Resource m, MonadBaseControl IO m) => Source m a -> ResourceT m [a]-lazyConsume src0 = do-    go src0-  where--    go src = liftBaseOp_ unsafeInterleaveIO $ do-        ra <- resourceActive-        if ra-            then do-                res <- sourcePull src-                case res of-                    Closed -> return []-                    Open src' x -> do-                        y <- go src'-                        return $ x : y-            else return []+-- This function relies on 'monadActive' to determine if the underlying monadic+-- state has been closed.+--+-- Since 0.3.0+lazyConsume :: (MonadBaseControl IO m, MonadActive m) => Source m a -> m [a]+lazyConsume Closed = return []+lazyConsume (Open src _ x) = do+    xs <- lazyConsume src+    return $ x : xs+lazyConsume (SourceM msrc _) = liftBaseOp_ unsafeInterleaveIO $ do+    a <- monadActive+    if a+        then msrc >>= lazyConsume+        else return []

@@ -41,63 +41,66 @@  import Prelude     ( ($), return, (==), (-), Int-    , (.), id, Maybe (..), fmap, Monad+    , (.), id, Maybe (..), Monad     , Bool (..)     , (>>)+    , flip+    , seq+    , otherwise     )-import qualified Prelude import Data.Conduit-import Control.Monad.Trans.Class (lift) import Data.Monoid (mempty)+import Control.Monad (liftM, liftM2)  -- | A strict left fold. ----- Since 0.2.0-fold :: Resource m+-- Since 0.3.0+fold :: Monad m      => (b -> a -> b)      -> b      -> Sink a m b-fold f accum0 = sinkState-    accum0-    (\accum input -> return (StateProcessing $ f accum input))-    return+fold f accum0 =+    go accum0+  where+    go accum = Processing (push accum) (return accum) +    push accum input =+        let accum' = f accum input+         in accum' `seq` go accum'+ -- | A monadic strict left fold. ----- Since 0.2.0-foldM :: Resource m+-- Since 0.3.0+foldM :: Monad m       => (b -> a -> m b)       -> b       -> Sink a m b foldM f accum0 = sinkState     accum0     (\accum input -> do-        accum' <- lift $ f accum input+        accum' <- f accum input         return $ StateProcessing accum'     )     return  -- | Apply the action to all values in the stream. ----- Since 0.2.0-mapM_ :: Resource m+-- Since 0.3.0+mapM_ :: Monad m       => (a -> m ())       -> Sink a m () mapM_ f =-    SinkData push close+    Processing push close   where-    push input = lift (f input) >> return (Processing push close)+    push input = SinkM $ f input >> return (Processing push close)     close = return ()  -- | Convert a list into a source. ----- Since 0.2.0-sourceList :: Resource m => [a] -> Source m a-sourceList l0 =-    sourceState l0 go-  where-    go [] = return StateClosed-    go (x:xs) = return $ StateOpen xs x+-- Since 0.3.0+sourceList :: Monad m => [a] -> Source m a+sourceList [] = Closed+sourceList (x:xs) = Open (sourceList xs) (return ()) x  -- | Ignore a certain number of values in the stream. This function is -- semantically equivalent to:@@ -107,22 +110,18 @@ -- However, @drop@ is more efficient as it does not need to hold values in -- memory. ----- Since 0.2.0-drop :: Resource m+-- Since 0.3.0+drop :: Monad m      => Int      -> Sink a m ()-drop count0 = sinkState-    count0-    push-    close+drop 0 = Processing (flip Done () . Just) (return ())+drop count =+    Processing push (return ())   where-    push 0 x = return $ StateDone (Just x) ()-    push count _ = do-        let count' = count - 1-        return $ if count' == 0-            then StateDone Nothing ()-            else StateProcessing count'-    close _ = return ()+    count' = count - 1+    push _+        | count' == 0 = Done Nothing ()+        | otherwise   = drop count'  -- | Take some values from the stream and return as a list. If you want to -- instead create a conduit that pipes data to another sink, see 'isolate'.@@ -130,98 +129,93 @@ -- -- > take i = isolate i =$ consume ----- Since 0.2.0-take :: Resource m+-- Since 0.3.0+take :: Monad m      => Int      -> Sink a m [a]-take count0 = sinkState-    (count0, id)-    push-    close+take count0 =+    go count0 id   where-    push (0, front) x = return (StateDone (Just x) (front []))-    push (count, front) x = do-        let count' = count - 1-            front' = front . (x:)-        return $ if count' == 0-                    then StateDone Nothing (front' [])-                    else StateProcessing (count', front')-    close (_, front) = return $ front []+    go count front = Processing (push count front) (return $ front []) +    push 0 front x = Done (Just x) (front [])+    push count front x+        | count' == 0 = Done Nothing (front [x])+        | otherwise   = Processing (push count' front') (return $ front' [])+      where+        count' = count - 1+        front' = front . (x:)+ -- | Take a single value from the stream, if available. ----- Since 0.2.0-head :: Resource m => Sink a m (Maybe a)+-- Since 0.3.0+head :: Monad m => Sink a m (Maybe a) head =-    SinkData push close+    Processing push close   where-    push x = return $ Done Nothing (Just x)+    push x = Done Nothing (Just x)     close = return Nothing  -- | Look at the next value in the stream, if available. This function will not -- change the state of the stream. ----- Since 0.2.0-peek :: Resource m => Sink a m (Maybe a)+-- Since 0.3.0+peek :: Monad m => Sink a m (Maybe a) peek =-    SinkData push close+    Processing push close   where-    push x = return $ Done (Just x) (Just x)+    push x = Done (Just x) (Just x)     close = return Nothing  -- | Apply a transformation to all values in a stream. ----- Since 0.2.0+-- Since 0.3.0 map :: Monad m => (a -> b) -> Conduit a m b map f =-    conduit+    NeedInput push close   where-    conduit = Conduit push close-    push = return . Producing conduit . return . f-    close = return []+    push i = HaveOutput (NeedInput push close) (return ()) (f i)+    close = mempty  -- | 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.2.0+-- Since 0.3.0 mapM :: Monad m => (a -> m b) -> Conduit a m b mapM f =-    conduit+    NeedInput push close   where-    conduit = Conduit push close-    push = fmap (Producing conduit . return) . lift . f-    close = return []+    push = flip ConduitM (return ()) . liftM (HaveOutput (NeedInput push close) (return ())) . f+    close = mempty  -- | Apply a transformation to all values in a stream, concatenating the output -- values. ----- Since 0.2.0+-- Since 0.3.0 concatMap :: Monad m => (a -> [b]) -> Conduit a m b concatMap f =-    conduit+    NeedInput push close   where-    conduit = Conduit push close-    push = return . Producing conduit . f-    close = return []+    push = haveMore (NeedInput push close) (return ()) . f+    close = mempty  -- | Apply a monadic transformation to all values in a stream, concatenating -- the output values. ----- Since 0.2.0+-- Since 0.3.0 concatMapM :: Monad m => (a -> m [b]) -> Conduit a m b concatMapM f =-    conduit+    NeedInput push close   where-    conduit = Conduit push close-    push = fmap (Producing conduit) . lift . f-    close = return []+    push = flip ConduitM (return ()) . liftM (haveMore (NeedInput push close) (return ())) . f+    close = mempty  -- | 'concatMap' with an accumulator. ----- Since 0.2.0-concatMapAccum :: Resource m => (a -> accum -> (accum, [b])) -> accum -> Conduit a m b+-- Since 0.3.0+concatMapAccum :: Monad 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@@ -230,11 +224,11 @@  -- | 'concatMapM' with an accumulator. ----- Since 0.2.0-concatMapAccumM :: Resource m => (a -> accum -> m (accum, [b])) -> accum -> Conduit a m b+-- Since 0.3.0+concatMapAccumM :: Monad 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)+    push state input = do (state', result) <- f input state                           return $ StateProducing state' result     close _ = return [] @@ -242,17 +236,18 @@ -- will pull all values into memory. For a lazy variant, see -- "Data.Conduit.Lazy". ----- Since 0.2.0-consume :: Resource m => Sink a m [a]-consume = sinkState-    id-    (\front input -> return (StateProcessing $ front . (input :)))-    (\front -> return $ front [])+-- Since 0.3.0+consume :: Monad m => Sink a m [a]+consume =+    go id+  where+    go front = Processing (push front) (return $ front [])+    push front x = go (front . (x:))  -- | Grouping input according to an equality function. ----- Since 0.2.0-groupBy :: Resource m => (a -> a -> Bool) -> Conduit a m [a]+-- Since 0.3.0+groupBy :: Monad m => (a -> a -> Bool) -> Conduit a m [a] groupBy f = conduitState     []     push@@ -278,8 +273,8 @@ -- >     someOtherSink -- >     ... ----- Since 0.2.0-isolate :: Resource m => Int -> Conduit a m a+-- Since 0.3.0+isolate :: Monad m => Int -> Conduit a m a isolate count0 = conduitState     count0     push@@ -297,46 +292,44 @@  -- | Keep only values in the stream passing a given predicate. ----- Since 0.2.0-filter :: Resource m => (a -> Bool) -> Conduit a m a+-- Since 0.3.0+filter :: Monad m => (a -> Bool) -> Conduit a m a filter f =-    conduit+    NeedInput push close   where-    conduit = Conduit push close-    push = return . Producing conduit . Prelude.filter f . return-    close = return []+    push i | f i = HaveOutput (NeedInput push close) (return ()) i+    push _       = NeedInput push close+    close = mempty  -- | Ignore the remainder of values in the source. Particularly useful when -- combined with 'isolate'. ----- Since 0.2.0-sinkNull :: Resource m => Sink a m ()+-- Since 0.3.0+sinkNull :: Monad m => Sink a m () sinkNull =-    SinkData push close+    Processing push close   where-    push _ = return $ Processing push close+    push _ = sinkNull     close = return ()  -- | A source that returns nothing. Note that this is just a type-restricted -- synonym for 'mempty'. ----- Since 0.2.0-sourceNull :: Resource m => Source m a+-- Since 0.3.0+sourceNull :: Monad m => Source m a sourceNull = mempty  -- | Combines two sources. The new source will stop producing once either --   source has been exhausted. ----- Since 0.2.2-zip :: Resource m => Source m a -> Source m b -> Source m (a, b)-zip sa sb = Source pull close-    where-        pull = do ra <- sourcePull sa-                  case ra of-                    Closed -> return Closed-                    Open ra' a -> do rb <- sourcePull sb-                                     case rb of-                                        Closed -> return Closed-                                        Open rb' b -> return $ Open (zip ra' rb') (a, b)-        close = sourceClose sa >> sourceClose sb-+-- Since 0.3.0+zip :: Monad m => Source m a -> Source m b -> Source m (a, b)+zip Closed Closed = Closed+zip Closed (Open _ close _) = SourceM (close >> return Closed) close+zip (Open _ close _) Closed = SourceM (close >> return Closed) close+zip Closed (SourceM _ close) = SourceM (close >> return Closed) close+zip (SourceM _ close) Closed = SourceM (close >> return Closed) close+zip (SourceM mx closex) (SourceM my closey) = SourceM (liftM2 zip mx my) (closex >> closey)+zip (SourceM mx closex) y@(Open _ closey _) = SourceM (liftM (\x -> zip x y) mx) (closex >> closey)+zip x@(Open _ closex _) (SourceM my closey) = SourceM (liftM (\y -> zip x y) my) (closex >> closey)+zip (Open srcx closex x) (Open srcy closey y) = Open (zip srcx srcy) (closex >> closey) (x, y)

@@ -29,7 +29,6 @@  import           Control.Arrow (first) import qualified Control.Exception as Exc-import           Control.Monad.Trans.Class (lift) import           Data.Bits ((.&.), (.|.), shiftL) import qualified Data.ByteString as B import qualified Data.ByteString.Char8 as B8@@ -43,11 +42,11 @@  import qualified Data.Conduit as C import qualified Data.Conduit.List as CL-import Control.Monad.Trans.Resource (ResourceThrow (..))+import Control.Monad.Trans.Resource (MonadThrow (..))  -- | A specific character encoding. ----- Since 0.2.0+-- Since 0.3.0 data Codec = Codec     { codecName :: T.Text     , codecEncode@@ -67,18 +66,18 @@ -- | 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.2.0-encode :: ResourceThrow m => Codec -> C.Conduit T.Text m B.ByteString+-- Since 0.3.0+encode :: MonadThrow m => Codec -> C.Conduit T.Text m B.ByteString encode codec = CL.mapM $ \t -> do     let (bs, mexc) = codecEncode codec t-    maybe (return bs) (resourceThrow . fst) mexc+    maybe (return bs) (monadThrow . fst) mexc   -- | 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.2.0-decode :: ResourceThrow m => Codec -> C.Conduit B.ByteString m T.Text+-- Since 0.3.0+decode :: MonadThrow m => Codec -> C.Conduit B.ByteString m T.Text decode codec = C.conduitState     Nothing     push@@ -92,11 +91,11 @@             Nothing -> return []             Just b                 | B.null b -> error "Data.Conduit.Text.decode: Received a null chunk"-                | otherwise -> lift $ resourceThrow $ DecodeException codec (B.head b)+                | otherwise -> monadThrow $ DecodeException codec (B.head b)      go' mb input = do -- FIXME This can be simplified significantly since input is now only a single BS         let bss = maybe id (:) mb [input]-        either (lift . resourceThrow) return $ go bss id+        either monadThrow return $ go bss id      go [] front = Right (Nothing, front [])     go (x:xs) front@@ -115,7 +114,7 @@         front' = front . (text:)  -- |--- Since 0.2.0+-- Since 0.3.0 data TextException = DecodeException Codec Word8                    | EncodeException Codec Char     deriving (Show, Typeable)@@ -148,7 +147,7 @@             Right _ -> Right B.empty)  -- |--- Since 0.2.0+-- Since 0.3.0 utf8 :: Codec utf8 = Codec name enc dec where     name = T.pack "UTF-8"@@ -181,7 +180,7 @@                     else decodeMore  -- |--- Since 0.2.0+-- Since 0.3.0 utf16_le :: Codec utf16_le = Codec name enc dec where     name = T.pack "UTF-16-LE"@@ -208,7 +207,7 @@         decodeAll = (TE.decodeUtf16LE bytes, B.empty)  -- |--- Since 0.2.0+-- Since 0.3.0 utf16_be :: Codec utf16_be = Codec name enc dec where     name = T.pack "UTF-16-BE"@@ -243,7 +242,7 @@     x = (fromIntegral x1 `shiftL` 8) .|. fromIntegral x0  -- |--- Since 0.2.0+-- Since 0.3.0 utf32_le :: Codec utf32_le = Codec name enc dec where     name = T.pack "UTF-32-LE"@@ -253,7 +252,7 @@         Nothing -> splitSlowly TE.decodeUtf32LE bs  -- |--- Since 0.2.0+-- Since 0.3.0 utf32_be :: Codec utf32_be = Codec name enc dec where     name = T.pack "UTF-32-BE"@@ -276,7 +275,7 @@         else B.splitAt lenToDecode bytes  -- |--- Since 0.2.0+-- Since 0.3.0 ascii :: Codec ascii = Codec name enc dec where     name = T.pack "ASCII"@@ -295,7 +294,7 @@             else Left (DecodeException ascii (B.head unsafe), unsafe)  -- |--- Since 0.2.0+-- Since 0.3.0 iso8859_1 :: Codec iso8859_1 = Codec name enc dec where     name = T.pack "ISO-8859-1"

@@ -1,49 +1,63 @@ -- | Defines the types for a conduit, which is a transformer of data. A conduit -- is almost always connected either left (to a source) or right (to a sink). module Data.Conduit.Types.Conduit-    ( ConduitResult (..)-    , Conduit (..)+    ( Conduit (..)     , ConduitPush     , ConduitClose     ) where -import Control.Monad.Trans.Resource (ResourceT) import Control.Monad (liftM)+import Data.Conduit.Types.Source --- | The value of the @conduitPush@ record.-type ConduitPush input m output = input -> ResourceT m (ConduitResult input m output)+-- | Pushing new data to a @Conduit@ produces a new @Conduit@.+--+-- Since 0.3.0+type ConduitPush input m output = input -> Conduit input m output --- | The value of the @conduitClose@ record.-type ConduitClose m output = ResourceT m [output]+-- | When closing a @Conduit@, it can produce a final stream of values.+--+-- Since 0.3.0+type ConduitClose m output = Source 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.+-- | A @Conduit@ allows data to be pushed to it, and for each new input, can+-- produce a stream of output values (possibly an empty stream). It can be+-- considered a hybrid of a @Sink@ and a @Source@. ----- The @Producing@ constructor provides a new @Conduit@ to be used in place of--- the previous one.+-- A @Conduit@ has four constructors, corresponding to four distinct states of+-- operation. ----- Since 0.2.0-data ConduitResult input m output =-    Producing (Conduit input m output) [output]-  | Finished (Maybe input) [output]+-- Since 0.3.0+data Conduit input m output =+    NeedInput (ConduitPush input m output) (ConduitClose m output)+    -- ^ Indicates that the @Conduit@ needs more input in order to produce+    -- output. It also provides an action to close the @Conduit@ early, for+    -- cases when there is no more input available, or when no more output is+    -- requested. Closing at this point returns a @Source@ to allow for either+    -- consuming or ignoring the new stream. -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)+  | HaveOutput (Conduit input m output) (m ()) output+    -- ^ Indicates that the @Conduit@ has more output available. It has three+    -- records: the next @Conduit@ to continue the stream, a close action for+    -- early termination, and the output currently available. Note that, unlike+    -- @NeedInput@, the close action here returns @()@ instead of @Source@. The+    -- reasoning is that @HaveOutput@ will only be closed early if no more+    -- output is requested, since no input is required. --- | A conduit has two operations: it can receive new input (a push), and can--- be closed.------ Since 0.2.0-data Conduit input m output = Conduit-    { conduitPush :: ConduitPush input m output-    , conduitClose :: ConduitClose m output-    }+  | Finished (Maybe input)+    -- ^ Indicates that no more output is available, and no more input may be+    -- sent. It provides an optional leftover input record. Note: It is a+    -- violation of @Conduit@'s invariants to return leftover output that was+    -- never consumed, similar to the invariants of a @Sink@. +  | ConduitM (m (Conduit input m output)) (m ())+    -- ^ Indicates that a monadic action must be taken to determine the next+    -- @Conduit@. It also provides an early close action. Like @HaveOutput@,+    -- this action returns @()@, since it should only be used when no more+    -- output is requested.+ instance Monad m => Functor (Conduit input m) where-    fmap f c = c-        { conduitPush = liftM (fmap f) . conduitPush c-        , conduitClose = liftM (fmap f) (conduitClose c)-        }+    fmap f (NeedInput p c) = NeedInput (fmap f . p) (fmap f c)+    fmap _ (Finished i) = Finished i+    fmap f (HaveOutput pull close output) = HaveOutput+        (fmap f pull) close (f output)+    fmap f (ConduitM mcon c) = ConduitM (liftM (fmap f) mcon) c

@@ -5,66 +5,26 @@ {-# LANGUAGE UndecidableInstances #-} -- | Defines the types for a sink, which is a consumer of data. module Data.Conduit.Types.Sink-    ( SinkResult (..)-    , Sink (..)+    ( 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, 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.------ The @Processing@ constructors provides updated push and close functions to--- be used in place of the original @Sink@.+-- | Push a value into a @Sink@ and get a new @Sink@ as a result. ----- 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.+-- Since 0.3.0+type SinkPush input m output = input -> Sink input m output -So the current three-constructor approach to `Sink` may not be as pretty, but-it enforce the invariants much better.--}+-- | Closing a @Sink@ returns the final output.+--+-- Since 0.3.0+type SinkClose m output = m output  -- | 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:@@ -75,74 +35,57 @@ -- * Some sinks will consume all available data and only produce a result at -- the \"end\" of a data stream (e.g., @sum@). ----- To allow for the first caveat, we have the 'SinkNoData' constructor. For the--- second, the 'SinkData' constructor has two records: one for receiving more--- input, and the other to indicate the end of a stream. Note that, at the end--- of a stream, some output is required. If a specific 'Sink' implementation--- cannot always produce output, this should be indicated in its return value,--- using something like a 'Maybe' or 'Either'.+-- Note that you can indicate any leftover data from processing via the @Maybe+-- input@ field of the @Done@ constructor. However, it is a violation of the+-- @Sink@ invariants to return leftover data when no input has been consumed.+-- Concrete, that means that a function like yield is invalid: --+-- > yield :: input -> Sink input m ()+-- > yield input = Done (Just input) ()+-- -- A @Sink@ should clean up any resources it has allocated when it returns a--- value, whether that be via @sinkPush@ or @sinkClose@.+-- value. ----- Since 0.2.0+-- Since 0.3.0 data Sink input m output =-    SinkNoData output-  | SinkData-        { 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))+    Processing (SinkPush input m output) (SinkClose m output) -- ^ Awaiting more input.+  | Done (Maybe input) output -- ^ Processing complete.+  | SinkM (m (Sink input m output)) -- ^ Perform some monadic action to continue.  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-        }-    fmap f (SinkLift msink) = SinkLift (liftM (fmap f) msink)+    fmap f (Processing push close) = Processing (fmap f . push) (liftM f close)+    fmap f (Done minput output) = Done minput (f output)+    fmap f (SinkM msink) = SinkM (liftM (fmap f) msink) -instance Resource m => Applicative (Sink input m) where+instance Monad m => Applicative (Sink input m) where     pure = return     (<*>) = ap -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)+instance Monad m => Monad (Sink input m) where+    return = Done Nothing+    Done Nothing x >>= f = f x+    Done (Just leftover) x >>= f =+        sinkPush (f x)       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 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--        close close' = do-            output <- close'-            closeHelper (f output)+        sinkPush (Processing push _) = push leftover+        sinkPush (Done Nothing output) = Done (Just leftover) output+        sinkPush (Done Just{} _) = error $ "Sink invariant violated: leftover input returned without any push"+        sinkPush (SinkM msink) = SinkM (liftM sinkPush msink)+    SinkM msink >>= f = SinkM (liftM (>>= f) msink)+    Processing push close >>= f = Processing+        (\input -> push input >>= f)+        (close >>= sinkClose . f) -        closeHelper (SinkNoData y) = return y-        closeHelper (SinkData _ closeF) = closeF-        closeHelper (SinkLift msink) = msink >>= closeHelper+sinkClose :: Monad m => Sink input m output -> m output+sinkClose (Done _ output) = return output+sinkClose (Processing _ close) = close+sinkClose (SinkM msink) = msink >>= sinkClose -instance (Resource m, Base m ~ base, Applicative base) => MonadBase base (Sink input m) where-    liftBase = lift . resourceLiftBase+instance MonadBase base m => MonadBase base (Sink input m) where+    liftBase = lift . liftBase  instance MonadTrans (Sink input) where-    lift = SinkLift . liftM SinkNoData . lift+    lift = SinkM . liftM return -instance (Resource m, MonadIO m) => MonadIO (Sink input m) where+instance MonadIO m => MonadIO (Sink input m) where     liftIO = lift . liftIO

@@ -1,71 +1,33 @@ {-# LANGUAGE FlexibleContexts #-} -- | Defines the types for a source, which is a producer of data. module Data.Conduit.Types.Source-    ( SourceResult (..)-    , Source (..)+    ( Source (..)     ) where -import Control.Monad.Trans.Resource import Data.Monoid (Monoid (..)) import Control.Monad (liftM) --- | Result of pulling from a source. Either a new piece of data (@Open@), or--- indicates that the source is now @Closed@.------ 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 Monad m => Functor (SourceResult m) where-    fmap f (Open p a) = Open (fmap f p) (f a)-    fmap _ Closed = Closed- -- | 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 @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.+-- @Source@. A @Source@ should free any resources it allocated when either it+-- returns @Closed@ or when it is explicitly closed (the second record on+-- either the @Open@ or @SourceM@ constructors). ----- Since 0.2.0-data Source m a = Source-    { sourcePull :: ResourceT m (SourceResult m a)-    , sourceClose :: ResourceT m ()-    }+-- Since 0.3.0+data Source m a =+    Open (Source m a) (m ()) a -- ^ A @Source@ providing more data. Provides records for the next @Source@ in the stream, a close action, and the data provided.+  | Closed -- ^ A @Source@ which has no more data available.+  | SourceM (m (Source m a)) (m ()) -- ^ Requires a monadic action to retrieve the next @Source@ in the stream. Second record allows you to close the @Source@.  instance Monad m => Functor (Source m) where-    fmap f src = src-        { sourcePull = liftM (fmap f) (sourcePull src)-        }--instance Resource m => Monoid (Source m a) where-    mempty = Source-        { sourcePull = return Closed-        , sourceClose = return ()-        }-    mappend a b = mconcat [a, b]-    mconcat [] = mempty-    mconcat (next0:rest0) =-        src next0 rest0-      where-        src next rest = Source (pull next rest) (close next rest)+    fmap f (Open next close a) = Open (fmap f next) close (f a)+    fmap _ Closed = Closed+    fmap f (SourceM msrc close) = SourceM (liftM (fmap f) msrc) close -        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-            sourceClose current+instance Monad m => Monoid (Source m a) where+    mempty = Closed+    mappend x Closed = x+    mappend Closed y = y+    mappend (Open next close a) y = Open (mappend next y) close a+    mappend (SourceM msrc close) y = SourceM (do+        src <- msrc+        return (mappend src y)) close

@@ -5,11 +5,13 @@ -- | Utilities for constructing and covnerting conduits. Please see -- "Data.Conduit.Types.Conduit" for more information on the base types. module Data.Conduit.Util.Conduit-    ( conduitState+    ( haveMore+    , conduitState     , ConduitStateResult (..)     , conduitIO     , ConduitIOResult (..)     , transConduit+    , conduitClose       -- *** Sequencing     , SequencedSink     , sequenceSink@@ -19,15 +21,29 @@  import Prelude hiding (sequence) import Control.Monad.Trans.Resource-import Control.Monad.Trans.Class import Data.Conduit.Types.Conduit import Data.Conduit.Types.Sink+import Data.Conduit.Types.Source+import Data.Conduit.Util.Source+import Data.Conduit.Util.Sink+import Control.Monad (liftM)+import Data.Monoid (mempty) +-- | A helper function for returning a list of values from a @Conduit@.+--+-- Since 0.3.0+haveMore :: Conduit a m b -- ^ The next @Conduit@ to return after the list has been exhausted.+         -> m () -- ^ A close action for early termination.+         -> [b] -- ^ The values to send down the stream.+         -> Conduit a m b+haveMore res _ [] = res+haveMore res close (x:xs) = HaveOutput (haveMore res close xs) close x+ -- | 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.2.0+-- Since 0.3.0 data ConduitStateResult state input output =     StateFinished (Maybe input) [output]   | StateProducing state [output]@@ -39,28 +55,35 @@ -- | Construct a 'Conduit' with some stateful functions. This function addresses -- threading the state value for you. ----- Since 0.2.0+-- Since 0.3.0 conduitState-    :: Resource m+    :: Monad m     => state -- ^ initial state-    -> (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.+    -> (state -> input -> m (ConduitStateResult state input output)) -- ^ Push function.+    -> (state -> m [output]) -- ^ Close function. The state need not be returned, since it will not be used again.     -> Conduit input m output conduitState state0 push0 close0 =-    Conduit (push state0) (close0 state0)+    NeedInput (push state0) (close 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+    push state input = ConduitM (liftM goRes' $ state `seq` push0 state input) (return ()) +    close state = SourceM (do+        os <- close0 state+        return $ fromList os) (return ())++    goRes' (StateFinished leftover output) = haveMore+        (Finished leftover)+        (return ())+        output+    goRes' (StateProducing state output) = haveMore+        (NeedInput (push state) (close state))+        (return ())+        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+-- Since 0.3.0 data ConduitIOResult input output =     IOFinished (Maybe input) [output]   | IOProducing [output]@@ -71,62 +94,68 @@  -- | Construct a 'Conduit'. ----- Since 0.2.0-conduitIO :: ResourceIO m+-- Since 0.3.0+conduitIO :: MonadResource m            => IO state -- ^ resource and/or state allocation            -> (state -> IO ()) -- ^ resource and/or state 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 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-    }+conduitIO alloc cleanup push0 close0 = NeedInput+    (\input -> flip ConduitM (return ()) $ do+        (key, state) <- allocate alloc cleanup+        push key state input)+    (SourceM (do+        (key, state) <- allocate alloc cleanup+        os <- close0 state+        release key+        return $ fromList os) (return ()))   where     push key state input = do-        res <- lift $ push0 state input+        res <- push0 state input         case res of-            IOProducing output -> return $ Producing-                (Conduit (push key state) (close key state))+            IOProducing output -> return $ haveMore+                (NeedInput (flip ConduitM (release key) . push key state) (close key state))+                (release key >> return ())                 output-            IOFinished a b -> do+            IOFinished leftover output -> do                 release key-                return $ Finished a b-    close key state = do-        output <- lift $ close0 state+                return $ haveMore+                    (Finished leftover)+                    (return ())+                    output++    close key state = SourceM (do+        output <- close0 state         release key-        return output+        return $ fromList output) (release key) +fromList :: Monad m => [a] -> Source m a+fromList [] = Closed+fromList (x:xs) = Open (fromList xs) (return ()) x+ -- | Transform the monad a 'Conduit' lives in. -- -- See @transSource@ for more information. ----- Since 0.2.0-transConduit :: (Monad m, Base m ~ Base n)+-- Since 0.3.0+transConduit :: Monad m              => (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)-              -> ConduitResult input m output-              -> ConduitResult input n output-transConduitPush _ (Finished a b) = Finished a b-transConduitPush f (Producing conduit output) = Producing-    (transConduit f conduit)+transConduit _ (Finished a) = Finished a+transConduit f (NeedInput push close) = NeedInput+    (transConduit f . push)+    (transSource f close)+transConduit f (HaveOutput pull close output) = HaveOutput+    (transConduit f pull)+    (f close)     output+transConduit f (ConduitM mcon close) = ConduitM (f (liftM (transConduit f) mcon)) (f close)  -- | Return value from a 'SequencedSink'. ----- Since 0.2.0+-- Since 0.3.0 data SequencedSinkResponse state input m output =     Emit state [output] -- ^ Set a new state, and emit some new output.   | Stop -- ^ End the conduit.@@ -136,78 +165,62 @@ -- to write higher-level code that takes advantage of existing conduits and -- sinks, and leverages a sink's monadic interface. ----- Since 0.2.0+-- Since 0.3.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 (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.2.0+-- Since 0.3.0 sequenceSink-    :: Resource m+    :: Monad m     => state -- ^ initial state     -> SequencedSink state input m output     -> Conduit input m output-sequenceSink state0 fsink = conduitState-    (SCNewState state0)-    (scPush id fsink)-    scClose+sequenceSink state0 fsink = NeedInput (scPush fsink $ fsink state0) mempty -- FIXME investigate if we can bypass getting input -goRes :: Resource m-      => SequencedSinkResponse state input m output-      -> Maybe input-      -> ([output] -> [output])-      -> SequencedSink state input m 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 $ StateProducing (SCNewState state) $ front output-goRes Stop minput front _ =-    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 :: Monad m+       => SequencedSink state input m output+       -> Sink input m (SequencedSinkResponse state input m output)+       -> ConduitPush input m output+scPush fsink (Processing pushI _) input = scGoRes fsink $ pushI input+scPush fsink (Done Nothing res) input = scGoRes fsink (Done (Just input) res)+scPush _ (Done Just{} _) _ = error "Invariant violated: Sink returned leftover without input"+scPush fsink (SinkM msink) input = ConduitM (liftM (\sink -> scPush fsink sink input) msink) (msink >>= sinkClose) -scPush :: Resource m-     => ([output] -> [output])-     -> SequencedSink state input m output-     -> SCState state input m output-     -> input-     -> ResourceT m (ConduitStateResult (SCState state input m output) input output)-scPush front fsink (SCNewState state) input =-    go (fsink state)+scGoRes :: Monad m+        => SequencedSink state input m output+        -> Sink input m (SequencedSinkResponse state input m output)+        -> Conduit input m output+scGoRes fsink (Done (Just leftover) (Emit state os)) = haveMore+    (scPush fsink (fsink state) leftover)+    (return ())+    os+scGoRes fsink (Done Nothing (Emit state os)) = haveMore+    (NeedInput p c)+    (return ())+    os   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-    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 push' close' -> return (StateProducing (SCSink push' close') $ front [])--scClose :: Monad m => SCState state inptu m output -> ResourceT m [output]-scClose (SCNewState _) = return []-scClose (SCConduit conduit) = conduitClose conduit-scClose (SCSink _ close) = do-    res <- close-    case res of-        Emit _ os -> return os-        Stop -> return []-        StartConduit c -> conduitClose c+    NeedInput p c = sequenceSink state fsink -- FIXME+scGoRes fsink (Processing pushI closeI) = NeedInput+    (scPush fsink (Processing pushI closeI))+    (SourceM (closeI >>= goRes) (closeI >> return ()))+  where+    goRes (Emit _ os) = return $ fromList os+    goRes Stop = return Closed+    goRes (StartConduit (NeedInput _ closeC)) = return closeC+    goRes (StartConduit (Finished _)) = return Closed+    goRes (StartConduit (ConduitM mcon _)) = mcon >>= goRes . StartConduit+    goRes (StartConduit HaveOutput{}) = error "scGoRes:goRes: StartConduit HaveOutput not supported yet"+scGoRes _ (Done mleftover Stop) = Finished mleftover+scGoRes _ (Done Nothing (StartConduit c)) = c+scGoRes _ (Done (Just leftover) (StartConduit (Finished Nothing))) = Finished (Just leftover)+scGoRes _ (Done Just{} (StartConduit (Finished Just{}))) = error "Invariant violated: conduit returns leftover without push"+scGoRes _ (Done (Just leftover) (StartConduit (NeedInput p _))) = p leftover+scGoRes _ (Done Just{} (StartConduit HaveOutput{})) = error "scGoRes: StartConduit HaveOutput not supported yet"+scGoRes fsink (Done mleftover (StartConduit (ConduitM mcon close))) =+    ConduitM (liftM (scGoRes fsink . Done mleftover . StartConduit) mcon) close+scGoRes fsink (SinkM msink) = ConduitM (liftM (scGoRes fsink) msink) (msink >>= sinkClose)  -- | Specialised version of 'sequenceSink' --@@ -215,32 +228,45 @@ -- @SinkNoData@ constructor. In other words, you probably don\'t want to do -- @sequence . return@. ----- Since 0.2.1-sequence :: Resource m => Sink input m output -> Conduit input m output-sequence (SinkData spush sclose) = Conduit (push spush) (close sclose)+-- Since 0.3.0+sequence :: Monad m => Sink input m output -> Conduit input m output+sequence (Processing spush0 sclose0) =+    NeedInput (push spush0) (close sclose0)   where-    push spush' input = do-        res <- spush' input+    push spush input = goRes $ spush input++    goRes res =         case res of             Processing spush'' sclose'' ->-                return $ Producing (Conduit (push spush'') (close sclose'')) []-            Done Nothing output ->-                return $ Producing (Conduit (push spush) (close sclose)) [output]-            Done (Just input') output -> do-                res' <- push spush input'-                case res' of-                    Producing conduit' output' ->-                        return $ Producing conduit' (output:output')-                    Finished _ _ -> error "impossible [sequence]"-    close sclose' = fmap (:[]) sclose'+                NeedInput (push spush'') (close sclose'')+            Done Nothing output -> HaveOutput+                (NeedInput (push spush0) (close sclose0))+                (return ())+                output+            Done (Just input') output -> HaveOutput+                (goRes $ spush0 input')+                (return ())+                output+            SinkM msink -> ConduitM (liftM goRes msink) (msink >>= sinkClose) -sequence (SinkNoData output) = Conduit-    { conduitPush = \input -> return $ Finished (Just input) (repeat output)-    , conduitClose = return $ repeat output-    }-sequence (SinkLift msink) = Conduit-    { conduitPush = \input -> do-        sink <- msink-        conduitPush (sequence sink) input-    , conduitClose = return []-    }+    close sclose = SourceM (do+        output <- sclose+        return $ Open Closed (return ()) output) (return ())++sequence (Done Nothing output) = NeedInput+    (\_input ->+        let x = HaveOutput x (return ()) output+         in x)+    (   let src = Open src (return ()) output+         in src)+sequence (Done Just{} _) = error "Invariant violated: sink returns leftover without push"+sequence (SinkM msink) = ConduitM (liftM sequence msink) (msink >>= sinkClose)++-- | Close a @Conduit@ early, discarding any output.+--+-- Since 0.3.0+conduitClose :: Monad m => Conduit input m output -> m ()+conduitClose (NeedInput _ c) = sourceClose c+conduitClose Finished{} = return ()+conduitClose (HaveOutput _ c _) = c+conduitClose (ConduitM _ c) = c

@@ -10,10 +10,10 @@     , sinkIO     , SinkIOResult (..)     , transSink+    , sinkClose     ) where  import Control.Monad.Trans.Resource-import Control.Monad.Trans.Class (lift) import Data.Conduit.Types.Sink import Control.Monad (liftM) @@ -21,7 +21,7 @@ -- It can either indicate that processing is done, or to continue with the -- updated state. ----- Since 0.2.0+-- Since 0.3.0 data SinkStateResult state input output =     StateDone (Maybe input) output   | StateProcessing state@@ -29,17 +29,17 @@ -- | Construct a 'Sink' with some stateful functions. This function addresses -- threading the state value for you. ----- Since 0.2.0+-- Since 0.3.0 sinkState-    :: Resource m+    :: Monad m     => state -- ^ initial state-    -> (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.+    -> (state -> input -> m (SinkStateResult state input output)) -- ^ push+    -> (state -> m output) -- ^ Close. Note that the state is not returned, as it is not needed.     -> Sink input m output sinkState state0 push0 close0 =-    SinkData (push state0) (close0 state0)+    Processing (push state0) (close0 state0)   where-    push state input = do+    push state input = SinkM $ do         res <- state `seq` push0 state input         case res of             StateProcessing state' -> return $ Processing (push state') (close0 state')@@ -48,39 +48,38 @@ -- | 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+-- Since 0.3.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.2.0-sinkIO :: ResourceIO m-        => IO state -- ^ resource and/or state allocation-        -> (state -> IO ()) -- ^ resource and/or state cleanup-        -> (state -> input -> m (SinkIOResult input output)) -- ^ push-        -> (state -> m output) -- ^ close-        -> Sink input m output-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-    }+-- Since 0.3.0+sinkIO :: MonadResource m+       => IO state -- ^ resource and/or state allocation+       -> (state -> IO ()) -- ^ resource and/or state cleanup+       -> (state -> input -> m (SinkIOResult input output)) -- ^ push+       -> (state -> m output) -- ^ close+       -> Sink input m output+sinkIO alloc cleanup push0 close0 = Processing+    (\input -> SinkM $ do+        (key, state) <- allocate alloc cleanup+        push key state input)+    (do+        (key, state) <- allocate alloc cleanup+        close key state)   where     push key state input = do-        res <- lift $ push0 state input+        res <- push0 state input         case res of             IODone a b -> do                 release key                 return $ Done a b             IOProcessing -> return $ Processing-                (push key state)+                (SinkM . push key state)                 (close key state)     close key state = do-        res <- lift $ close0 state+        res <- close0 state         release key         return res @@ -88,22 +87,19 @@ -- -- See @transSource@ for more information. ----- Since 0.2.0-transSink :: (Base m ~ Base n, Monad m)+-- Since 0.3.0+transSink :: Monad m           => (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)+transSink _ (Done a b) = Done a b+transSink f (Processing push close) = Processing (transSink f . push) (f close)+transSink f (SinkM msink) = SinkM (f (liftM (transSink f) msink)) -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)+-- | Close a @Sink@ if it is still open, discarding any output it produces.+--+-- Since 0.3.0+sinkClose :: Monad m => Sink input m output -> m ()+sinkClose (SinkM msink) = msink >>= sinkClose+sinkClose Done{} = return ()+sinkClose (Processing _ close) = close >> return ()

@@ -12,95 +12,90 @@     , sourceIO     , SourceIOResult (..)     , transSource+    , sourceClose     ) where  import Control.Monad.Trans.Resource-import Control.Monad.Trans.Class (lift) import Data.Conduit.Types.Source+import Control.Monad (liftM)  -- | The return value when pulling in the @sourceState@ function. Either -- indicates no more data, or the next value and an updated state. ----- Since 0.2.0+-- Since 0.3.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+-- Since 0.3.0 sourceState-    :: Resource m+    :: Monad m     => state -- ^ Initial state-    -> (state -> ResourceT m (SourceStateResult state output)) -- ^ Pull function+    -> (state -> m (SourceStateResult state output)) -- ^ Pull function     -> Source m output sourceState state0 pull0 =     src state0   where-    src state = Source (pull state) close+    src state = SourceM (pull state) (return ())      pull state = do         res <- pull0 state         return $ case res of-            StateOpen state' val -> Open (src state') val+            StateOpen state' val -> Open (src state') (return ()) val             StateClosed -> Closed -    close = return ()- -- | The return value when pulling in the @sourceIO@ function. Either indicates -- no more data, or the next value.+--+-- Since 0.3.0 data SourceIOResult output = IOOpen output | IOClosed  -- | Construct a 'Source' based on some IO actions for alloc/release. ----- Since 0.2.0-sourceIO :: ResourceIO m+-- Since 0.3.0+sourceIO :: MonadResource m           => IO state -- ^ resource and/or state allocation           -> (state -> IO ()) -- ^ resource and/or state cleanup-          -> (state -> m (SourceIOResult output)) -- ^ Pull function. Note that this need not explicitly perform any cleanup.+          -> (state -> m (SourceIOResult output)) -- ^ Pull function. Note that this should not perform any cleanup.           -> Source m output sourceIO alloc cleanup pull0 =-    Source-        { sourcePull = do-            (key, state) <- withIO alloc cleanup-            pull key state-        , sourceClose = return ()-        }+    SourceM (do+        (key, state) <- allocate alloc cleanup+        pull key state) (return ())   where-    src key state = Source (pull key state) (release key)+    src key state = SourceM (pull key state) (release key)      pull key state = do-        res <- lift $ pull0 state+        res <- pull0 state         case res of             IOClosed -> do                 release key                 return Closed-            IOOpen val -> return $ Open (src key state) val+            IOOpen val -> return $ Open (src key state) (release key) val  -- | A combination of 'sourceIO' and 'sourceState'. ----- Since 0.2.1-sourceStateIO :: ResourceIO m+-- Since 0.3.0+sourceStateIO :: MonadResource m               => IO state -- ^ resource and/or state allocation               -> (state -> IO ()) -- ^ resource and/or state cleanup               -> (state -> m (SourceStateResult state output)) -- ^ Pull function. Note that this need not explicitly perform any cleanup.               -> Source m output sourceStateIO alloc cleanup pull0 =-    Source-        { sourcePull = do-            (key, state) <- withIO alloc cleanup-            pull key state-        , sourceClose = return ()-        }+    SourceM (do+        (key, state) <- allocate alloc cleanup+        pull key state) (return ())   where-    src key state = Source (pull key state) (release key)+    src key state = SourceM (pull key state) (release key)      pull key state = do-        res <- lift $ pull0 state+        res <- pull0 state         case res of             StateClosed -> do                 release key                 return Closed-            StateOpen state' val -> return $ Open (src key state') val+            StateOpen state' val -> return $ Open (src key state') (release key) val  -- | Transform the monad a 'Source' lives in. --@@ -108,15 +103,19 @@ -- 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)+-- Since 0.3.0+transSource :: Monad m             => (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-    go2 (Open p a) = Open (transSource f p) a-    go2 Closed = Closed+transSource f (Open next close output) = Open (transSource f next) (f close) output+transSource _ Closed = Closed+transSource f (SourceM msrc close) = SourceM (f (liftM (transSource f) msrc)) (f close)++-- | Close a @Source@, regardless of its current state.+--+-- Since 0.3.0+sourceClose :: Monad m => Source m a -> m ()+sourceClose Closed = return ()+sourceClose (Open _ close _) = close+sourceClose (SourceM _ close) = close

@@ -1,10 +1,12 @@ Name:                conduit-Version:             0.2.2+Version:             0.3.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.3] ResourceT has been greatly simplified, specialized for IO, and moved into a separate package. Instead of hard-coding ResourceT into the conduit datatypes, they can now live around any monad. The Conduit datatype has been enhanced to better allow generation of streaming output. The SourceResult, SinkResult, and ConduitResult datatypes have been removed entirely. 	.     [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.     .@@ -40,7 +42,6 @@                        Data.Conduit.Text                        Data.Conduit.List                        Data.Conduit.Lazy-                       Control.Monad.Trans.Resource   Other-modules:       Data.Conduit.Types.Source                        Data.Conduit.Types.Sink                        Data.Conduit.Types.Conduit@@ -48,6 +49,7 @@                        Data.Conduit.Util.Sink                        Data.Conduit.Util.Conduit   Build-depends:       base                     >= 4.3          && < 5+                     , resourcet                >= 0.3          && < 0.4                      , lifted-base              >= 0.1          && < 0.2                      , transformers-base        >= 0.4.1        && < 0.5                      , monad-control            >= 0.3.1        && < 0.4@@ -72,6 +74,7 @@                    , bytestring                    , transformers                    , text+                   , resourcet     ghc-options:     -Wall  source-repository head

@@ -23,10 +23,11 @@ import qualified Data.Text as T import qualified Data.Text.Lazy as TL import qualified Data.Text.Lazy.Encoding as TLE-import Control.Monad.Trans.Resource (runExceptionT_, withIO, resourceForkIO)+import Control.Monad.Trans.Resource (runExceptionT_, allocate, resourceForkIO) import Control.Concurrent (threadDelay, killThread) import Control.Monad.IO.Class (liftIO) import Control.Applicative (pure, (<$>), (<*>))+import Data.Functor.Identity (runIdentity)  main :: IO () main = hspecX $ do@@ -53,7 +54,7 @@     describe "ResourceT" $ do         it "resourceForkIO" $ do             counter <- I.newIORef 0-            let w = withIO+            let w = allocate                         (I.atomicModifyIORef counter $ \i ->                             (i + 1, ()))                         (const $ I.atomicModifyIORef counter $ \i ->@@ -79,7 +80,7 @@             x <- runResourceT $ CL.sourceList [1..10] C.$$ CL.fold (+) (0 :: Int)             x @?= sum [1..10]         prop "is idempotent" $ \list ->-            (runST $ runResourceT $ CL.sourceList list C.$$ CL.fold (+) (0 :: Int))+            (runST $ CL.sourceList list C.$$ CL.fold (+) (0 :: Int))             == sum list      describe "Monoid instance for Source" $ do@@ -254,7 +255,7 @@      describe "sequence" $ do         it "simple sink" $ do-            let sumSink :: C.Resource m => C.Sink Int m Int+            let sumSink :: C.MonadResource m => C.Sink Int m Int                 sumSink = do                     ma <- CL.head                     case ma of@@ -267,7 +268,7 @@             res @?= [3, 7, 11, 15, 19, 11]          it "sink with unpull behaviour" $ do-            let sumSink :: C.Resource m => C.Sink Int m Int+            let sumSink :: C.MonadResource m => C.Sink Int m Int                 sumSink = do                     ma <- CL.head                     case ma of@@ -320,19 +321,19 @@      describe "text" $ do         let go enc tenc cenc = do-                prop (enc ++ " single chunk") $ \chars -> runST $ runExceptionT_ $ runResourceT $ do+                prop (enc ++ " single chunk") $ \chars -> runST $ runExceptionT_ $ do                     let tl = TL.pack chars                         lbs = tenc tl                         src = CL.sourceList $ L.toChunks lbs                     ts <- src C.$= CT.decode cenc C.$$ CL.consume                     return $ TL.fromChunks ts == tl-                prop (enc ++ " many chunks") $ \chars -> runST $ runExceptionT_ $ runResourceT $ do+                prop (enc ++ " many chunks") $ \chars -> runIdentity $ runExceptionT_ $ do                     let tl = TL.pack chars                         lbs = tenc tl                         src = mconcat $ map (CL.sourceList . return . S.singleton) $ L.unpack lbs                     ts <- src C.$= CT.decode cenc C.$$ CL.consume                     return $ TL.fromChunks ts == tl-                prop (enc ++ " encoding") $ \chars -> runST $ runExceptionT_ $ runResourceT $ do+                prop (enc ++ " encoding") $ \chars -> runIdentity $ runExceptionT_ $ do                     let tss = map T.pack chars                         lbs = tenc $ TL.fromChunks tss                         src = mconcat $ map (CL.sourceList . return) tss@@ -376,19 +377,19 @@          prop "works" $ \bss' ->             let bss = map S.pack bss'-             in runST $ runResourceT $+             in runIdentity $                 CL.sourceList bss C.$$ go (L.fromChunks bss)     describe "binary takeWhile" $ do         prop "works" $ \bss' ->             let bss = map S.pack bss'-             in runST $ runResourceT $ do+             in runIdentity $ do                 bss2 <- CL.sourceList bss C.$$ CB.takeWhile (>= 5) C.=$ CL.consume                 return $ L.fromChunks bss2 == L.takeWhile (>= 5) (L.fromChunks bss)      describe "binary dropWhile" $ do         prop "works" $ \bss' ->             let bss = map S.pack bss'-             in runST $ runResourceT $ do+             in runIdentity $ do                 bss2 <- CL.sourceList bss C.$$ do                     CB.dropWhile (< 5)                     CL.consume@@ -439,7 +440,7 @@             x @?= Just "foobarbazb"      describe "binary" $ do-        prop "lines" $ \bss' -> runST $ runResourceT $ do+        prop "lines" $ \bss' -> runIdentity $ do             let bss = map S.pack bss'                 bs = S.concat bss                 src = CL.sourceList bss