conduit 0.0.4 → 0.1.0
raw patch · 18 files changed
+3326/−3197 lines, 18 filessetup-changedPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
- Control.Monad.Trans.Resource: modifyRef :: Resource m => Ref (Base m) a -> (a -> (a, b)) -> ResourceT m b
- Control.Monad.Trans.Resource: modifyRef' :: HasRef m => Ref m a -> (a -> (a, b)) -> m b
- Data.Conduit: BufferedSource :: ResourceT m (SourceResult a) -> (a -> ResourceT m ()) -> ResourceT m () -> BufferedSource m a
- Data.Conduit: bsourcePull :: BufferedSource m a -> ResourceT m (SourceResult a)
- Data.Conduit: bsourceUnpull :: BufferedSource m a -> a -> ResourceT m ()
- Data.Conduit: class BufferSource s
- Data.Conduit: unsafeBufferSource :: (BufferSource s, Resource m) => s m a -> ResourceT m (BufferedSource m a)
+ Control.Monad.Trans.Resource: atomicModifyRef' :: HasRef m => Ref m a -> (a -> (a, b)) -> m b
+ Data.Conduit: class IsSource src
+ Data.Conduit: instance IsSource BufferedSource
+ Data.Conduit: instance IsSource Source
- Control.Monad.Trans.Resource: class Monad m => HasRef m where { type family Ref m :: * -> *; { modifyRef' sa f = do { a0 <- readRef' sa; let (a, b) = f a0; writeRef' sa a; return b } mask f = f id mask_ = mask . const try = liftM Right } }
+ Control.Monad.Trans.Resource: class Monad m => HasRef m where { type family Ref m :: * -> *; { atomicModifyRef' sa f = do { a0 <- readRef' sa; let (a, b) = f a0; writeRef' sa a; return b } mask f = f id mask_ = mask . const try = liftM Right } }
- Data.Conduit: ($$) :: (BufferSource bsrc, Resource m) => bsrc m a -> Sink a m b -> ResourceT m b
+ Data.Conduit: ($$) :: (IsSource src, Resource m) => src m a -> Sink a m b -> ResourceT m b
- Data.Conduit: ($=) :: (Resource m, BufferSource bsrc) => bsrc m a -> Conduit a m b -> Source m b
+ Data.Conduit: ($=) :: (IsSource src, Resource m) => src m a -> Conduit a m b -> Source m b
- Data.Conduit: bsourceClose :: BufferedSource m a -> ResourceT m ()
+ Data.Conduit: bsourceClose :: Resource m => BufferedSource m a -> ResourceT m ()
- Data.Conduit: bufferSource :: (BufferSource s, Resource m) => s m a -> ResourceT m (BufferedSource m a)
+ Data.Conduit: bufferSource :: Resource m => Source m a -> ResourceT m (BufferedSource m a)
- Data.Conduit: unbufferSource :: Monad m => BufferedSource m a -> Source m a
+ Data.Conduit: unbufferSource :: Resource m => BufferedSource m a -> Source m a
Files
- Control/Monad/Trans/Resource.hs +525/−530
- Data/Conduit.hs +484/−239
- Data/Conduit/Binary.hs +242/−242
- Data/Conduit/Lazy.hs +28/−28
- Data/Conduit/List.hs +290/−290
- Data/Conduit/Text.hs +317/−317
- Data/Conduit/Types/Conduit.hs +52/−52
- Data/Conduit/Types/Sink.hs +198/−198
- Data/Conduit/Types/Source.hs +128/−247
- Data/Conduit/Util/Conduit.hs +201/−201
- Data/Conduit/Util/Sink.hs +107/−107
- Data/Conduit/Util/Source.hs +106/−106
- LICENSE +30/−30
- Setup.lhs +7/−7
- System/PosixFile.hsc +57/−57
- System/Win32File.hsc +89/−89
- conduit.cabal +75/−68
- test/main.hs +390/−389
Control/Monad/Trans/Resource.hs view
@@ -1,530 +1,525 @@-{-# 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 - , modifyRef - , 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 (..) - ) where - -import Data.Typeable -import Data.IntMap (IntMap) -import qualified Data.IntMap as IntMap -import Control.Exception (SomeException) -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 #-} - --- | Modify a value in a reference. Note that, in the case of @IO@ stacks, this --- is an atomic action. -modifyRef :: Resource m => Ref (Base m) a -> (a -> (a, b)) -> ResourceT m b -modifyRef r = lift . resourceLiftBase . modifyRef' r -{-# INLINE modifyRef #-} - --- | 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 () - - modifyRef' :: Ref m a -> (a -> (a, b)) -> m b - modifyRef' 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' #-} - modifyRef' = I.atomicModifyIORef - {-# INLINE modifyRef' #-} - 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 ())) - --- | 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 = modifyRef' istate $ \(ReleaseMap key rf m) -> - ( ReleaseMap (key + 1) rf (IntMap.insert key rel m) - , ReleaseKey key - ) - --- | 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 <- modifyRef' 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 - ) - -stateAlloc :: HasRef m => Ref m (ReleaseMap m) -> m () -stateAlloc istate = do - modifyRef' istate $ \(ReleaseMap nk rf m) -> - (ReleaseMap nk (rf + 1) m, ()) - -stateCleanup :: HasRef m => Ref m (ReleaseMap m) -> m () -stateCleanup istate = mask_ $ do - (rf, m) <- modifyRef' istate $ \(ReleaseMap nk rf m) -> - (ReleaseMap nk (rf - 1) m, (rf - 1, m)) - if rf == minBound - then do - mapM_ (\x -> try x >> return ()) $ IntMap.elems m - -- Trigger an exception consistently for one race condition: - -- let's put an undefined value in the state. If somehow - -- another thread is still able to access it, at least we get - -- clearer error messages. - writeRef' istate $ error "Control.Monad.Trans.Resource.stateCleanup: There is a bug in the implementation. The mutable state is being accessed after cleanup. Please contact the maintainers." - else 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 a) - -> ResourceT m a - -> ResourceT n a -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)) +{-# 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 (..)+ ) where++import Data.Typeable+import Data.IntMap (IntMap)+import qualified Data.IntMap as IntMap+import Control.Exception (SomeException)+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 ()))++-- | 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 $ \(ReleaseMap key rf m) ->+ ( ReleaseMap (key + 1) rf (IntMap.insert key rel m)+ , ReleaseKey key+ )++-- | 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+ )++stateAlloc :: HasRef m => Ref m (ReleaseMap m) -> m ()+stateAlloc istate = do+ atomicModifyRef' istate $ \(ReleaseMap nk rf m) ->+ (ReleaseMap nk (rf + 1) m, ())++stateCleanup :: HasRef m => Ref m (ReleaseMap m) -> m ()+stateCleanup istate = mask_ $ do+ (rf, m) <- atomicModifyRef' istate $ \(ReleaseMap nk rf m) ->+ (ReleaseMap nk (rf - 1) m, (rf - 1, m))+ if rf == minBound+ then do+ mapM_ (\x -> try x >> return ()) $ IntMap.elems m+ -- Trigger an exception consistently for one race condition:+ -- let's put an undefined value in the state. If somehow+ -- another thread is still able to access it, at least we get+ -- clearer error messages.+ writeRef' istate $ error "Control.Monad.Trans.Resource.stateCleanup: There is a bug in the implementation. The mutable state is being accessed after cleanup. Please contact the maintainers."+ else 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 a)+ -> ResourceT m a+ -> ResourceT n a+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))
Data/Conduit.hs view
@@ -1,239 +1,484 @@-{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE MultiParamTypeClasses #-} -{-# LANGUAGE UndecidableInstances #-} -{-# LANGUAGE DeriveDataTypeable #-} --- | The main module, exporting types, utility functions, and fuse and connect --- operators. -module Data.Conduit - ( -- * Types - -- ** Source - module Data.Conduit.Types.Source - -- ** Sink - , module Data.Conduit.Types.Sink - -- ** Conduit - , module Data.Conduit.Types.Conduit - , -- * Connect/fuse operators - ($$) - , ($=) - , (=$) - , (=$=) - -- * Utility functions - -- ** Source - , module Data.Conduit.Util.Source - -- ** Sink - , module Data.Conduit.Util.Sink - -- ** Conduit - , module Data.Conduit.Util.Conduit - -- * Convenience re-exports - , ResourceT - , Resource (..) - , ResourceIO - , ResourceUnsafeIO - , runResourceT - , ResourceThrow (..) - ) where - -import Control.Monad.Trans.Resource -import Data.Conduit.Types.Source hiding (BufferSource (..)) -import Data.Conduit.Types.Source (BufferSource (bufferSource)) -import qualified Data.Conduit.Types.Source -import Data.Conduit.Util.Source -import Data.Conduit.Types.Sink -import Data.Conduit.Util.Sink -import Data.Conduit.Types.Conduit -import Data.Conduit.Util.Conduit - -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. --- --- 2. The sink returns @Done@, in which case any leftover input is returned via --- @bsourceUnpull@ the source is closed. --- --- 3. The source return @Closed@, in which case the sink is closed. --- --- Note that the input source is converted to a 'BufferedSource' via --- 'bufferSource'. As such, if the input to this function is itself a --- 'BufferedSource', the call to 'bsourceClose' will have no effect, as --- described in the comments on that instance. -($$) :: (BufferSource bsrc, Resource m) => bsrc m a -> Sink a m b -> ResourceT m b -bs' $$ Sink msink = do - sinkI <- msink - case sinkI of - SinkNoData output -> return output - SinkData push close -> do - bs <- Data.Conduit.Types.Source.unsafeBufferSource bs' - connect' bs push close - where - connect' bs push close = - loop - where - loop = do - res <- bsourcePull bs - case res of - Closed -> do - res' <- close - return res' - Open a -> do - mres <- push a - case mres of - Done leftover res' -> do - maybe (return ()) (bsourceUnpull bs) leftover - bsourceClose bs - return res' - Processing -> loop -{-# SPECIALIZE ($$) :: Resource m => Source m a -> Sink a m b -> ResourceT m b #-} - -data FuseLeftState a = FLClosed [a] | FLOpen [a] - -infixl 1 $= - --- | Left fuse, combining a source and a conduit together into a new source. -($=) :: (Resource m, BufferSource bsrc) - => bsrc m a - -> Conduit a m b - -> Source m b -bsrc' $= Conduit mc = Source $ do - istate <- newRef $ FLOpen [] -- still open, no buffer - bsrc <- Data.Conduit.Types.Source.unsafeBufferSource bsrc' - c <- mc - return $ PreparedSource - (pull istate bsrc c) - (close istate bsrc c) - where - pull istate bsrc c = do - state' <- readRef istate - case state' of - FLClosed [] -> return Closed - FLClosed (x:xs) -> do - writeRef istate $ FLClosed xs - return $ Open x - FLOpen (x:xs) -> do - writeRef istate $ FLOpen xs - return $ Open x - FLOpen [] -> do - mres <- bsourcePull bsrc - case mres of - Closed -> do - res <- conduitClose c - case res of - [] -> do - writeRef istate $ FLClosed [] - return Closed - x:xs -> do - writeRef istate $ FLClosed xs - return $ Open x - Open input -> do - res' <- conduitPush c input - case res' of - Producing [] -> pull istate bsrc c - Producing (x:xs) -> do - writeRef istate $ FLOpen xs - return $ Open x - Finished leftover output -> do - maybe (return ()) (bsourceUnpull bsrc) leftover - bsourceClose bsrc - case output of - [] -> do - writeRef istate $ FLClosed [] - return Closed - x:xs -> do - writeRef istate $ FLClosed xs - return $ Open x - close istate bsrc c = do - -- Invariant: sourceClose cannot be called twice, so we will assume - -- it is currently open. We could add a sanity check here. - writeRef istate $ FLClosed [] - _ignored <- conduitClose c - bsourceClose bsrc - -infixr 0 =$ - --- | Right fuse, combining a conduit and a sink together into a new sink. -(=$) :: Resource m => Conduit a m b -> Sink b m c -> Sink a m c -Conduit mc =$ Sink ms = Sink $ do - s <- ms - case s of - SinkData pushI closeI -> mc >>= go pushI closeI - SinkNoData mres -> return $ SinkNoData mres - where - go pushI closeI c = do - return SinkData - { sinkPush = \cinput -> do - res <- conduitPush c cinput - case res of - Producing sinput -> do - let push [] = return Processing - push (i:is) = do - mres <- pushI i - case mres of - Processing -> push is - Done _sleftover res' -> do - _ <- conduitClose c - return $ Done Nothing res' - push sinput - Finished cleftover sinput -> do - let push [] = closeI - push (i:is) = do - mres <- pushI i - case mres of - Processing -> push is - Done _sleftover res' -> return res' - res' <- push sinput - return $ Done cleftover res' - , sinkClose = do - sinput <- conduitClose c - let push [] = closeI - push (i:is) = do - mres <- pushI i - case mres of - Processing -> push is - Done _sleftover res' -> return res' - push sinput - } - -infixr 0 =$= - --- | Middle fuse, combining two conduits together into a new conduit. -(=$=) :: Resource m => Conduit a m b -> Conduit b m c -> Conduit a m c -Conduit outerM =$= Conduit innerM = Conduit $ do - outer <- outerM - inner <- innerM - return PreparedConduit - { conduitPush = \inputO -> do - res <- conduitPush outer inputO - case res of - Producing inputI -> do - let push [] front = return $ Producing $ front [] - push (i:is) front = do - resI <- conduitPush inner i - case resI of - Producing c -> push is (front . (c ++)) - Finished _leftover c -> do - _ <- conduitClose outer - return $ Finished Nothing $ front c - push inputI id - Finished leftoverO inputI -> do - c <- conduitPushClose inner inputI - return $ Finished leftoverO c - , conduitClose = do - b <- conduitClose outer - c <- conduitPushClose inner b - return c - } - --- | Push some data to a conduit, then close it if necessary. -conduitPushClose :: Monad m => PreparedConduit a m b -> [a] -> ResourceT m [b] -conduitPushClose c [] = conduitClose c -conduitPushClose c (input:rest) = do - res <- conduitPush c input - case res of - Finished _ b -> return b - Producing b -> do - b' <- conduitPushClose c rest - return $ b ++ b' +{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE DeriveDataTypeable #-}+-- | The main module, exporting types, utility functions, and fuse and connect+-- operators.+module Data.Conduit+ ( -- * Types+ -- ** Source+ module Data.Conduit.Types.Source+ -- *** Buffering+ , BufferedSource+ , bufferSource+ , unbufferSource+ , bsourceClose+ -- *** Unifying+ , IsSource+ -- ** Sink+ , module Data.Conduit.Types.Sink+ -- ** Conduit+ , module Data.Conduit.Types.Conduit+ , -- * Connect/fuse operators+ ($$)+ , ($=)+ , (=$)+ , (=$=)+ -- * Utility functions+ -- ** Source+ , module Data.Conduit.Util.Source+ -- ** Sink+ , module Data.Conduit.Util.Sink+ -- ** Conduit+ , module Data.Conduit.Util.Conduit+ -- * Convenience re-exports+ , ResourceT+ , Resource (..)+ , ResourceIO+ , ResourceUnsafeIO+ , runResourceT+ , ResourceThrow (..)+ ) where++import Control.Monad.Trans.Resource+import Data.Conduit.Types.Source+import Data.Conduit.Util.Source+import Data.Conduit.Types.Sink+import Data.Conduit.Util.Sink+import Data.Conduit.Types.Conduit+import Data.Conduit.Util.Conduit++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.+--+-- 2. The sink returns @Done@, in which case any leftover input is returned via+-- @bsourceUnpull@ the source is closed.+--+-- 3. The source return @Closed@, in which case the sink is closed.+--+-- Note that this function will automatically close any 'Source's, but will not+-- close any 'BufferedSource's, allowing them to be reused.+--+-- Since 0.0.0+($$) :: (IsSource src, Resource m) => src m a -> Sink a m b -> ResourceT m b+($$) = connect+{-# INLINE ($$) #-}++-- | A typeclass allowing us to unify operators for 'Source' and+-- 'BufferedSource'.+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++instance IsSource Source where+ connect = normalConnect+ {-# INLINE connect #-}+ fuseLeft = normalFuseLeft+ {-# INLINE fuseLeft #-}++instance IsSource BufferedSource where+ connect = bufferedConnect+ {-# INLINE connect #-}+ fuseLeft = bufferedFuseLeft+ {-# INLINE fuseLeft #-}++normalConnect :: Resource m => Source m a -> Sink a m b -> ResourceT m b+normalConnect (Source msrc) (Sink msink) = do+ sinkI <- msink+ case sinkI of+ SinkNoData output -> return output+ SinkData push close -> do+ src <- msrc+ connect' src push close+ where+ connect' src push close =+ loop+ where+ loop = do+ res <- sourcePull src+ case res of+ Closed -> do+ res' <- close+ return res'+ Open a -> do+ mres <- push a+ case mres of+ Done _leftover res' -> do+ sourceClose src+ return res'+ Processing -> loop++data FuseLeftState a = FLClosed [a] | FLOpen [a]++infixl 1 $=++-- | Left fuse, combining a source and a conduit together into a new source.+--+-- Since 0.0.0+($=) :: (IsSource src, Resource 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 (Source msrc) (Conduit mc) = Source $ do+ istate <- newRef $ FLOpen [] -- still open, no buffer+ src <- msrc+ c <- mc+ return $ PreparedSource+ (pull istate src c)+ (close istate src c)+ where+ pull istate src c = do+ state' <- readRef istate+ case state' of+ FLClosed [] -> return Closed+ FLClosed (x:xs) -> do+ writeRef istate $ FLClosed xs+ return $ Open x+ FLOpen (x:xs) -> do+ writeRef istate $ FLOpen xs+ return $ Open x+ FLOpen [] -> do+ mres <- sourcePull src+ case mres of+ Closed -> do+ res <- conduitClose c+ case res of+ [] -> do+ writeRef istate $ FLClosed []+ return Closed+ x:xs -> do+ writeRef istate $ FLClosed xs+ return $ Open x+ Open input -> do+ res' <- conduitPush c input+ case res' of+ Producing [] -> pull istate src c+ Producing (x:xs) -> do+ writeRef istate $ FLOpen xs+ return $ Open x+ Finished _leftover output -> do+ sourceClose src+ case output of+ [] -> do+ writeRef istate $ FLClosed []+ return Closed+ x:xs -> do+ writeRef istate $ FLClosed xs+ return $ Open x+ close istate src c = do+ -- Invariant: sourceClose cannot be called twice, so we will assume+ -- it is currently open. We could add a sanity check here.+ writeRef istate $ FLClosed []+ _ignored <- conduitClose c+ sourceClose src++infixr 0 =$++-- | Right fuse, combining a conduit and a sink together into a new sink.+--+-- Since 0.0.0+(=$) :: Resource m => Conduit a m b -> Sink b m c -> Sink a m c+Conduit mc =$ Sink ms = Sink $ do+ s <- ms+ case s of+ SinkData pushI closeI -> mc >>= go pushI closeI+ SinkNoData mres -> return $ SinkNoData mres+ where+ go pushI closeI c = do+ return SinkData+ { sinkPush = \cinput -> do+ res <- conduitPush c cinput+ case res of+ Producing sinput -> do+ let push [] = return Processing+ push (i:is) = do+ mres <- pushI i+ case mres of+ Processing -> push is+ Done _sleftover res' -> do+ _ <- conduitClose c+ return $ Done Nothing res'+ push sinput+ Finished cleftover sinput -> do+ let push [] = closeI+ push (i:is) = do+ mres <- pushI i+ case mres of+ Processing -> push is+ Done _sleftover res' -> return res'+ res' <- push sinput+ return $ Done cleftover res'+ , sinkClose = do+ sinput <- conduitClose c+ let push [] = closeI+ push (i:is) = do+ mres <- pushI i+ case mres of+ Processing -> push is+ Done _sleftover res' -> return res'+ push sinput+ }++infixr 0 =$=++-- | Middle fuse, combining two conduits together into a new conduit.+--+-- Since 0.0.0+(=$=) :: Resource m => Conduit a m b -> Conduit b m c -> Conduit a m c+Conduit outerM =$= Conduit innerM = Conduit $ do+ outer <- outerM+ inner <- innerM+ return PreparedConduit+ { conduitPush = \inputO -> do+ res <- conduitPush outer inputO+ case res of+ Producing inputI -> do+ let push [] front = return $ Producing $ front []+ push (i:is) front = do+ resI <- conduitPush inner i+ case resI of+ Producing c -> push is (front . (c ++))+ Finished _leftover c -> do+ _ <- conduitClose outer+ return $ Finished Nothing $ front c+ push inputI id+ Finished leftoverO inputI -> do+ c <- conduitPushClose inner inputI+ return $ Finished leftoverO c+ , conduitClose = do+ b <- conduitClose outer+ c <- conduitPushClose inner b+ return c+ }++-- | Push some data to a conduit, then close it if necessary.+conduitPushClose :: Monad m => PreparedConduit a m b -> [a] -> ResourceT m [b]+conduitPushClose c [] = conduitClose c+conduitPushClose c (input:rest) = do+ res <- conduitPush c input+ case res of+ Finished _ b -> return b+ Producing b -> do+ b' <- conduitPushClose c rest+ return $ b ++ b'++-- | When actually interacting with 'Source's, we usually want to be able to+-- buffer the output, in case any intermediate steps return leftover data. A+-- 'BufferedSource' allows for such buffering.+--+-- A 'BufferedSource', unlike a 'Source', is resumable, meaning it can be passed to+-- multiple 'Sink's without restarting.+--+-- Finally, a 'BufferedSource' relaxes one of the invariants of a 'Source':+-- pulling after an the source is closed is allowed.+--+-- A @BufferedSource@ is also known as a /resumable source/, in that it can be+-- called multiple times, and each time will provide new data. One caveat:+-- while the types will allow you to use the buffered source in multiple+-- threads, there is no guarantee that all @BufferedSource@s will handle this+-- correctly.+--+-- Since 0.0.0+data BufferedSource m a = BufferedSource+ { bsSource :: PreparedSource m a+ , bsBuffer :: Ref (Base m) (BSState a)+ }++data BSState a = ClosedEmpty | OpenEmpty | ClosedFull a | OpenFull a++-- | Prepare a 'Source' and initialize a buffer. Note that you should manually+-- call 'bsourceClose' when the 'BufferedSource' is no longer in use.+--+-- Since 0.0.0+bufferSource :: Resource m => Source m a -> ResourceT m (BufferedSource m a)+bufferSource (Source msrc) = do+ src <- msrc+ buf <- newRef OpenEmpty+ return $ BufferedSource src buf++-- | Turn a 'BufferedSource' into a 'Source'. Note that in general this will+-- mean your original 'BufferedSource' will be closed. Additionally, all+-- leftover data from usage of the returned @Source@ will be discarded. In+-- other words: this is a no-going-back move.+--+-- Note: @bufferSource@ . @unbufferSource@ is /not/ the identity function.+--+-- Since 0.0.1+unbufferSource :: Resource m+ => BufferedSource m a+ -> Source m a+unbufferSource (BufferedSource src bufRef) = Source $ do+ buf <- readRef bufRef+ case buf of+ OpenEmpty -> return src+ OpenFull a -> do+ isUsedRef <- newRef False+ return PreparedSource+ { sourcePull = do+ isUsed <- readRef isUsedRef+ if isUsed+ then sourcePull src+ else do+ writeRef isUsedRef True+ return $ Open a+ , sourceClose = sourceClose src+ }+ ClosedEmpty -> return PreparedSource+ -- Note: we could put some invariant checking in here if we wanted+ { sourcePull = return Closed+ , sourceClose = return ()+ }+ ClosedFull a -> do+ isUsedRef <- newRef False+ return PreparedSource+ { sourcePull = do+ isUsed <- readRef isUsedRef+ if isUsed+ then return Closed+ else do+ writeRef isUsedRef True+ return $ Open a+ , sourceClose = sourceClose src+ }++bufferedConnect :: Resource m => BufferedSource m a -> Sink a m b -> ResourceT m b+bufferedConnect bs (Sink msink) = do+ sinkI <- msink+ case sinkI of+ SinkNoData output -> return output+ SinkData push close -> do+ bsState <- readRef $ bsBuffer bs+ case bsState of+ ClosedEmpty -> close+ OpenEmpty -> connect' push close+ ClosedFull a -> do+ res <- push a+ case res of+ Done mleftover res' -> do+ writeRef (bsBuffer bs) $ maybe ClosedEmpty ClosedFull mleftover+ return res'+ Processing -> do+ writeRef (bsBuffer bs) ClosedEmpty+ close+ OpenFull a -> push a >>= onRes (connect' push close)+ where+ connect' push close =+ loop+ where+ loop = do+ res <- sourcePull $ bsSource bs+ case res of+ Closed -> do+ writeRef (bsBuffer bs) ClosedEmpty+ res' <- close+ return res'+ Open a -> push a >>= onRes loop+ onRes _ (Done mleftover res) = do+ writeRef (bsBuffer bs) (maybe OpenEmpty OpenFull mleftover)+ return res+ onRes loop Processing = loop++bufferedFuseLeft+ :: Resource m+ => BufferedSource m a+ -> Conduit a m b+ -> Source m b+bufferedFuseLeft bsrc (Conduit mc) = Source $ do+ istate <- newRef $ FLOpen [] -- still open, no buffer+ c <- mc+ return $ PreparedSource+ (pull istate c)+ (close istate c)+ where+ pull istate c = do+ state' <- readRef istate+ case state' of+ FLClosed [] -> return Closed+ FLClosed (x:xs) -> do+ writeRef istate $ FLClosed xs+ return $ Open x+ FLOpen (x:xs) -> do+ writeRef istate $ FLOpen xs+ return $ Open x+ FLOpen [] -> do+ mres <- bsourcePull bsrc+ case mres of+ Closed -> do+ res <- conduitClose c+ case res of+ [] -> do+ writeRef istate $ FLClosed []+ return Closed+ x:xs -> do+ writeRef istate $ FLClosed xs+ return $ Open x+ Open input -> do+ res' <- conduitPush c input+ case res' of+ Producing [] -> pull istate c+ Producing (x:xs) -> do+ writeRef istate $ FLOpen xs+ return $ Open x+ Finished leftover output -> do+ bsourceUnpull bsrc leftover+ case output of+ [] -> do+ writeRef istate $ FLClosed []+ return Closed+ x:xs -> do+ writeRef istate $ FLClosed xs+ return $ Open x+ close istate c = do+ writeRef istate $ FLClosed []+ _ignored <- conduitClose c+ return ()++bsourcePull :: Resource m => BufferedSource m a -> ResourceT m (SourceResult a)+bsourcePull (BufferedSource src bufRef) = do+ buf <- readRef bufRef+ case buf of+ OpenEmpty -> do+ res <- sourcePull src+ case res of+ Open _ -> return res+ Closed -> writeRef bufRef ClosedEmpty >> return Closed+ ClosedEmpty -> return Closed+ OpenFull a -> do+ writeRef bufRef OpenEmpty+ return $ Open a+ ClosedFull a -> do+ writeRef bufRef ClosedEmpty+ return $ Open a++bsourceUnpull :: Resource m => BufferedSource m a -> Maybe a -> ResourceT m ()+bsourceUnpull _ Nothing = return ()+bsourceUnpull (BufferedSource _ bufRef) (Just a) = do+ buf <- readRef bufRef+ case buf of+ OpenEmpty -> writeRef bufRef $ OpenFull a+ ClosedEmpty -> writeRef bufRef $ ClosedFull a+ _ -> error $ "Invariant violated: bsourceUnpull called on full data"++-- | Close the underlying 'PreparedSource' for the given 'BufferedSource'. Note+-- that this function can safely be called multiple times, as it will first+-- check if the 'PreparedSource' was previously closed.+--+-- Since 0.0.0+bsourceClose :: Resource m => BufferedSource m a -> ResourceT m ()+bsourceClose (BufferedSource src bufRef) = do+ buf <- readRef bufRef+ case buf of+ OpenEmpty -> sourceClose src+ OpenFull _ -> sourceClose src+ ClosedEmpty -> return ()+ ClosedFull _ -> return ()
Data/Conduit/Binary.hs view
@@ -1,242 +1,242 @@-{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE CPP #-} --- | Functions for interacting with bytes. -module Data.Conduit.Binary - ( sourceFile - , sourceHandle - , sourceFileRange - , sinkFile - , sinkHandle - , conduitFile - , isolate - , openFile - , head - , takeWhile - , dropWhile - , take - ) where - -import Prelude hiding (head, take, takeWhile, dropWhile) -import qualified Data.ByteString as S -import qualified Data.ByteString.Lazy as L -import Data.Conduit -import qualified Data.Conduit.List as CL -import Control.Exception (assert) -import Control.Monad (liftM) -import Control.Monad.IO.Class (liftIO) -import qualified System.IO as IO -import Control.Monad.Trans.Resource - ( withIO, release, newRef, readRef, writeRef - ) -import Data.Word (Word8) -#if CABAL_OS_WINDOWS -import qualified System.Win32File as F -#elif NO_HANDLES -import qualified System.PosixFile as F -#endif - --- | Open a file 'IO.Handle' safely by automatically registering a release --- action. --- --- While you are not required to call @hClose@ on the resulting handle, you --- should do so as early as possible to free scarce resources. --- --- Since 0.0.2 -openFile :: ResourceIO m - => FilePath - -> IO.IOMode - -> ResourceT m IO.Handle -openFile fp mode = fmap snd $ withIO (IO.openBinaryFile fp mode) IO.hClose - --- | Stream the contents of a file as binary data. --- --- Since 0.0.0 -sourceFile :: ResourceIO m - => FilePath - -> Source m S.ByteString -sourceFile fp = sourceIO -#if CABAL_OS_WINDOWS || NO_HANDLES - (F.openRead fp) - F.close - (liftIO . F.read) -#else - (IO.openBinaryFile fp IO.ReadMode) - IO.hClose - (\handle -> do - bs <- liftIO $ S.hGetSome handle 4096 - if S.null bs - then return Closed - else return $ Open bs) -#endif - --- | Stream the contents of a 'IO.Handle' as binary data. Note that this --- function will /not/ automatically close the @Handle@ when processing --- completes, since it did not acquire the @Handle@ in the first place. --- --- Since 0.0.2. -sourceHandle :: ResourceIO m - => IO.Handle - -> Source m S.ByteString -sourceHandle h = Source $ return $ PreparedSource - { sourcePull = do - bs <- liftIO (S.hGetSome h 4096) - if S.null bs - then return Closed - else return (Open bs) - , sourceClose = return () - } - --- | Stream all incoming data to the given 'IO.Handle'. Note that this function --- will /not/ automatically close the @Handle@ when processing completes. --- --- Since 0.0.2. -sinkHandle :: ResourceIO m - => IO.Handle - -> Sink S.ByteString m () -sinkHandle h = Sink $ return $ SinkData - { sinkPush = \input -> liftIO (S.hPut h input) >> return Processing - , sinkClose = return () - } - --- | Stream the contents of a file as binary data, starting from a certain --- offset and only consuming up to a certain number of bytes. --- --- Since 0.0.0 -sourceFileRange :: ResourceIO m - => FilePath - -> Maybe Integer -- ^ Offset - -> Maybe Integer -- ^ Maximum count - -> Source m S.ByteString -sourceFileRange fp offset count = Source $ do - (key, handle) <- withIO (IO.openBinaryFile fp IO.ReadMode) IO.hClose - case offset of - Nothing -> return () - Just off -> liftIO $ IO.hSeek handle IO.AbsoluteSeek off - pull <- - case count of - Nothing -> return $ pullUnlimited handle key - Just c -> do - ic <- newRef c - return $ pullLimited ic handle key - return PreparedSource - { sourcePull = pull - , sourceClose = release key - } - where - pullUnlimited handle key = do - bs <- liftIO $ S.hGetSome handle 4096 - if S.null bs - then do - release key - return Closed - else return $ Open bs - pullLimited ic handle key = do - c <- fmap fromInteger $ readRef ic - bs <- liftIO $ S.hGetSome handle (min c 4096) - let c' = c - S.length bs - assert (c' >= 0) $ - if S.null bs - then do - release key - return Closed - else do - writeRef ic $ toInteger c' - return $ Open bs - --- | Stream all incoming data to the given file. --- --- Since 0.0.0 -sinkFile :: ResourceIO m - => FilePath - -> Sink S.ByteString m () -sinkFile fp = sinkIO - (IO.openBinaryFile fp IO.WriteMode) - IO.hClose - (\handle bs -> liftIO (S.hPut handle bs) >> return Processing) - (const $ return ()) - --- | Stream the contents of the input to a file, and also send it along the --- pipeline. Similar in concept to the Unix command @tee@. --- --- Since 0.0.0 -conduitFile :: ResourceIO m - => FilePath - -> Conduit S.ByteString m S.ByteString -conduitFile fp = conduitIO - (IO.openBinaryFile fp IO.WriteMode) - IO.hClose - (\handle bs -> do - liftIO $ S.hPut handle bs - return $ Producing [bs]) - (const $ return []) - --- | Ensure that only up to the given number of bytes are consume by the inner --- sink. Note that this does /not/ ensure that all of those bytes are in fact --- consumed. --- --- Since 0.0.0 -isolate :: Resource m - => Int - -> Conduit S.ByteString m S.ByteString -isolate count0 = conduitState - count0 - push - close - where - push 0 bs = return (0, Finished (Just bs) []) - push count bs = do - let (a, b) = S.splitAt count bs - let count' = count - S.length a - return (count', - if count' == 0 - then Finished (if S.null b then Nothing else Just b) (if S.null a then [] else [a]) - else assert (S.null b) $ Producing [a]) - close _ = return [] - --- | Return the next byte from the stream, if available. --- --- Since 0.0.2 -head :: Resource m => Sink S.ByteString m (Maybe Word8) -head = Sink $ return $ SinkData - { sinkPush = \bs -> - case S.uncons bs of - Nothing -> return Processing - Just (w, bs') -> do - let lo = if S.null bs' then Nothing else Just bs' - return $ Done lo (Just w) - , sinkClose = return Nothing - } - --- | Return all bytes while the predicate returns @True@. --- --- Since 0.0.2 -takeWhile :: Resource m => (Word8 -> Bool) -> Conduit S.ByteString m S.ByteString -takeWhile p = Conduit $ return $ PreparedConduit - { conduitPush = \bs -> do - let (x, y) = S.span p bs - return $ - if S.null y - then Producing [x] - else Finished (Just y) (if S.null x then [] else [x]) - , conduitClose = return [] - } - --- | Ignore all bytes while the predicate returns @True@. --- --- Since 0.0.2 -dropWhile :: Resource m => (Word8 -> Bool) -> Sink S.ByteString m () -dropWhile p = Sink $ return $ SinkData - { sinkPush = \bs -> do - let bs' = S.dropWhile p bs - return $ - if S.null bs' - then Processing - else Done (Just bs') () - , sinkClose = return () - } - --- | Take the given number of bytes, if available. --- --- Since 0.0.3 -take :: Resource m => Int -> Sink S.ByteString m L.ByteString -take n = L.fromChunks `liftM` (isolate n =$ CL.consume) +{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE CPP #-}+-- | Functions for interacting with bytes.+module Data.Conduit.Binary+ ( sourceFile+ , sourceHandle+ , sourceFileRange+ , sinkFile+ , sinkHandle+ , conduitFile+ , isolate+ , openFile+ , head+ , takeWhile+ , dropWhile+ , take+ ) where++import Prelude hiding (head, take, takeWhile, dropWhile)+import qualified Data.ByteString as S+import qualified Data.ByteString.Lazy as L+import Data.Conduit+import qualified Data.Conduit.List as CL+import Control.Exception (assert)+import Control.Monad (liftM)+import Control.Monad.IO.Class (liftIO)+import qualified System.IO as IO+import Control.Monad.Trans.Resource+ ( withIO, release, newRef, readRef, writeRef+ )+import Data.Word (Word8)+#if CABAL_OS_WINDOWS+import qualified System.Win32File as F+#elif NO_HANDLES+import qualified System.PosixFile as F+#endif++-- | Open a file 'IO.Handle' safely by automatically registering a release+-- action.+--+-- While you are not required to call @hClose@ on the resulting handle, you+-- should do so as early as possible to free scarce resources.+--+-- Since 0.0.2+openFile :: ResourceIO m+ => FilePath+ -> IO.IOMode+ -> ResourceT m IO.Handle+openFile fp mode = fmap snd $ withIO (IO.openBinaryFile fp mode) IO.hClose++-- | Stream the contents of a file as binary data.+--+-- Since 0.0.0+sourceFile :: ResourceIO m+ => FilePath+ -> Source m S.ByteString+sourceFile fp = sourceIO+#if CABAL_OS_WINDOWS || NO_HANDLES+ (F.openRead fp)+ F.close+ (liftIO . F.read)+#else+ (IO.openBinaryFile fp IO.ReadMode)+ IO.hClose+ (\handle -> do+ bs <- liftIO $ S.hGetSome handle 4096+ if S.null bs+ then return Closed+ else return $ Open bs)+#endif++-- | Stream the contents of a 'IO.Handle' as binary data. Note that this+-- function will /not/ automatically close the @Handle@ when processing+-- completes, since it did not acquire the @Handle@ in the first place.+--+-- Since 0.0.2.+sourceHandle :: ResourceIO m+ => IO.Handle+ -> Source m S.ByteString+sourceHandle h = Source $ return $ PreparedSource+ { sourcePull = do+ bs <- liftIO (S.hGetSome h 4096)+ if S.null bs+ then return Closed+ else return (Open bs)+ , sourceClose = return ()+ }++-- | Stream all incoming data to the given 'IO.Handle'. Note that this function+-- will /not/ automatically close the @Handle@ when processing completes.+--+-- Since 0.0.2.+sinkHandle :: ResourceIO m+ => IO.Handle+ -> Sink S.ByteString m ()+sinkHandle h = Sink $ return $ SinkData+ { sinkPush = \input -> liftIO (S.hPut h input) >> return Processing+ , sinkClose = return ()+ }++-- | Stream the contents of a file as binary data, starting from a certain+-- offset and only consuming up to a certain number of bytes.+--+-- Since 0.0.0+sourceFileRange :: ResourceIO m+ => FilePath+ -> Maybe Integer -- ^ Offset+ -> Maybe Integer -- ^ Maximum count+ -> Source m S.ByteString+sourceFileRange fp offset count = Source $ do+ (key, handle) <- withIO (IO.openBinaryFile fp IO.ReadMode) IO.hClose+ case offset of+ Nothing -> return ()+ Just off -> liftIO $ IO.hSeek handle IO.AbsoluteSeek off+ pull <-+ case count of+ Nothing -> return $ pullUnlimited handle key+ Just c -> do+ ic <- newRef c+ return $ pullLimited ic handle key+ return PreparedSource+ { sourcePull = pull+ , sourceClose = release key+ }+ where+ pullUnlimited handle key = do+ bs <- liftIO $ S.hGetSome handle 4096+ if S.null bs+ then do+ release key+ return Closed+ else return $ Open bs+ pullLimited ic handle key = do+ c <- fmap fromInteger $ readRef ic+ bs <- liftIO $ S.hGetSome handle (min c 4096)+ let c' = c - S.length bs+ assert (c' >= 0) $+ if S.null bs+ then do+ release key+ return Closed+ else do+ writeRef ic $ toInteger c'+ return $ Open bs++-- | Stream all incoming data to the given file.+--+-- Since 0.0.0+sinkFile :: ResourceIO m+ => FilePath+ -> Sink S.ByteString m ()+sinkFile fp = sinkIO+ (IO.openBinaryFile fp IO.WriteMode)+ IO.hClose+ (\handle bs -> liftIO (S.hPut handle bs) >> return Processing)+ (const $ return ())++-- | Stream the contents of the input to a file, and also send it along the+-- pipeline. Similar in concept to the Unix command @tee@.+--+-- Since 0.0.0+conduitFile :: ResourceIO m+ => FilePath+ -> Conduit S.ByteString m S.ByteString+conduitFile fp = conduitIO+ (IO.openBinaryFile fp IO.WriteMode)+ IO.hClose+ (\handle bs -> do+ liftIO $ S.hPut handle bs+ return $ Producing [bs])+ (const $ return [])++-- | Ensure that only up to the given number of bytes are consume by the inner+-- sink. Note that this does /not/ ensure that all of those bytes are in fact+-- consumed.+--+-- Since 0.0.0+isolate :: Resource m+ => Int+ -> Conduit S.ByteString m S.ByteString+isolate count0 = conduitState+ count0+ push+ close+ where+ push 0 bs = return (0, Finished (Just bs) [])+ push count bs = do+ let (a, b) = S.splitAt count bs+ let count' = count - S.length a+ return (count',+ if count' == 0+ then Finished (if S.null b then Nothing else Just b) (if S.null a then [] else [a])+ else assert (S.null b) $ Producing [a])+ close _ = return []++-- | Return the next byte from the stream, if available.+--+-- Since 0.0.2+head :: Resource m => Sink S.ByteString m (Maybe Word8)+head = Sink $ return $ SinkData+ { sinkPush = \bs ->+ case S.uncons bs of+ Nothing -> return Processing+ Just (w, bs') -> do+ let lo = if S.null bs' then Nothing else Just bs'+ return $ Done lo (Just w)+ , sinkClose = return Nothing+ }++-- | Return all bytes while the predicate returns @True@.+--+-- Since 0.0.2+takeWhile :: Resource m => (Word8 -> Bool) -> Conduit S.ByteString m S.ByteString+takeWhile p = Conduit $ return $ PreparedConduit+ { conduitPush = \bs -> do+ let (x, y) = S.span p bs+ return $+ if S.null y+ then Producing [x]+ else Finished (Just y) (if S.null x then [] else [x])+ , conduitClose = return []+ }++-- | Ignore all bytes while the predicate returns @True@.+--+-- Since 0.0.2+dropWhile :: Resource m => (Word8 -> Bool) -> Sink S.ByteString m ()+dropWhile p = Sink $ return $ SinkData+ { sinkPush = \bs -> do+ let bs' = S.dropWhile p bs+ return $+ if S.null bs'+ then Processing+ else Done (Just bs') ()+ , sinkClose = return ()+ }++-- | Take the given number of bytes, if available.+--+-- Since 0.0.3+take :: Resource m => Int -> Sink S.ByteString m L.ByteString+take n = L.fromChunks `liftM` (isolate n =$ CL.consume)
Data/Conduit/Lazy.hs view
@@ -1,28 +1,28 @@-{-# LANGUAGE FlexibleContexts #-} --- | Use lazy I\/O for consuming the contents of a source. Warning: All normal --- warnings of lazy I\/O apply. However, if you consume the content within the --- ResourceT, you should be safe. -module Data.Conduit.Lazy - ( lazyConsume - ) where - -import Data.Conduit -import System.IO.Unsafe (unsafeInterleaveIO) -import Control.Monad.Trans.Control - --- | Use lazy I\/O to consume all elements from a @Source@. --- --- Since 0.0.0 -lazyConsume :: MonadBaseControl IO m => Source m a -> ResourceT m [a] -lazyConsume (Source msrc) = do - src <- msrc - go src - where - - go src = liftBaseOp_ unsafeInterleaveIO $ do - res <- sourcePull src - case res of - Closed -> return [] - Open x -> do - y <- go src - return $ x : y +{-# LANGUAGE FlexibleContexts #-}+-- | Use lazy I\/O for consuming the contents of a source. Warning: All normal+-- warnings of lazy I\/O apply. However, if you consume the content within the+-- ResourceT, you should be safe.+module Data.Conduit.Lazy+ ( lazyConsume+ ) where++import Data.Conduit+import System.IO.Unsafe (unsafeInterleaveIO)+import Control.Monad.Trans.Control++-- | Use lazy I\/O to consume all elements from a @Source@.+--+-- Since 0.0.0+lazyConsume :: MonadBaseControl IO m => Source m a -> ResourceT m [a]+lazyConsume (Source msrc) = do+ src <- msrc+ go src+ where++ go src = liftBaseOp_ unsafeInterleaveIO $ do+ res <- sourcePull src+ case res of+ Closed -> return []+ Open x -> do+ y <- go src+ return $ x : y
Data/Conduit/List.hs view
@@ -1,290 +1,290 @@-{-# LANGUAGE FlexibleContexts #-} --- | Higher-level functions to interact with the elements of a stream. Most of --- these are based on list functions. --- --- Note that these functions all deal with individual elements of a stream as a --- sort of \"black box\", where there is no introspection of the contained --- elements. Values such as @ByteString@ and @Text@ will likely need to be --- treated specially to deal with their contents properly (@Word8@ and @Char@, --- respectively). See the "Data.Conduit.Binary" and "Data.Conduit.Text" --- modules. -module Data.Conduit.List - ( -- * Sources - sourceList - , sourceNull - -- * Sinks - -- ** Pure - , fold - , take - , drop - , head - , peek - , consume - , sinkNull - -- ** Monadic - , foldM - , mapM_ - -- Conduits - -- ** Pure - , map - , concatMap - , groupBy - , isolate - , filter - -- ** Monadic - , mapM - , concatMapM - ) where - -import Prelude - ( ($), return, (==), (-), Int - , (.), id, Maybe (..), fmap, Monad - , Bool (..) - , (>>) - ) -import qualified Prelude -import Data.Conduit -import Control.Monad.Trans.Class (lift) -import Data.Monoid (mempty) - --- | A strict left fold. --- --- Since 0.0.0 -fold :: Resource m - => (b -> a -> b) - -> b - -> Sink a m b -fold f accum0 = sinkState - accum0 - (\accum input -> return (f accum input, Processing)) - return - --- | A monadic strict left fold. --- --- Since 0.0.0 -foldM :: Resource m - => (b -> a -> m b) - -> b - -> Sink a m b -foldM f accum0 = sinkState - accum0 - (\accum input -> do - accum' <- lift $ f accum input - return (accum', Processing) - ) - return - --- | Apply the action to all values in the stream. --- --- Since 0.0.0 -mapM_ :: Resource m - => (a -> m ()) - -> Sink a m () -mapM_ f = Sink $ return $ SinkData - (\input -> lift (f input) >> return Processing) - (return ()) - --- | Convert a list into a source. --- --- Since 0.0.0 -sourceList :: Resource m => [a] -> Source m a -sourceList l0 = - sourceState l0 go - where - go [] = return ([], Closed) - go (x:xs) = return (xs, Open x) - --- | Ignore a certain number of values in the stream. This function is --- semantically equivalent to: --- --- > drop i = take i >> return () --- --- However, @drop@ is more efficient as it does not need to hold values in --- memory. --- --- Since 0.0.0 -drop :: Resource m - => Int - -> Sink a m () -drop count0 = sinkState - count0 - push - close - where - push 0 x = return (0, Done (Just x) ()) - push count _ = do - let count' = count - 1 - return (count', if count' == 0 - then Done Nothing () - else Processing) - close _ = return () - --- | 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'. --- This function is semantically equivalent to: --- --- > take i = isolate i =$ consume --- --- Since 0.0.0 -take :: Resource m - => Int - -> Sink a m [a] -take count0 = sinkState - (count0, id) - push - close - where - push (0, front) x = return ((0, front), Done (Just x) (front [])) - push (count, front) x = do - let count' = count - 1 - front' = front . (x:) - res = if count' == 0 - then Done Nothing (front' []) - else Processing - return ((count', front'), res) - close (_, front) = return $ front [] - --- | Take a single value from the stream, if available. --- --- Since 0.0.0 -head :: Resource m => Sink a m (Maybe a) -head = - Sink $ return $ SinkData push close - where - push x = return $ 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.0.0 -peek :: Resource m => Sink a m (Maybe a) -peek = - Sink $ return $ SinkData push close - where - push x = return $ Done (Just x) (Just x) - close = return Nothing - --- | Apply a transformation to all values in a stream. --- --- Since 0.0.0 -map :: Monad m => (a -> b) -> Conduit a m b -map f = Conduit $ return $ PreparedConduit - { conduitPush = return . Producing . return . f - , conduitClose = return [] - } - --- | Apply a monadic transformation to all values in a stream. --- --- If you do not need the transformed values, and instead just want the monadic --- side-effects of running the action, see 'mapM_'. --- --- Since 0.0.0 -mapM :: Monad m => (a -> m b) -> Conduit a m b -mapM f = Conduit $ return $ PreparedConduit - { conduitPush = fmap (Producing . return) . lift . f - , conduitClose = return [] - } - --- | Apply a transformation to all values in a stream, concatenating the output --- values. --- --- Since 0.0.0 -concatMap :: Monad m => (a -> [b]) -> Conduit a m b -concatMap f = Conduit $ return $ PreparedConduit - { conduitPush = return . Producing . f - , conduitClose = return [] - } - --- | Apply a monadic transformation to all values in a stream, concatenating --- the output values. --- --- Since 0.0.0 -concatMapM :: Monad m => (a -> m [b]) -> Conduit a m b -concatMapM f = Conduit $ return $ PreparedConduit - { conduitPush = fmap Producing . lift . f - , conduitClose = return [] - } - --- | Consume all values from the stream and return as a list. Note that this --- will pull all values into memory. For a lazy variant, see --- "Data.Conduit.Lazy". --- --- Since 0.0.0 -consume :: Resource m => Sink a m [a] -consume = sinkState - id - (\front input -> return (front . (input :), Processing)) - (\front -> return $ front []) - --- | Grouping input according to an equality function. --- --- Since 0.0.2 -groupBy :: Resource m => (a -> a -> Bool) -> Conduit a m [a] -groupBy f = conduitState - [] - push - close - where - push [] v = return ([v], Producing []) - push s@(x:_) v = - if f x v then - return (v:s, Producing []) - else - return ([v], Producing [s]) - close s = return [s] - --- | Ensure that the inner sink consumes no more than the given number of --- values. Note this this does /not/ ensure that the sink consumes all of those --- values. To get the latter behavior, combine with 'sinkNull', e.g.: --- --- > src $$ do --- > x <- isolate count =$ do --- > x <- someSink --- > sinkNull --- > return x --- > someOtherSink --- > ... --- --- Since 0.0.0 -isolate :: Resource m => Int -> Conduit a m a -isolate count0 = conduitState - count0 - push - close - where - close _ = return [] - push count x = do - if count == 0 - then return (count, Finished (Just x) []) - else do - let count' = count - 1 - return (count', - if count' == 0 - then Finished Nothing [x] - else Producing [x]) - --- | Keep only values in the stream passing a given predicate. --- --- Since 0.0.0 -filter :: Resource m => (a -> Bool) -> Conduit a m a -filter f = Conduit $ return $ PreparedConduit - { conduitPush = return . Producing . Prelude.filter f . return - , conduitClose = return [] - } - --- | Ignore the remainder of values in the source. Particularly useful when --- combined with 'isolate'. --- --- Since 0.0.0 -sinkNull :: Resource m => Sink a m () -sinkNull = Sink $ return $ SinkData - (\_ -> return Processing) - (return ()) - --- | A source that returns nothing. Note that this is just a type-restricted --- synonym for 'mempty'. --- --- Since 0.0.4 -sourceNull :: Resource m => Source m a -sourceNull = mempty +{-# LANGUAGE FlexibleContexts #-}+-- | Higher-level functions to interact with the elements of a stream. Most of+-- these are based on list functions.+--+-- Note that these functions all deal with individual elements of a stream as a+-- sort of \"black box\", where there is no introspection of the contained+-- elements. Values such as @ByteString@ and @Text@ will likely need to be+-- treated specially to deal with their contents properly (@Word8@ and @Char@,+-- respectively). See the "Data.Conduit.Binary" and "Data.Conduit.Text"+-- modules.+module Data.Conduit.List+ ( -- * Sources+ sourceList+ , sourceNull+ -- * Sinks+ -- ** Pure+ , fold+ , take+ , drop+ , head+ , peek+ , consume+ , sinkNull+ -- ** Monadic+ , foldM+ , mapM_+ -- Conduits+ -- ** Pure+ , map+ , concatMap+ , groupBy+ , isolate+ , filter+ -- ** Monadic+ , mapM+ , concatMapM+ ) where++import Prelude+ ( ($), return, (==), (-), Int+ , (.), id, Maybe (..), fmap, Monad+ , Bool (..)+ , (>>)+ )+import qualified Prelude+import Data.Conduit+import Control.Monad.Trans.Class (lift)+import Data.Monoid (mempty)++-- | A strict left fold.+--+-- Since 0.0.0+fold :: Resource m+ => (b -> a -> b)+ -> b+ -> Sink a m b+fold f accum0 = sinkState+ accum0+ (\accum input -> return (f accum input, Processing))+ return++-- | A monadic strict left fold.+--+-- Since 0.0.0+foldM :: Resource m+ => (b -> a -> m b)+ -> b+ -> Sink a m b+foldM f accum0 = sinkState+ accum0+ (\accum input -> do+ accum' <- lift $ f accum input+ return (accum', Processing)+ )+ return++-- | Apply the action to all values in the stream.+--+-- Since 0.0.0+mapM_ :: Resource m+ => (a -> m ())+ -> Sink a m ()+mapM_ f = Sink $ return $ SinkData+ (\input -> lift (f input) >> return Processing)+ (return ())++-- | Convert a list into a source.+--+-- Since 0.0.0+sourceList :: Resource m => [a] -> Source m a+sourceList l0 =+ sourceState l0 go+ where+ go [] = return ([], Closed)+ go (x:xs) = return (xs, Open x)++-- | Ignore a certain number of values in the stream. This function is+-- semantically equivalent to:+--+-- > drop i = take i >> return ()+--+-- However, @drop@ is more efficient as it does not need to hold values in+-- memory.+--+-- Since 0.0.0+drop :: Resource m+ => Int+ -> Sink a m ()+drop count0 = sinkState+ count0+ push+ close+ where+ push 0 x = return (0, Done (Just x) ())+ push count _ = do+ let count' = count - 1+ return (count', if count' == 0+ then Done Nothing ()+ else Processing)+ close _ = return ()++-- | 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'.+-- This function is semantically equivalent to:+--+-- > take i = isolate i =$ consume+--+-- Since 0.0.0+take :: Resource m+ => Int+ -> Sink a m [a]+take count0 = sinkState+ (count0, id)+ push+ close+ where+ push (0, front) x = return ((0, front), Done (Just x) (front []))+ push (count, front) x = do+ let count' = count - 1+ front' = front . (x:)+ res = if count' == 0+ then Done Nothing (front' [])+ else Processing+ return ((count', front'), res)+ close (_, front) = return $ front []++-- | Take a single value from the stream, if available.+--+-- Since 0.0.0+head :: Resource m => Sink a m (Maybe a)+head =+ Sink $ return $ SinkData push close+ where+ push x = return $ 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.0.0+peek :: Resource m => Sink a m (Maybe a)+peek =+ Sink $ return $ SinkData push close+ where+ push x = return $ Done (Just x) (Just x)+ close = return Nothing++-- | Apply a transformation to all values in a stream.+--+-- Since 0.0.0+map :: Monad m => (a -> b) -> Conduit a m b+map f = Conduit $ return $ PreparedConduit+ { conduitPush = return . Producing . return . f+ , conduitClose = return []+ }++-- | Apply a monadic transformation to all values in a stream.+--+-- If you do not need the transformed values, and instead just want the monadic+-- side-effects of running the action, see 'mapM_'.+--+-- Since 0.0.0+mapM :: Monad m => (a -> m b) -> Conduit a m b+mapM f = Conduit $ return $ PreparedConduit+ { conduitPush = fmap (Producing . return) . lift . f+ , conduitClose = return []+ }++-- | Apply a transformation to all values in a stream, concatenating the output+-- values.+--+-- Since 0.0.0+concatMap :: Monad m => (a -> [b]) -> Conduit a m b+concatMap f = Conduit $ return $ PreparedConduit+ { conduitPush = return . Producing . f+ , conduitClose = return []+ }++-- | Apply a monadic transformation to all values in a stream, concatenating+-- the output values.+--+-- Since 0.0.0+concatMapM :: Monad m => (a -> m [b]) -> Conduit a m b+concatMapM f = Conduit $ return $ PreparedConduit+ { conduitPush = fmap Producing . lift . f+ , conduitClose = return []+ }++-- | Consume all values from the stream and return as a list. Note that this+-- will pull all values into memory. For a lazy variant, see+-- "Data.Conduit.Lazy".+--+-- Since 0.0.0+consume :: Resource m => Sink a m [a]+consume = sinkState+ id+ (\front input -> return (front . (input :), Processing))+ (\front -> return $ front [])++-- | Grouping input according to an equality function.+--+-- Since 0.0.2+groupBy :: Resource m => (a -> a -> Bool) -> Conduit a m [a]+groupBy f = conduitState+ []+ push+ close+ where+ push [] v = return ([v], Producing [])+ push s@(x:_) v =+ if f x v then+ return (v:s, Producing [])+ else+ return ([v], Producing [s])+ close s = return [s]++-- | Ensure that the inner sink consumes no more than the given number of+-- values. Note this this does /not/ ensure that the sink consumes all of those+-- values. To get the latter behavior, combine with 'sinkNull', e.g.:+--+-- > src $$ do+-- > x <- isolate count =$ do+-- > x <- someSink+-- > sinkNull+-- > return x+-- > someOtherSink+-- > ...+--+-- Since 0.0.0+isolate :: Resource m => Int -> Conduit a m a+isolate count0 = conduitState+ count0+ push+ close+ where+ close _ = return []+ push count x = do+ if count == 0+ then return (count, Finished (Just x) [])+ else do+ let count' = count - 1+ return (count',+ if count' == 0+ then Finished Nothing [x]+ else Producing [x])++-- | Keep only values in the stream passing a given predicate.+--+-- Since 0.0.0+filter :: Resource m => (a -> Bool) -> Conduit a m a+filter f = Conduit $ return $ PreparedConduit+ { conduitPush = return . Producing . Prelude.filter f . return+ , conduitClose = return []+ }++-- | Ignore the remainder of values in the source. Particularly useful when+-- combined with 'isolate'.+--+-- Since 0.0.0+sinkNull :: Resource m => Sink a m ()+sinkNull = Sink $ return $ SinkData+ (\_ -> return Processing)+ (return ())++-- | A source that returns nothing. Note that this is just a type-restricted+-- synonym for 'mempty'.+--+-- Since 0.0.4+sourceNull :: Resource m => Source m a+sourceNull = mempty
Data/Conduit/Text.hs view
@@ -1,317 +1,317 @@-{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE FlexibleContexts #-} --- | --- Copyright: 2011 Michael Snoyman, 2010-2011 John Millikin --- License: MIT --- --- Handle streams of text. --- --- Parts of this code were taken from enumerator and adapted for conduits. -module Data.Conduit.Text - ( - - -- * Text codecs - Codec - , encode - , decode - , utf8 - , utf16_le - , utf16_be - , utf32_le - , utf32_be - , ascii - , iso8859_1 - - ) where - -import qualified Prelude -import Prelude hiding (head, drop, takeWhile, lines, zip, zip3, zipWith, zipWith3) - -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 -import Data.Char (ord) -import Data.Maybe (catMaybes) -import qualified Data.Text as T -import qualified Data.Text.Encoding as TE -import Data.Word (Word8, Word16) -import System.IO.Unsafe (unsafePerformIO) -import Data.Typeable (Typeable) - -import qualified Data.Conduit as C -import qualified Data.Conduit.List as CL -import Control.Monad.Trans.Resource (ResourceThrow (..)) - --- | A specific character encoding. --- --- Since 0.0.0 -data Codec = Codec - { codecName :: T.Text - , codecEncode - :: T.Text - -> (B.ByteString, Maybe (TextException, T.Text)) - , codecDecode - :: B.ByteString - -> (T.Text, Either - (TextException, B.ByteString) - B.ByteString) - } - -instance Show Codec where - showsPrec d c = showParen (d > 10) $ - showString "Codec " . shows (codecName c) - --- | Convert text into bytes, using the provided codec. If the codec is --- not capable of representing an input character, an exception will be thrown. --- --- Since 0.0.0 -encode :: ResourceThrow 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 - - --- | Convert bytes into text, using the provided codec. If the codec is --- not capable of decoding an input byte sequence, an exception will be thrown. --- --- Since 0.0.0 -decode :: ResourceThrow m => Codec -> C.Conduit B.ByteString m T.Text -decode codec = C.conduitState - Nothing - push - close - where - push mb input = do - (mb', ts) <- go' mb input - return $ (mb', C.Producing ts) - close mb = - case mb of - Nothing -> return [] - Just b - | B.null b -> error "Data.Conduit.Text.decode: Received a null chunk" - | otherwise -> lift $ resourceThrow $ 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 - - go [] front = Right (Nothing, front []) - go (x:xs) front - | B.null x = go xs front - go (x:xs) front = - case extra of - Left (exc, _) -> Left exc - Right bs - | B.null bs -> go xs front' - | otherwise -> - case xs of - y:ys -> go (B.append bs y:ys) front' - [] -> Right (Just bs, front' []) - where - (text, extra) = codecDecode codec x - front' = front . (text:) - --- | --- Since 0.0.0 -data TextException = DecodeException Codec Word8 - | EncodeException Codec Char - deriving (Show, Typeable) -instance Exc.Exception TextException - -byteSplits :: B.ByteString - -> [(B.ByteString, B.ByteString)] -byteSplits bytes = loop (B.length bytes) where - loop 0 = [(B.empty, bytes)] - loop n = B.splitAt n bytes : loop (n - 1) - -splitSlowly :: (B.ByteString -> T.Text) - -> B.ByteString - -> (T.Text, Either - (TextException, B.ByteString) - B.ByteString) -splitSlowly dec bytes = valid where - valid = firstValid (Prelude.map decFirst splits) - splits = byteSplits bytes - firstValid = Prelude.head . catMaybes - tryDec = tryEvaluate . dec - - decFirst (a, b) = case tryDec a of - Left _ -> Nothing - Right text -> Just (text, case tryDec b of - Left exc -> Left (exc, b) - - -- this case shouldn't occur, since splitSlowly - -- is only called when parsing failed somewhere - Right _ -> Right B.empty) - --- | --- Since 0.0.0 -utf8 :: Codec -utf8 = Codec name enc dec where - name = T.pack "UTF-8" - enc text = (TE.encodeUtf8 text, Nothing) - dec bytes = case splitQuickly bytes of - Just (text, extra) -> (text, Right extra) - Nothing -> splitSlowly TE.decodeUtf8 bytes - - splitQuickly bytes = loop 0 >>= maybeDecode where - required x0 - | x0 .&. 0x80 == 0x00 = 1 - | x0 .&. 0xE0 == 0xC0 = 2 - | x0 .&. 0xF0 == 0xE0 = 3 - | x0 .&. 0xF8 == 0xF0 = 4 - - -- Invalid input; let Text figure it out - | otherwise = 0 - - maxN = B.length bytes - - loop n | n == maxN = Just (TE.decodeUtf8 bytes, B.empty) - loop n = let - req = required (B.index bytes n) - tooLong = first TE.decodeUtf8 (B.splitAt n bytes) - decodeMore = loop $! n + req - in if req == 0 - then Nothing - else if n + req > maxN - then Just tooLong - else decodeMore - --- | --- Since 0.0.0 -utf16_le :: Codec -utf16_le = Codec name enc dec where - name = T.pack "UTF-16-LE" - enc text = (TE.encodeUtf16LE text, Nothing) - dec bytes = case splitQuickly bytes of - Just (text, extra) -> (text, Right extra) - Nothing -> splitSlowly TE.decodeUtf16LE bytes - - splitQuickly bytes = maybeDecode (loop 0) where - maxN = B.length bytes - - loop n | n == maxN = decodeAll - | (n + 1) == maxN = decodeTo n - loop n = let - req = utf16Required - (B.index bytes n) - (B.index bytes (n + 1)) - decodeMore = loop $! n + req - in if n + req > maxN - then decodeTo n - else decodeMore - - decodeTo n = first TE.decodeUtf16LE (B.splitAt n bytes) - decodeAll = (TE.decodeUtf16LE bytes, B.empty) - --- | --- Since 0.0.0 -utf16_be :: Codec -utf16_be = Codec name enc dec where - name = T.pack "UTF-16-BE" - enc text = (TE.encodeUtf16BE text, Nothing) - dec bytes = case splitQuickly bytes of - Just (text, extra) -> (text, Right extra) - Nothing -> splitSlowly TE.decodeUtf16BE bytes - - splitQuickly bytes = maybeDecode (loop 0) where - maxN = B.length bytes - - loop n | n == maxN = decodeAll - | (n + 1) == maxN = decodeTo n - loop n = let - req = utf16Required - (B.index bytes (n + 1)) - (B.index bytes n) - decodeMore = loop $! n + req - in if n + req > maxN - then decodeTo n - else decodeMore - - decodeTo n = first TE.decodeUtf16BE (B.splitAt n bytes) - decodeAll = (TE.decodeUtf16BE bytes, B.empty) - -utf16Required :: Word8 -> Word8 -> Int -utf16Required x0 x1 = required where - required = if x >= 0xD800 && x <= 0xDBFF - then 4 - else 2 - x :: Word16 - x = (fromIntegral x1 `shiftL` 8) .|. fromIntegral x0 - --- | --- Since 0.0.0 -utf32_le :: Codec -utf32_le = Codec name enc dec where - name = T.pack "UTF-32-LE" - enc text = (TE.encodeUtf32LE text, Nothing) - dec bs = case utf32SplitBytes TE.decodeUtf32LE bs of - Just (text, extra) -> (text, Right extra) - Nothing -> splitSlowly TE.decodeUtf32LE bs - --- | --- Since 0.0.0 -utf32_be :: Codec -utf32_be = Codec name enc dec where - name = T.pack "UTF-32-BE" - enc text = (TE.encodeUtf32BE text, Nothing) - dec bs = case utf32SplitBytes TE.decodeUtf32BE bs of - Just (text, extra) -> (text, Right extra) - Nothing -> splitSlowly TE.decodeUtf32BE bs - -utf32SplitBytes :: (B.ByteString -> T.Text) - -> B.ByteString - -> Maybe (T.Text, B.ByteString) -utf32SplitBytes dec bytes = split where - split = maybeDecode (dec toDecode, extra) - len = B.length bytes - lenExtra = mod len 4 - - lenToDecode = len - lenExtra - (toDecode, extra) = if lenExtra == 0 - then (bytes, B.empty) - else B.splitAt lenToDecode bytes - --- | --- Since 0.0.0 -ascii :: Codec -ascii = Codec name enc dec where - name = T.pack "ASCII" - enc text = (bytes, extra) where - (safe, unsafe) = T.span (\c -> ord c <= 0x7F) text - bytes = B8.pack (T.unpack safe) - extra = if T.null unsafe - then Nothing - else Just (EncodeException ascii (T.head unsafe), unsafe) - - dec bytes = (text, extra) where - (safe, unsafe) = B.span (<= 0x7F) bytes - text = T.pack (B8.unpack safe) - extra = if B.null unsafe - then Right B.empty - else Left (DecodeException ascii (B.head unsafe), unsafe) - --- | --- Since 0.0.0 -iso8859_1 :: Codec -iso8859_1 = Codec name enc dec where - name = T.pack "ISO-8859-1" - enc text = (bytes, extra) where - (safe, unsafe) = T.span (\c -> ord c <= 0xFF) text - bytes = B8.pack (T.unpack safe) - extra = if T.null unsafe - then Nothing - else Just (EncodeException iso8859_1 (T.head unsafe), unsafe) - - dec bytes = (T.pack (B8.unpack bytes), Right B.empty) - -tryEvaluate :: a -> Either TextException a -tryEvaluate = unsafePerformIO . Exc.try . Exc.evaluate - -maybeDecode:: (a, b) -> Maybe (a, b) -maybeDecode (a, b) = case tryEvaluate a of - Left _ -> Nothing - Right _ -> Just (a, b) +{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE FlexibleContexts #-}+-- |+-- Copyright: 2011 Michael Snoyman, 2010-2011 John Millikin+-- License: MIT+--+-- Handle streams of text.+--+-- Parts of this code were taken from enumerator and adapted for conduits.+module Data.Conduit.Text+ (++ -- * Text codecs+ Codec+ , encode+ , decode+ , utf8+ , utf16_le+ , utf16_be+ , utf32_le+ , utf32_be+ , ascii+ , iso8859_1++ ) where++import qualified Prelude+import Prelude hiding (head, drop, takeWhile, lines, zip, zip3, zipWith, zipWith3)++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+import Data.Char (ord)+import Data.Maybe (catMaybes)+import qualified Data.Text as T+import qualified Data.Text.Encoding as TE+import Data.Word (Word8, Word16)+import System.IO.Unsafe (unsafePerformIO)+import Data.Typeable (Typeable)++import qualified Data.Conduit as C+import qualified Data.Conduit.List as CL+import Control.Monad.Trans.Resource (ResourceThrow (..))++-- | A specific character encoding.+--+-- Since 0.0.0+data Codec = Codec+ { codecName :: T.Text+ , codecEncode+ :: T.Text+ -> (B.ByteString, Maybe (TextException, T.Text))+ , codecDecode+ :: B.ByteString+ -> (T.Text, Either+ (TextException, B.ByteString)+ B.ByteString)+ }++instance Show Codec where+ showsPrec d c = showParen (d > 10) $+ showString "Codec " . shows (codecName c)++-- | Convert text into bytes, using the provided codec. If the codec is+-- not capable of representing an input character, an exception will be thrown.+--+-- Since 0.0.0+encode :: ResourceThrow 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+++-- | Convert bytes into text, using the provided codec. If the codec is+-- not capable of decoding an input byte sequence, an exception will be thrown.+--+-- Since 0.0.0+decode :: ResourceThrow m => Codec -> C.Conduit B.ByteString m T.Text+decode codec = C.conduitState+ Nothing+ push+ close+ where+ push mb input = do+ (mb', ts) <- go' mb input+ return $ (mb', C.Producing ts)+ close mb =+ case mb of+ Nothing -> return []+ Just b+ | B.null b -> error "Data.Conduit.Text.decode: Received a null chunk"+ | otherwise -> lift $ resourceThrow $ 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++ go [] front = Right (Nothing, front [])+ go (x:xs) front+ | B.null x = go xs front+ go (x:xs) front =+ case extra of+ Left (exc, _) -> Left exc+ Right bs+ | B.null bs -> go xs front'+ | otherwise ->+ case xs of+ y:ys -> go (B.append bs y:ys) front'+ [] -> Right (Just bs, front' [])+ where+ (text, extra) = codecDecode codec x+ front' = front . (text:)++-- |+-- Since 0.0.0+data TextException = DecodeException Codec Word8+ | EncodeException Codec Char+ deriving (Show, Typeable)+instance Exc.Exception TextException++byteSplits :: B.ByteString+ -> [(B.ByteString, B.ByteString)]+byteSplits bytes = loop (B.length bytes) where+ loop 0 = [(B.empty, bytes)]+ loop n = B.splitAt n bytes : loop (n - 1)++splitSlowly :: (B.ByteString -> T.Text)+ -> B.ByteString+ -> (T.Text, Either+ (TextException, B.ByteString)+ B.ByteString)+splitSlowly dec bytes = valid where+ valid = firstValid (Prelude.map decFirst splits)+ splits = byteSplits bytes+ firstValid = Prelude.head . catMaybes+ tryDec = tryEvaluate . dec++ decFirst (a, b) = case tryDec a of+ Left _ -> Nothing+ Right text -> Just (text, case tryDec b of+ Left exc -> Left (exc, b)++ -- this case shouldn't occur, since splitSlowly+ -- is only called when parsing failed somewhere+ Right _ -> Right B.empty)++-- |+-- Since 0.0.0+utf8 :: Codec+utf8 = Codec name enc dec where+ name = T.pack "UTF-8"+ enc text = (TE.encodeUtf8 text, Nothing)+ dec bytes = case splitQuickly bytes of+ Just (text, extra) -> (text, Right extra)+ Nothing -> splitSlowly TE.decodeUtf8 bytes++ splitQuickly bytes = loop 0 >>= maybeDecode where+ required x0+ | x0 .&. 0x80 == 0x00 = 1+ | x0 .&. 0xE0 == 0xC0 = 2+ | x0 .&. 0xF0 == 0xE0 = 3+ | x0 .&. 0xF8 == 0xF0 = 4++ -- Invalid input; let Text figure it out+ | otherwise = 0++ maxN = B.length bytes++ loop n | n == maxN = Just (TE.decodeUtf8 bytes, B.empty)+ loop n = let+ req = required (B.index bytes n)+ tooLong = first TE.decodeUtf8 (B.splitAt n bytes)+ decodeMore = loop $! n + req+ in if req == 0+ then Nothing+ else if n + req > maxN+ then Just tooLong+ else decodeMore++-- |+-- Since 0.0.0+utf16_le :: Codec+utf16_le = Codec name enc dec where+ name = T.pack "UTF-16-LE"+ enc text = (TE.encodeUtf16LE text, Nothing)+ dec bytes = case splitQuickly bytes of+ Just (text, extra) -> (text, Right extra)+ Nothing -> splitSlowly TE.decodeUtf16LE bytes++ splitQuickly bytes = maybeDecode (loop 0) where+ maxN = B.length bytes++ loop n | n == maxN = decodeAll+ | (n + 1) == maxN = decodeTo n+ loop n = let+ req = utf16Required+ (B.index bytes n)+ (B.index bytes (n + 1))+ decodeMore = loop $! n + req+ in if n + req > maxN+ then decodeTo n+ else decodeMore++ decodeTo n = first TE.decodeUtf16LE (B.splitAt n bytes)+ decodeAll = (TE.decodeUtf16LE bytes, B.empty)++-- |+-- Since 0.0.0+utf16_be :: Codec+utf16_be = Codec name enc dec where+ name = T.pack "UTF-16-BE"+ enc text = (TE.encodeUtf16BE text, Nothing)+ dec bytes = case splitQuickly bytes of+ Just (text, extra) -> (text, Right extra)+ Nothing -> splitSlowly TE.decodeUtf16BE bytes++ splitQuickly bytes = maybeDecode (loop 0) where+ maxN = B.length bytes++ loop n | n == maxN = decodeAll+ | (n + 1) == maxN = decodeTo n+ loop n = let+ req = utf16Required+ (B.index bytes (n + 1))+ (B.index bytes n)+ decodeMore = loop $! n + req+ in if n + req > maxN+ then decodeTo n+ else decodeMore++ decodeTo n = first TE.decodeUtf16BE (B.splitAt n bytes)+ decodeAll = (TE.decodeUtf16BE bytes, B.empty)++utf16Required :: Word8 -> Word8 -> Int+utf16Required x0 x1 = required where+ required = if x >= 0xD800 && x <= 0xDBFF+ then 4+ else 2+ x :: Word16+ x = (fromIntegral x1 `shiftL` 8) .|. fromIntegral x0++-- |+-- Since 0.0.0+utf32_le :: Codec+utf32_le = Codec name enc dec where+ name = T.pack "UTF-32-LE"+ enc text = (TE.encodeUtf32LE text, Nothing)+ dec bs = case utf32SplitBytes TE.decodeUtf32LE bs of+ Just (text, extra) -> (text, Right extra)+ Nothing -> splitSlowly TE.decodeUtf32LE bs++-- |+-- Since 0.0.0+utf32_be :: Codec+utf32_be = Codec name enc dec where+ name = T.pack "UTF-32-BE"+ enc text = (TE.encodeUtf32BE text, Nothing)+ dec bs = case utf32SplitBytes TE.decodeUtf32BE bs of+ Just (text, extra) -> (text, Right extra)+ Nothing -> splitSlowly TE.decodeUtf32BE bs++utf32SplitBytes :: (B.ByteString -> T.Text)+ -> B.ByteString+ -> Maybe (T.Text, B.ByteString)+utf32SplitBytes dec bytes = split where+ split = maybeDecode (dec toDecode, extra)+ len = B.length bytes+ lenExtra = mod len 4++ lenToDecode = len - lenExtra+ (toDecode, extra) = if lenExtra == 0+ then (bytes, B.empty)+ else B.splitAt lenToDecode bytes++-- |+-- Since 0.0.0+ascii :: Codec+ascii = Codec name enc dec where+ name = T.pack "ASCII"+ enc text = (bytes, extra) where+ (safe, unsafe) = T.span (\c -> ord c <= 0x7F) text+ bytes = B8.pack (T.unpack safe)+ extra = if T.null unsafe+ then Nothing+ else Just (EncodeException ascii (T.head unsafe), unsafe)++ dec bytes = (text, extra) where+ (safe, unsafe) = B.span (<= 0x7F) bytes+ text = T.pack (B8.unpack safe)+ extra = if B.null unsafe+ then Right B.empty+ else Left (DecodeException ascii (B.head unsafe), unsafe)++-- |+-- Since 0.0.0+iso8859_1 :: Codec+iso8859_1 = Codec name enc dec where+ name = T.pack "ISO-8859-1"+ enc text = (bytes, extra) where+ (safe, unsafe) = T.span (\c -> ord c <= 0xFF) text+ bytes = B8.pack (T.unpack safe)+ extra = if T.null unsafe+ then Nothing+ else Just (EncodeException iso8859_1 (T.head unsafe), unsafe)++ dec bytes = (T.pack (B8.unpack bytes), Right B.empty)++tryEvaluate :: a -> Either TextException a+tryEvaluate = unsafePerformIO . Exc.try . Exc.evaluate++maybeDecode:: (a, b) -> Maybe (a, b)+maybeDecode (a, b) = case tryEvaluate a of+ Left _ -> Nothing+ Right _ -> Just (a, b)
Data/Conduit/Types/Conduit.hs view
@@ -1,52 +1,52 @@--- | 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 (..) - , PreparedConduit (..) - , Conduit (..) - ) where - -import Control.Monad.Trans.Resource (ResourceT) -import Control.Monad (liftM) - --- | When data is pushed to a @Conduit@, it may either indicate that it is --- still producing output and provide some, or indicate that it is finished --- producing output, in which case it returns optional leftover input and some --- final output. --- --- Since 0.0.0 -data ConduitResult input output = Producing [output] | Finished (Maybe input) [output] - -instance Functor (ConduitResult input) where - fmap f (Producing o) = Producing (fmap f o) - fmap f (Finished i o) = Finished i (fmap f o) - --- | A conduit has two operations: it can receive new input (a push), and can --- be closed. --- --- Invariants: --- --- * Neither a push nor close may be performed after a conduit returns a --- 'Finished' from a push, or after a close is performed. --- --- Since 0.0.0 -data PreparedConduit input m output = PreparedConduit - { conduitPush :: input -> ResourceT m (ConduitResult input output) - , conduitClose :: ResourceT m [output] - } - -instance Monad m => Functor (PreparedConduit input m) where - fmap f c = c - { conduitPush = liftM (fmap f) . conduitPush c - , conduitClose = liftM (fmap f) (conduitClose c) - } - --- | A monadic action generating a 'PreparedConduit'. See @Source@ and @Sink@ --- for more motivation. --- --- Since 0.0.0 -newtype Conduit input m output = - Conduit { prepareConduit :: ResourceT m (PreparedConduit input m output) } - -instance Monad m => Functor (Conduit input m) where - fmap f (Conduit mc) = Conduit (liftM (fmap f) mc) +-- | 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 (..)+ , PreparedConduit (..)+ , Conduit (..)+ ) where++import Control.Monad.Trans.Resource (ResourceT)+import Control.Monad (liftM)++-- | When data is pushed to a @Conduit@, it may either indicate that it is+-- still producing output and provide some, or indicate that it is finished+-- producing output, in which case it returns optional leftover input and some+-- final output.+--+-- Since 0.0.0+data ConduitResult input output = Producing [output] | Finished (Maybe input) [output]++instance Functor (ConduitResult input) where+ fmap f (Producing o) = Producing (fmap f o)+ fmap f (Finished i o) = Finished i (fmap f o)++-- | A conduit has two operations: it can receive new input (a push), and can+-- be closed.+--+-- Invariants:+--+-- * Neither a push nor close may be performed after a conduit returns a+-- 'Finished' from a push, or after a close is performed.+--+-- Since 0.0.0+data PreparedConduit input m output = PreparedConduit+ { conduitPush :: input -> ResourceT m (ConduitResult input output)+ , conduitClose :: ResourceT m [output]+ }++instance Monad m => Functor (PreparedConduit input m) where+ fmap f c = c+ { conduitPush = liftM (fmap f) . conduitPush c+ , conduitClose = liftM (fmap f) (conduitClose c)+ }++-- | A monadic action generating a 'PreparedConduit'. See @Source@ and @Sink@+-- for more motivation.+--+-- Since 0.0.0+newtype Conduit input m output =+ Conduit { prepareConduit :: ResourceT m (PreparedConduit input m output) }++instance Monad m => Functor (Conduit input m) where+ fmap f (Conduit mc) = Conduit (liftM (fmap f) mc)
Data/Conduit/Types/Sink.hs view
@@ -1,198 +1,198 @@-{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE MultiParamTypeClasses #-} -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE TypeFamilies #-} -{-# LANGUAGE UndecidableInstances #-} --- | Defines the types for a sink, which is a consumer of data. -module Data.Conduit.Types.Sink - ( SinkResult (..) - , PreparedSink (..) - , Sink (..) - ) where - -import Control.Monad.Trans.Resource -import Control.Monad.Trans.Class (MonadTrans (lift)) -import Control.Monad.IO.Class (MonadIO (liftIO)) -import Control.Monad (liftM) -import Control.Applicative (Applicative (..)) -import Control.Monad.Base (MonadBase (liftBase)) - --- | A @Sink@ ultimately returns a single output value. Each time data is --- pushed to it, a @Sink@ may indicate that it is still processing data, or --- that it is done, in which case it returns some optional leftover input and --- an output value. --- --- Since 0.0.0 -data SinkResult input output = Processing | Done (Maybe input) output -instance Functor (SinkResult input) where - fmap _ Processing = Processing - fmap f (Done input output) = Done input (f 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: --- --- * Some sinks do not actually require any data to produce an output. This is --- included with a sink in order to allow for a 'Monad' instance. --- --- * 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'. --- --- Invariants: --- --- * After a 'PreparedSink' produces a result (either via 'sinkPush' or --- 'sinkClose'), neither of those two functions may be called on the @Sink@ --- again. --- --- * If a @Sink@ needs to clean up any resources (e.g., close a file handle), --- it must do so whenever it returns a result, either via @sinkPush@ or --- @sinkClose@. Note that, due to usage of @ResourceT@, this is merely an --- optimization. --- --- Since 0.0.0 -data PreparedSink input m output = - SinkNoData output - | SinkData - { sinkPush :: input -> ResourceT m (SinkResult input output) - , sinkClose :: ResourceT m output - } - -instance Monad m => Functor (PreparedSink input m) where - fmap f (SinkNoData x) = SinkNoData (f x) - fmap f (SinkData p c) = SinkData - { sinkPush = liftM (fmap f) . p - , sinkClose = liftM f c - } - --- | Most 'PreparedSink's require some type of state, similar to --- 'PreparedSource's. Like a @Source@ for a @PreparedSource@, a @Sink@ is a --- simple monadic wrapper around a @PreparedSink@ which allows initialization --- of such state. See @Source@ for further caveats. --- --- Note that this type provides a 'Monad' instance, allowing you to easily --- compose @Sink@s together. --- --- Since 0.0.0 -newtype Sink input m output = Sink { prepareSink :: ResourceT m (PreparedSink input m output) } - -instance Monad m => Functor (Sink input m) where - fmap f (Sink msink) = Sink (liftM (fmap f) msink) - -instance Resource m => Applicative (Sink input m) where - pure x = Sink (return (SinkNoData x)) - Sink mf <*> Sink ma = Sink $ do - f <- mf - a <- ma - case (f, a) of - (SinkNoData f', SinkNoData a') -> return (SinkNoData (f' a')) - _ -> do - istate <- newRef (toEither f, toEither a) - return $ appHelper istate - -toEither :: PreparedSink input m output -> SinkEither input m output -toEither (SinkData x y) = SinkPair x y -toEither (SinkNoData x) = SinkOutput x - -type SinkPush input m output = input -> ResourceT m (SinkResult input output) -type SinkClose input m output = ResourceT m output -data SinkEither input m output - = SinkPair (SinkPush input m output) (SinkClose input m output) - | SinkOutput output -type SinkState input m a b = Ref (Base m) (SinkEither input m (a -> b), SinkEither input m a) - -appHelper :: Resource m => SinkState input m a b -> PreparedSink input m b -appHelper istate = SinkData (pushHelper istate) (closeHelper istate) - -pushHelper :: Resource m - => SinkState input m a b - -> input - -> ResourceT m (SinkResult input b) -pushHelper istate stream0 = do - state <- readRef istate - go state stream0 - where - go (SinkPair f _, eb) stream = do - mres <- f stream - case mres of - Processing -> return Processing - Done leftover res -> do - let state' = (SinkOutput res, eb) - writeRef istate state' - maybe (return Processing) (go state') leftover - go (f@SinkOutput{}, SinkPair b _) stream = do - mres <- b stream - case mres of - Processing -> return Processing - Done leftover res -> do - let state' = (f, SinkOutput res) - writeRef istate state' - maybe (return Processing) (go state') leftover - go (SinkOutput f, SinkOutput b) leftover = return $ Done (Just leftover) $ f b - -closeHelper :: Resource m - => SinkState input m a b - -> ResourceT m b -closeHelper istate = do - (sf, sa) <- readRef istate - case sf of - SinkOutput f -> go' f sa - SinkPair _ close -> do - f <- close - go' f sa - where - go' f (SinkPair _ close) = do - a <- close - return (f a) - go' f (SinkOutput a) = return (f a) - -instance Resource m => Monad (Sink input m) where - return = pure - mx >>= f = Sink $ do - x <- prepareSink mx - case x of - SinkNoData x' -> prepareSink $ f x' - SinkData push' close' -> do - istate <- newRef $ Left (push', close') - return $ SinkData (push istate) (close istate) - where - push istate input = do - state <- readRef istate - case state of - Left (push', _) -> do - res <- push' input - case res of - Done leftover output -> do - f' <- prepareSink $ f output - case f' of - SinkNoData y -> - return $ Done leftover y - SinkData pushF closeF -> do - writeRef istate $ Right (pushF, closeF) - maybe (return Processing) (push istate) leftover - Processing -> return Processing - Right (push', _) -> push' input - close istate = do - state <- readRef istate - case state of - Left (_, close') -> do - output <- close' - f' <- prepareSink $ f output - case f' of - SinkNoData y -> return y - SinkData _ closeF -> closeF - Right (_, close') -> close' - -instance (Resource m, Base m ~ base, Applicative base) => MonadBase base (Sink input m) where - liftBase = lift . resourceLiftBase - -instance MonadTrans (Sink input) where - lift f = Sink (lift (liftM SinkNoData f)) - -instance (Resource m, MonadIO m) => MonadIO (Sink input m) where - liftIO = lift . liftIO +{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}+-- | Defines the types for a sink, which is a consumer of data.+module Data.Conduit.Types.Sink+ ( SinkResult (..)+ , PreparedSink (..)+ , Sink (..)+ ) where++import Control.Monad.Trans.Resource+import Control.Monad.Trans.Class (MonadTrans (lift))+import Control.Monad.IO.Class (MonadIO (liftIO))+import Control.Monad (liftM)+import Control.Applicative (Applicative (..))+import Control.Monad.Base (MonadBase (liftBase))++-- | A @Sink@ ultimately returns a single output value. Each time data is+-- pushed to it, a @Sink@ may indicate that it is still processing data, or+-- that it is done, in which case it returns some optional leftover input and+-- an output value.+--+-- Since 0.0.0+data SinkResult input output = Processing | Done (Maybe input) output+instance Functor (SinkResult input) where+ fmap _ Processing = Processing+ fmap f (Done input output) = Done input (f 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:+--+-- * Some sinks do not actually require any data to produce an output. This is+-- included with a sink in order to allow for a 'Monad' instance.+--+-- * 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'.+--+-- Invariants:+--+-- * After a 'PreparedSink' produces a result (either via 'sinkPush' or+-- 'sinkClose'), neither of those two functions may be called on the @Sink@+-- again.+--+-- * If a @Sink@ needs to clean up any resources (e.g., close a file handle),+-- it must do so whenever it returns a result, either via @sinkPush@ or+-- @sinkClose@. Note that, due to usage of @ResourceT@, this is merely an+-- optimization.+--+-- Since 0.0.0+data PreparedSink input m output =+ SinkNoData output+ | SinkData+ { sinkPush :: input -> ResourceT m (SinkResult input output)+ , sinkClose :: ResourceT m output+ }++instance Monad m => Functor (PreparedSink input m) where+ fmap f (SinkNoData x) = SinkNoData (f x)+ fmap f (SinkData p c) = SinkData+ { sinkPush = liftM (fmap f) . p+ , sinkClose = liftM f c+ }++-- | Most 'PreparedSink's require some type of state, similar to+-- 'PreparedSource's. Like a @Source@ for a @PreparedSource@, a @Sink@ is a+-- simple monadic wrapper around a @PreparedSink@ which allows initialization+-- of such state. See @Source@ for further caveats.+--+-- Note that this type provides a 'Monad' instance, allowing you to easily+-- compose @Sink@s together.+--+-- Since 0.0.0+newtype Sink input m output = Sink { prepareSink :: ResourceT m (PreparedSink input m output) }++instance Monad m => Functor (Sink input m) where+ fmap f (Sink msink) = Sink (liftM (fmap f) msink)++instance Resource m => Applicative (Sink input m) where+ pure x = Sink (return (SinkNoData x))+ Sink mf <*> Sink ma = Sink $ do+ f <- mf+ a <- ma+ case (f, a) of+ (SinkNoData f', SinkNoData a') -> return (SinkNoData (f' a'))+ _ -> do+ istate <- newRef (toEither f, toEither a)+ return $ appHelper istate++toEither :: PreparedSink input m output -> SinkEither input m output+toEither (SinkData x y) = SinkPair x y+toEither (SinkNoData x) = SinkOutput x++type SinkPush input m output = input -> ResourceT m (SinkResult input output)+type SinkClose input m output = ResourceT m output+data SinkEither input m output+ = SinkPair (SinkPush input m output) (SinkClose input m output)+ | SinkOutput output+type SinkState input m a b = Ref (Base m) (SinkEither input m (a -> b), SinkEither input m a)++appHelper :: Resource m => SinkState input m a b -> PreparedSink input m b+appHelper istate = SinkData (pushHelper istate) (closeHelper istate)++pushHelper :: Resource m+ => SinkState input m a b+ -> input+ -> ResourceT m (SinkResult input b)+pushHelper istate stream0 = do+ state <- readRef istate+ go state stream0+ where+ go (SinkPair f _, eb) stream = do+ mres <- f stream+ case mres of+ Processing -> return Processing+ Done leftover res -> do+ let state' = (SinkOutput res, eb)+ writeRef istate state'+ maybe (return Processing) (go state') leftover+ go (f@SinkOutput{}, SinkPair b _) stream = do+ mres <- b stream+ case mres of+ Processing -> return Processing+ Done leftover res -> do+ let state' = (f, SinkOutput res)+ writeRef istate state'+ maybe (return Processing) (go state') leftover+ go (SinkOutput f, SinkOutput b) leftover = return $ Done (Just leftover) $ f b++closeHelper :: Resource m+ => SinkState input m a b+ -> ResourceT m b+closeHelper istate = do+ (sf, sa) <- readRef istate+ case sf of+ SinkOutput f -> go' f sa+ SinkPair _ close -> do+ f <- close+ go' f sa+ where+ go' f (SinkPair _ close) = do+ a <- close+ return (f a)+ go' f (SinkOutput a) = return (f a)++instance Resource m => Monad (Sink input m) where+ return = pure+ mx >>= f = Sink $ do+ x <- prepareSink mx+ case x of+ SinkNoData x' -> prepareSink $ f x'+ SinkData push' close' -> do+ istate <- newRef $ Left (push', close')+ return $ SinkData (push istate) (close istate)+ where+ push istate input = do+ state <- readRef istate+ case state of+ Left (push', _) -> do+ res <- push' input+ case res of+ Done leftover output -> do+ f' <- prepareSink $ f output+ case f' of+ SinkNoData y ->+ return $ Done leftover y+ SinkData pushF closeF -> do+ writeRef istate $ Right (pushF, closeF)+ maybe (return Processing) (push istate) leftover+ Processing -> return Processing+ Right (push', _) -> push' input+ close istate = do+ state <- readRef istate+ case state of+ Left (_, close') -> do+ output <- close'+ f' <- prepareSink $ f output+ case f' of+ SinkNoData y -> return y+ SinkData _ closeF -> closeF+ Right (_, close') -> close'++instance (Resource m, Base m ~ base, Applicative base) => MonadBase base (Sink input m) where+ liftBase = lift . resourceLiftBase++instance MonadTrans (Sink input) where+ lift f = Sink (lift (liftM SinkNoData f))++instance (Resource m, MonadIO m) => MonadIO (Sink input m) where+ liftIO = lift . liftIO
Data/Conduit/Types/Source.hs view
@@ -1,247 +1,128 @@-{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE DeriveDataTypeable #-} --- | Defines the types for a source, which is a producer of data. -module Data.Conduit.Types.Source - ( SourceResult (..) - , PreparedSource (..) - , Source (..) - , BufferedSource (..) - , SourceInvariantException (..) - , BufferSource (..) - , unbufferSource - ) where - -import Control.Monad.Trans.Resource -import Data.Monoid (Monoid (..)) -import Control.Monad (liftM) -import Data.Typeable (Typeable) -import Control.Exception (Exception, throw) - --- | Result of pulling from a source. Either a new piece of data (@Open@), or --- indicates that the source is now @Closed@. --- --- Since 0.0.0 -data SourceResult a = Open a | Closed - deriving (Show, Eq, Ord) - -instance Functor SourceResult where - fmap f (Open a) = Open (f a) - fmap _ Closed = Closed - --- | A 'PreparedSource' has two operations on it: pull some data, and close the --- 'PreparedSource'. Since 'PreparedSource' is built on top of 'ResourceT', all --- acquired resources should be automatically released anyway. Closing a --- 'PreparedSource' early --- is merely an optimization to free scarce resources as soon as possible. --- --- A 'PreparedSource' has three invariants: --- --- * It is illegal to call 'sourcePull' after a previous call returns 'Closed', or after a call to 'sourceClose'. --- --- * It is illegal to call 'sourceClose' multiple times, or after a previous --- 'sourcePull' returns a 'Closed'. --- --- * A 'PreparedSource' is responsible to free any resources when either 'sourceClose' --- is called or a 'Closed' is returned. However, based on the usage of --- 'ResourceT', this is simply an optimization. --- --- Since 0.0.0 -data PreparedSource m a = PreparedSource - { sourcePull :: ResourceT m (SourceResult a) - , sourceClose :: ResourceT m () - } - -instance Monad m => Functor (PreparedSource m) where - fmap f src = src - { sourcePull = liftM (fmap f) (sourcePull src) - } - --- | All but the simplest of 'PreparedSource's (e.g., @repeat@) require some --- type of state to track their current status. This may be in the form of a --- mutable variable (e.g., @IORef@), or via opening a resource like a @Handle@. --- While a 'PreparedSource' is given no opportunity to acquire such resources, --- this type is. --- --- A 'Source' is simply a monadic action that returns a 'PreparedSource'. One --- nice consequence of this is the possibility of creating an efficient --- 'Monoid' instance, which will only acquire one resource at a time, instead --- of bulk acquiring all resources at the beginning of running the 'Source'. --- --- Note that each time you \"call\" a @Source@, it is started from scratch. If --- you want a resumable source (e.g., one which can be passed to multiple --- @Sink@s), you likely want to use a 'BufferedSource'. --- --- Since 0.0.0 -newtype Source m a = Source { prepareSource :: ResourceT m (PreparedSource m a) } - -instance Monad m => Functor (Source m) where - fmap f (Source msrc) = Source (liftM (fmap f) msrc) - -instance Resource m => Monoid (Source m a) where - mempty = Source (return PreparedSource - { sourcePull = return Closed - , sourceClose = return () - }) - mappend a b = mconcat [a, b] - mconcat [] = mempty - mconcat (Source mnext:rest0) = Source $ do - -- open up the first Source... - next0 <- mnext - -- and place it in a mutable reference along with all of the upcoming - -- Sources - istate <- newRef (next0, rest0) - return PreparedSource - { sourcePull = pull istate - , sourceClose = close istate - } - where - pull istate = - readRef istate >>= pull' - where - pull' (current, rest) = do - res <- sourcePull current - case res of - -- end of the current Source - Closed -> do - case rest of - -- ... and open the next one - Source ma:as -> do - a <- ma - writeRef istate (a, as) - -- continue pulling base on this new state - pull istate - -- no more source, return an EOF - [] -> do - -- give an error message if the first Source - -- invariant is violated (read data after EOF) - writeRef istate $ - throw $ PullAfterEOF "Source:mconcat" - return Closed - Open _ -> return res - close istate = do - -- we only need to close the current Source, since they are opened - -- one at a time - (current, _) <- readRef istate - sourceClose current - --- | When actually interacting with 'Source's, we usually want to be able to --- buffer the output, in case any intermediate steps return leftover data. A --- 'BufferedSource' allows for such buffering, via the 'bsourceUnpull' function. --- --- A 'BufferedSource', unlike a 'Source', is resumable, meaning it can be passed to --- multiple 'Sink's without restarting. --- --- Finally, a 'BufferedSource' relaxes one of the invariants of a 'Source': calling --- 'bsourcePull' after an 'EOF' will simply return another 'EOF'. --- --- A @BufferedSource@ is also known as a /resumable source/, in that it can be --- called multiple times, and each time will provide new data. One caveat: --- while the types will allow you to use the buffered source in multiple --- threads, there is no guarantee that all @BufferedSource@s will handle this --- correctly. --- --- Since 0.0.0 -data BufferedSource m a = BufferedSource - { bsourcePull :: ResourceT m (SourceResult a) - , bsourceUnpull :: a -> ResourceT m () - , bsourceClose :: ResourceT m () - } - --- | --- Since 0.0.0 -data SourceInvariantException = PullAfterEOF String - deriving (Show, Typeable) -instance Exception SourceInvariantException - --- | This typeclass allows us to unify operators on 'Source' and 'BufferedSource'. --- --- Since 0.0.0 -class BufferSource s where - bufferSource :: Resource m => s m a -> ResourceT m (BufferedSource m a) - - -- | Same as 'bufferSource', but an implementation is guaranteed that the - -- resulting 'BufferedSource' will be used only once. As such, an - -- implementation may implement fake buffering, such as coding - -- 'bsourceUnpull' as a no-op. - unsafeBufferSource :: Resource m => s m a -> ResourceT m (BufferedSource m a) - unsafeBufferSource = bufferSource - --- | Note that this instance hides the 'bsourceClose' record, so that a --- @BufferedSource@ remains resumable. The correct way to handle closing of a --- resumable source would be to call @bsourceClose@ on the originally --- @BufferedSource@, e.g.: --- --- > bsrc <- bufferSource $ sourceFile "myfile.txt" --- > bsrc $$ drop 5 --- > rest <- bsrc $$ consume --- > bsourceClose bsrc --- --- Note that the call to the @$$@ operator allocates a /new/ 'BufferedSource' --- internally, so that when @$$@ calls @bsourceClose@ the first time, it does --- not close the actual file, thereby allowing us to pass the same @bsrc@ to --- the @consume@ function. Afterwards, we should call @bsourceClose@ manually --- (though @runResourceT@ will handle it for us eventually). -instance BufferSource BufferedSource where - bufferSource bsrc = return bsrc - { bsourceClose = return () - } - --- | State of a 'BufferedSource' -data BState a = BOpen [a] - | BClosed [a] - deriving Show - -instance BufferSource PreparedSource where - bufferSource src = do - istate <- newRef $ BOpen [] - return BufferedSource - { bsourcePull = do - mresult <- modifyRef istate $ \state -> - case state of - BOpen [] -> (state, Nothing) - BClosed [] -> (state, Just Closed) - BOpen (x:xs) -> (BOpen xs, Just $ Open x) - BClosed (x:xs) -> (BClosed xs, Just $ Open x) - case mresult of - Nothing -> do - result <- sourcePull src - case result of - Closed -> writeRef istate $ BClosed [] - Open _ -> return () - return result - Just result -> return result - , bsourceUnpull = \x -> - modifyRef istate $ \state -> - case state of - BOpen buffer -> (BOpen (x : buffer), ()) - BClosed buffer -> (BClosed (x : buffer), ()) - , bsourceClose = do - action <- modifyRef istate $ \state -> - case state of - BOpen x -> (BClosed x, sourceClose src) - BClosed _ -> (state, return ()) - action - } - unsafeBufferSource src = return BufferedSource - { bsourcePull = sourcePull src - , bsourceClose = sourceClose src - , bsourceUnpull = const $ return () - } - -instance BufferSource Source where - bufferSource (Source msrc) = msrc >>= bufferSource - unsafeBufferSource (Source msrc) = msrc >>= unsafeBufferSource - --- | Turn a 'BufferedSource' into a 'Source'. Note that in general this will --- mean your original 'BufferedSource' will be closed. Additionally, all --- leftover data from usage of the returned @Source@ will be discarded. In --- other words: this is a no-going-back move. --- --- Note: @bufferSource@ . @unbufferSource@ is /not/ the identity function. --- --- Since 0.0.1 -unbufferSource :: Monad m - => BufferedSource m a - -> Source m a -unbufferSource (BufferedSource pull _unpull close) = - Source $ return $ PreparedSource pull close +{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE DeriveDataTypeable #-}+-- | Defines the types for a source, which is a producer of data.+module Data.Conduit.Types.Source+ ( SourceResult (..)+ , PreparedSource (..)+ , Source (..)+ , SourceInvariantException (..)+ ) where++import Control.Monad.Trans.Resource+import Data.Monoid (Monoid (..))+import Control.Monad (liftM)+import Data.Typeable (Typeable)+import Control.Exception (Exception, throw)++-- | Result of pulling from a source. Either a new piece of data (@Open@), or+-- indicates that the source is now @Closed@.+--+-- Since 0.0.0+data SourceResult a = Open a | Closed+ deriving (Show, Eq, Ord)++instance Functor SourceResult where+ fmap f (Open a) = Open (f a)+ fmap _ Closed = Closed++-- | A 'PreparedSource' has two operations on it: pull some data, and close the+-- 'PreparedSource'. Since 'PreparedSource' is built on top of 'ResourceT', all+-- acquired resources should be automatically released anyway. Closing a+-- 'PreparedSource' early+-- is merely an optimization to free scarce resources as soon as possible.+--+-- A 'PreparedSource' has three invariants:+--+-- * It is illegal to call 'sourcePull' after a previous call returns 'Closed', or after a call to 'sourceClose'.+--+-- * It is illegal to call 'sourceClose' multiple times, or after a previous+-- 'sourcePull' returns a 'Closed'.+--+-- * A 'PreparedSource' is responsible to free any resources when either 'sourceClose'+-- is called or a 'Closed' is returned. However, based on the usage of+-- 'ResourceT', this is simply an optimization.+--+-- Since 0.0.0+data PreparedSource m a = PreparedSource+ { sourcePull :: ResourceT m (SourceResult a)+ , sourceClose :: ResourceT m ()+ }++instance Monad m => Functor (PreparedSource m) where+ fmap f src = src+ { sourcePull = liftM (fmap f) (sourcePull src)+ }++-- | All but the simplest of 'PreparedSource's (e.g., @repeat@) require some+-- type of state to track their current status. This may be in the form of a+-- mutable variable (e.g., @IORef@), or via opening a resource like a @Handle@.+-- While a 'PreparedSource' is given no opportunity to acquire such resources,+-- this type is.+--+-- A 'Source' is simply a monadic action that returns a 'PreparedSource'. One+-- nice consequence of this is the possibility of creating an efficient+-- 'Monoid' instance, which will only acquire one resource at a time, instead+-- of bulk acquiring all resources at the beginning of running the 'Source'.+--+-- Note that each time you \"call\" a @Source@, it is started from scratch. If+-- you want a resumable source (e.g., one which can be passed to multiple+-- @Sink@s), you likely want to use a 'BufferedSource'.+--+-- Since 0.0.0+newtype Source m a = Source { prepareSource :: ResourceT m (PreparedSource m a) }++instance Monad m => Functor (Source m) where+ fmap f (Source msrc) = Source (liftM (fmap f) msrc)++instance Resource m => Monoid (Source m a) where+ mempty = Source (return PreparedSource+ { sourcePull = return Closed+ , sourceClose = return ()+ })+ mappend a b = mconcat [a, b]+ mconcat [] = mempty+ mconcat (Source mnext:rest0) = Source $ do+ -- open up the first Source...+ next0 <- mnext+ -- and place it in a mutable reference along with all of the upcoming+ -- Sources+ istate <- newRef (next0, rest0)+ return PreparedSource+ { sourcePull = pull istate+ , sourceClose = close istate+ }+ where+ pull istate =+ readRef istate >>= pull'+ where+ pull' (current, rest) = do+ res <- sourcePull current+ case res of+ -- end of the current Source+ Closed -> do+ case rest of+ -- ... and open the next one+ Source ma:as -> do+ a <- ma+ writeRef istate (a, as)+ -- continue pulling base on this new state+ pull istate+ -- no more source, return an EOF+ [] -> do+ -- give an error message if the first Source+ -- invariant is violated (read data after EOF)+ writeRef istate $+ throw $ PullAfterEOF "Source:mconcat"+ return Closed+ Open _ -> return res+ close istate = do+ -- we only need to close the current Source, since they are opened+ -- one at a time+ (current, _) <- readRef istate+ sourceClose current++-- |+-- Since 0.0.0+data SourceInvariantException = PullAfterEOF String+ deriving (Show, Typeable)+instance Exception SourceInvariantException
Data/Conduit/Util/Conduit.hs view
@@ -1,201 +1,201 @@-{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE TypeFamilies #-} -{-# LANGUAGE CPP #-} --- | 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 - , conduitIO - , transConduit - -- *** Sequencing - , SequencedSink - , sequenceSink - , SequencedSinkResponse (..) - ) where - -import Control.Monad.Trans.Resource -import Control.Monad.Trans.Class -import Data.Conduit.Types.Conduit -import Data.Conduit.Types.Sink -import Control.Monad (liftM) - --- | Construct a 'Conduit' with some stateful functions. This function address --- all mutable state for you. --- --- Since 0.0.0 -conduitState - :: Resource m - => state -- ^ initial state - -> (state -> input -> ResourceT m (state, ConduitResult input output)) -- ^ Push function. - -> (state -> ResourceT m [output]) -- ^ Close function. The state need not be returned, since it will not be used again. - -> Conduit input m output -conduitState state0 push close = Conduit $ do -#if DEBUG - iclosed <- newRef False -#endif - istate <- newRef state0 - return PreparedConduit - { conduitPush = \input -> do -#if DEBUG - False <- readRef iclosed -#endif - state <- readRef istate - (state', res) <- state `seq` push state input - writeRef istate state' -#if DEBUG - case res of - Finished _ _ -> writeRef iclosed True - Producing _ -> return () -#endif - return res - , conduitClose = do -#if DEBUG - False <- readRef iclosed - writeRef iclosed True -#endif - readRef istate >>= close - } - --- | Construct a 'Conduit'. --- --- Since 0.0.0 -conduitIO :: ResourceIO m - => IO state -- ^ resource and/or state allocation - -> (state -> IO ()) -- ^ resource and/or state cleanup - -> (state -> input -> m (ConduitResult input output)) -- ^ Push function. Note that this need not explicitly perform any cleanup. - -> (state -> m [output]) -- ^ Close function. Note that this need not explicitly perform any cleanup. - -> Conduit input m output -conduitIO alloc cleanup push close = Conduit $ do -#if DEBUG - iclosed <- newRef False -#endif - (key, state) <- withIO alloc cleanup - return PreparedConduit - { conduitPush = \input -> do -#if DEBUG - False <- readRef iclosed -#endif - res <- lift $ push state input - case res of - Producing{} -> return () - Finished{} -> do -#if DEBUG - writeRef iclosed True -#endif - release key - return res - , conduitClose = do -#if DEBUG - False <- readRef iclosed - writeRef iclosed True -#endif - output <- lift $ close state - release key - return output - } - --- | Transform the monad a 'Conduit' lives in. --- --- Since 0.0.0 -transConduit :: (Monad m, Base m ~ Base n) - => (forall a. m a -> n a) - -> Conduit input m output - -> Conduit input n output -transConduit f (Conduit mc) = - Conduit (transResourceT f (liftM go mc)) - where - go c = c - { conduitPush = transResourceT f . conduitPush c - , conduitClose = transResourceT f (conduitClose c) - } - --- | Return value from a 'SequencedSink'. --- --- Since 0.0.0 -data SequencedSinkResponse state input m output = - Emit state [output] -- ^ Set a new state, and emit some new output. - | Stop -- ^ End the conduit. - | StartConduit (Conduit input m output) -- ^ Pass control to a new conduit. - --- | Helper type for constructing a @Conduit@ based on @Sink@s. This allows you --- to write higher-level code that takes advantage of existing conduits and --- sinks, and leverages a sink's monadic interface. --- --- Since 0.0.0 -type SequencedSink state input m output = - state -> Sink input m (SequencedSinkResponse state input m output) - -data SCState state input m output = - SCNewState state - | SCConduit (PreparedConduit input m output) - | SCSink (input -> ResourceT m (SinkResult input (SequencedSinkResponse state input m output))) - (ResourceT m (SequencedSinkResponse state input m output)) - --- | Convert a 'SequencedSink' into a 'Conduit'. --- --- Since 0.0.0 -sequenceSink - :: Resource m - => state -- ^ initial state - -> SequencedSink state input m output - -> Conduit input m output -sequenceSink state0 fsink = conduitState - (SCNewState state0) - (scPush id fsink) - scClose - -goRes :: Resource m - => SequencedSinkResponse state input m output - -> Maybe input - -> ([output] -> [output]) - -> SequencedSink state input m output - -> ResourceT m (SCState state input m output, ConduitResult input output) -goRes (Emit state output) (Just input) front fsink = - scPush (front . (output++)) fsink (SCNewState state) input -goRes (Emit state output) Nothing front _ = - return (SCNewState state, Producing $ front output) -goRes Stop minput front _ = - return (error "sequenceSink", Finished minput $ front []) -goRes (StartConduit c) Nothing front _ = do - pc <- prepareConduit c - return (SCConduit pc, Producing $ front []) -goRes (StartConduit c) (Just input) front fsink = do - pc <- prepareConduit c - scPush front fsink (SCConduit pc) input - -scPush :: Resource m - => ([output] -> [output]) - -> SequencedSink state input m output - -> SCState state input m output - -> input - -> ResourceT m (SCState state input m output, ConduitResult input output) -scPush front fsink (SCNewState state) input = do - sink <- prepareSink $ fsink state - case sink of - SinkData push' close' -> scPush front fsink (SCSink push' close') input - SinkNoData res -> goRes res (Just input) front fsink -scPush front _ (SCConduit conduit) input = do - res <- conduitPush conduit input - let res' = - case res of - Producing x -> Producing $ front x - Finished x y -> Finished x $ front y - return (SCConduit conduit, res') -scPush front fsink (SCSink push close) input = do - mres <- push input - case mres of - Done minput res -> goRes res minput front fsink - Processing -> return (SCSink push close, Producing $ 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 -> do - pc <- prepareConduit c - conduitClose pc +{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE CPP #-}+-- | 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+ , conduitIO+ , transConduit+ -- *** Sequencing+ , SequencedSink+ , sequenceSink+ , SequencedSinkResponse (..)+ ) where++import Control.Monad.Trans.Resource+import Control.Monad.Trans.Class+import Data.Conduit.Types.Conduit+import Data.Conduit.Types.Sink+import Control.Monad (liftM)++-- | Construct a 'Conduit' with some stateful functions. This function address+-- all mutable state for you.+--+-- Since 0.0.0+conduitState+ :: Resource m+ => state -- ^ initial state+ -> (state -> input -> ResourceT m (state, ConduitResult input output)) -- ^ Push function.+ -> (state -> ResourceT m [output]) -- ^ Close function. The state need not be returned, since it will not be used again.+ -> Conduit input m output+conduitState state0 push close = Conduit $ do+#if DEBUG+ iclosed <- newRef False+#endif+ istate <- newRef state0+ return PreparedConduit+ { conduitPush = \input -> do+#if DEBUG+ False <- readRef iclosed+#endif+ state <- readRef istate+ (state', res) <- state `seq` push state input+ writeRef istate state'+#if DEBUG+ case res of+ Finished _ _ -> writeRef iclosed True+ Producing _ -> return ()+#endif+ return res+ , conduitClose = do+#if DEBUG+ False <- readRef iclosed+ writeRef iclosed True+#endif+ readRef istate >>= close+ }++-- | Construct a 'Conduit'.+--+-- Since 0.0.0+conduitIO :: ResourceIO m+ => IO state -- ^ resource and/or state allocation+ -> (state -> IO ()) -- ^ resource and/or state cleanup+ -> (state -> input -> m (ConduitResult input output)) -- ^ Push function. Note that this need not explicitly perform any cleanup.+ -> (state -> m [output]) -- ^ Close function. Note that this need not explicitly perform any cleanup.+ -> Conduit input m output+conduitIO alloc cleanup push close = Conduit $ do+#if DEBUG+ iclosed <- newRef False+#endif+ (key, state) <- withIO alloc cleanup+ return PreparedConduit+ { conduitPush = \input -> do+#if DEBUG+ False <- readRef iclosed+#endif+ res <- lift $ push state input+ case res of+ Producing{} -> return ()+ Finished{} -> do+#if DEBUG+ writeRef iclosed True+#endif+ release key+ return res+ , conduitClose = do+#if DEBUG+ False <- readRef iclosed+ writeRef iclosed True+#endif+ output <- lift $ close state+ release key+ return output+ }++-- | Transform the monad a 'Conduit' lives in.+--+-- Since 0.0.0+transConduit :: (Monad m, Base m ~ Base n)+ => (forall a. m a -> n a)+ -> Conduit input m output+ -> Conduit input n output+transConduit f (Conduit mc) =+ Conduit (transResourceT f (liftM go mc))+ where+ go c = c+ { conduitPush = transResourceT f . conduitPush c+ , conduitClose = transResourceT f (conduitClose c)+ }++-- | Return value from a 'SequencedSink'.+--+-- Since 0.0.0+data SequencedSinkResponse state input m output =+ Emit state [output] -- ^ Set a new state, and emit some new output.+ | Stop -- ^ End the conduit.+ | StartConduit (Conduit input m output) -- ^ Pass control to a new conduit.++-- | Helper type for constructing a @Conduit@ based on @Sink@s. This allows you+-- to write higher-level code that takes advantage of existing conduits and+-- sinks, and leverages a sink's monadic interface.+--+-- Since 0.0.0+type SequencedSink state input m output =+ state -> Sink input m (SequencedSinkResponse state input m output)++data SCState state input m output =+ SCNewState state+ | SCConduit (PreparedConduit input m output)+ | SCSink (input -> ResourceT m (SinkResult input (SequencedSinkResponse state input m output)))+ (ResourceT m (SequencedSinkResponse state input m output))++-- | Convert a 'SequencedSink' into a 'Conduit'.+--+-- Since 0.0.0+sequenceSink+ :: Resource m+ => state -- ^ initial state+ -> SequencedSink state input m output+ -> Conduit input m output+sequenceSink state0 fsink = conduitState+ (SCNewState state0)+ (scPush id fsink)+ scClose++goRes :: Resource m+ => SequencedSinkResponse state input m output+ -> Maybe input+ -> ([output] -> [output])+ -> SequencedSink state input m output+ -> ResourceT m (SCState state input m output, ConduitResult input output)+goRes (Emit state output) (Just input) front fsink =+ scPush (front . (output++)) fsink (SCNewState state) input+goRes (Emit state output) Nothing front _ =+ return (SCNewState state, Producing $ front output)+goRes Stop minput front _ =+ return (error "sequenceSink", Finished minput $ front [])+goRes (StartConduit c) Nothing front _ = do+ pc <- prepareConduit c+ return (SCConduit pc, Producing $ front [])+goRes (StartConduit c) (Just input) front fsink = do+ pc <- prepareConduit c+ scPush front fsink (SCConduit pc) input++scPush :: Resource m+ => ([output] -> [output])+ -> SequencedSink state input m output+ -> SCState state input m output+ -> input+ -> ResourceT m (SCState state input m output, ConduitResult input output)+scPush front fsink (SCNewState state) input = do+ sink <- prepareSink $ fsink state+ case sink of+ SinkData push' close' -> scPush front fsink (SCSink push' close') input+ SinkNoData res -> goRes res (Just input) front fsink+scPush front _ (SCConduit conduit) input = do+ res <- conduitPush conduit input+ let res' =+ case res of+ Producing x -> Producing $ front x+ Finished x y -> Finished x $ front y+ return (SCConduit conduit, res')+scPush front fsink (SCSink push close) input = do+ mres <- push input+ case mres of+ Done minput res -> goRes res minput front fsink+ Processing -> return (SCSink push close, Producing $ 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 -> do+ pc <- prepareConduit c+ conduitClose pc
Data/Conduit/Util/Sink.hs view
@@ -1,107 +1,107 @@-{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE TypeFamilies #-} -{-# LANGUAGE CPP #-} --- | Utilities for constructing 'Sink's. Please see "Data.Conduit.Types.Sink" --- for more information on the base types. -module Data.Conduit.Util.Sink - ( sinkState - , sinkIO - , transSink - ) where - -import Control.Monad.Trans.Resource -import Control.Monad.Trans.Class (lift) -import Data.Conduit.Types.Sink -import Control.Monad (liftM) - --- | Construct a 'Sink' with some stateful functions. This function address --- all mutable state for you. --- --- Since 0.0.0 -sinkState - :: Resource m - => state -- ^ initial state - -> (state -> input -> ResourceT m (state, SinkResult input output)) -- ^ push - -> (state -> ResourceT m output) -- ^ Close. Note that the state is not returned, as it is not needed. - -> Sink input m output -sinkState state0 push close = Sink $ do - istate <- newRef state0 -#if DEBUG - iclosed <- newRef False -#endif - return SinkData - { sinkPush = \input -> do -#if DEBUG - False <- readRef iclosed -#endif - state <- readRef istate - (state', res) <- state `seq` push state input - writeRef istate state' -#if DEBUG - case res of - Done{} -> writeRef iclosed True - Processing -> return () -#endif - return res - , sinkClose = do -#if DEBUG - False <- readRef iclosed - writeRef iclosed True -#endif - readRef istate >>= close - } - --- | Construct a 'Sink'. Note that your push and close functions need not --- explicitly perform any cleanup. --- --- Since 0.0.0 -sinkIO :: ResourceIO m - => IO state -- ^ resource and/or state allocation - -> (state -> IO ()) -- ^ resource and/or state cleanup - -> (state -> input -> m (SinkResult input output)) -- ^ push - -> (state -> m output) -- ^ close - -> Sink input m output -sinkIO alloc cleanup push close = Sink $ do - (key, state) <- withIO alloc cleanup -#if DEBUG - iclosed <- newRef False -#endif - return SinkData - { sinkPush = \input -> do -#if DEBUG - False <- readRef iclosed -#endif - res <- lift $ push state input - case res of - Done{} -> do - release key -#if DEBUG - writeRef iclosed True -#endif - Processing -> return () - return res - , sinkClose = do -#if DEBUG - False <- readRef iclosed - writeRef iclosed True -#endif - res <- lift $ close state - release key - return res - } - --- | Transform the monad a 'Sink' lives in. --- --- Since 0.0.0 -transSink :: (Base m ~ Base n, Monad m) - => (forall a. m a -> n a) - -> Sink input m output - -> Sink input n output -transSink f (Sink mc) = - Sink (transResourceT f (liftM go mc)) - where - go c = c - { sinkPush = transResourceT f . sinkPush c - , sinkClose = transResourceT f (sinkClose c) - } +{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE CPP #-}+-- | Utilities for constructing 'Sink's. Please see "Data.Conduit.Types.Sink"+-- for more information on the base types.+module Data.Conduit.Util.Sink+ ( sinkState+ , sinkIO+ , transSink+ ) where++import Control.Monad.Trans.Resource+import Control.Monad.Trans.Class (lift)+import Data.Conduit.Types.Sink+import Control.Monad (liftM)++-- | Construct a 'Sink' with some stateful functions. This function address+-- all mutable state for you.+--+-- Since 0.0.0+sinkState+ :: Resource m+ => state -- ^ initial state+ -> (state -> input -> ResourceT m (state, SinkResult input output)) -- ^ push+ -> (state -> ResourceT m output) -- ^ Close. Note that the state is not returned, as it is not needed.+ -> Sink input m output+sinkState state0 push close = Sink $ do+ istate <- newRef state0+#if DEBUG+ iclosed <- newRef False+#endif+ return SinkData+ { sinkPush = \input -> do+#if DEBUG+ False <- readRef iclosed+#endif+ state <- readRef istate+ (state', res) <- state `seq` push state input+ writeRef istate state'+#if DEBUG+ case res of+ Done{} -> writeRef iclosed True+ Processing -> return ()+#endif+ return res+ , sinkClose = do+#if DEBUG+ False <- readRef iclosed+ writeRef iclosed True+#endif+ readRef istate >>= close+ }++-- | Construct a 'Sink'. Note that your push and close functions need not+-- explicitly perform any cleanup.+--+-- Since 0.0.0+sinkIO :: ResourceIO m+ => IO state -- ^ resource and/or state allocation+ -> (state -> IO ()) -- ^ resource and/or state cleanup+ -> (state -> input -> m (SinkResult input output)) -- ^ push+ -> (state -> m output) -- ^ close+ -> Sink input m output+sinkIO alloc cleanup push close = Sink $ do+ (key, state) <- withIO alloc cleanup+#if DEBUG+ iclosed <- newRef False+#endif+ return SinkData+ { sinkPush = \input -> do+#if DEBUG+ False <- readRef iclosed+#endif+ res <- lift $ push state input+ case res of+ Done{} -> do+ release key+#if DEBUG+ writeRef iclosed True+#endif+ Processing -> return ()+ return res+ , sinkClose = do+#if DEBUG+ False <- readRef iclosed+ writeRef iclosed True+#endif+ res <- lift $ close state+ release key+ return res+ }++-- | Transform the monad a 'Sink' lives in.+--+-- Since 0.0.0+transSink :: (Base m ~ Base n, Monad m)+ => (forall a. m a -> n a)+ -> Sink input m output+ -> Sink input n output+transSink f (Sink mc) =+ Sink (transResourceT f (liftM go mc))+ where+ go c = c+ { sinkPush = transResourceT f . sinkPush c+ , sinkClose = transResourceT f (sinkClose c)+ }
Data/Conduit/Util/Source.hs view
@@ -1,106 +1,106 @@-{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE TypeFamilies #-} -{-# LANGUAGE CPP #-} --- | Utilities for constructing and converting 'Source', 'Source' and --- 'BSource' types. Please see "Data.Conduit.Types.Source" for more information --- on the base types. -module Data.Conduit.Util.Source - ( sourceState - , sourceIO - , transSource - ) where - -import Control.Monad.Trans.Resource -import Control.Monad.Trans.Class (lift) -import Data.Conduit.Types.Source -import Control.Monad (liftM) - --- | Construct a 'Source' with some stateful functions. This function address --- all mutable state for you. --- --- Since 0.0.0 -sourceState - :: Resource m - => state -- ^ Initial state - -> (state -> ResourceT m (state, SourceResult output)) -- ^ Pull function - -> Source m output -sourceState state0 pull = Source $ do - istate <- newRef state0 -#if DEBUG - iclosed <- newRef False -#endif - return PreparedSource - { sourcePull = do -#if DEBUG - False <- readRef iclosed -#endif - state <- readRef istate - (state', res) <- pull state -#if DEBUG - let isClosed = - case res of - Closed -> True - Open _ -> False - writeRef iclosed isClosed -#endif - writeRef istate state' - return res - , sourceClose = do -#if DEBUG - False <- readRef iclosed - writeRef iclosed True -#else - return () -#endif - } - --- | Construct a 'Source' based on some IO actions for alloc/release. --- --- Since 0.0.0 -sourceIO :: ResourceIO m - => IO state -- ^ resource and/or state allocation - -> (state -> IO ()) -- ^ resource and/or state cleanup - -> (state -> m (SourceResult output)) -- ^ Pull function. Note that this need not explicitly perform any cleanup. - -> Source m output -sourceIO alloc cleanup pull = Source $ do - (key, state) <- withIO alloc cleanup -#if DEBUG - iclosed <- newRef False -#endif - return PreparedSource - { sourcePull = do -#if DEBUG - False <- readRef iclosed -#endif - res <- lift $ pull state - case res of - Closed -> do -#if DEBUG - writeRef iclosed True -#endif - release key - _ -> return () - return res - , sourceClose = do -#if DEBUG - False <- readRef iclosed - writeRef iclosed True -#endif - release key - } - --- | Transform the monad a 'Source' lives in. --- --- Since 0.0.0 -transSource :: (Base m ~ Base n, Monad m) - => (forall a. m a -> n a) - -> Source m output - -> Source n output -transSource f (Source mc) = - Source (transResourceT f (liftM go mc)) - where - go c = c - { sourcePull = transResourceT f (sourcePull c) - , sourceClose = transResourceT f (sourceClose c) - } +{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE CPP #-}+-- | Utilities for constructing and converting 'Source', 'Source' and+-- 'BSource' types. Please see "Data.Conduit.Types.Source" for more information+-- on the base types.+module Data.Conduit.Util.Source+ ( sourceState+ , sourceIO+ , transSource+ ) where++import Control.Monad.Trans.Resource+import Control.Monad.Trans.Class (lift)+import Data.Conduit.Types.Source+import Control.Monad (liftM)++-- | Construct a 'Source' with some stateful functions. This function address+-- all mutable state for you.+--+-- Since 0.0.0+sourceState+ :: Resource m+ => state -- ^ Initial state+ -> (state -> ResourceT m (state, SourceResult output)) -- ^ Pull function+ -> Source m output+sourceState state0 pull = Source $ do+ istate <- newRef state0+#if DEBUG+ iclosed <- newRef False+#endif+ return PreparedSource+ { sourcePull = do+#if DEBUG+ False <- readRef iclosed+#endif+ state <- readRef istate+ (state', res) <- pull state+#if DEBUG+ let isClosed =+ case res of+ Closed -> True+ Open _ -> False+ writeRef iclosed isClosed+#endif+ writeRef istate state'+ return res+ , sourceClose = do+#if DEBUG+ False <- readRef iclosed+ writeRef iclosed True+#else+ return ()+#endif+ }++-- | Construct a 'Source' based on some IO actions for alloc/release.+--+-- Since 0.0.0+sourceIO :: ResourceIO m+ => IO state -- ^ resource and/or state allocation+ -> (state -> IO ()) -- ^ resource and/or state cleanup+ -> (state -> m (SourceResult output)) -- ^ Pull function. Note that this need not explicitly perform any cleanup.+ -> Source m output+sourceIO alloc cleanup pull = Source $ do+ (key, state) <- withIO alloc cleanup+#if DEBUG+ iclosed <- newRef False+#endif+ return PreparedSource+ { sourcePull = do+#if DEBUG+ False <- readRef iclosed+#endif+ res <- lift $ pull state+ case res of+ Closed -> do+#if DEBUG+ writeRef iclosed True+#endif+ release key+ _ -> return ()+ return res+ , sourceClose = do+#if DEBUG+ False <- readRef iclosed+ writeRef iclosed True+#endif+ release key+ }++-- | Transform the monad a 'Source' lives in.+--+-- Since 0.0.0+transSource :: (Base m ~ Base n, Monad m)+ => (forall a. m a -> n a)+ -> Source m output+ -> Source n output+transSource f (Source mc) =+ Source (transResourceT f (liftM go mc))+ where+ go c = c+ { sourcePull = transResourceT f (sourcePull c)+ , sourceClose = transResourceT f (sourceClose c)+ }
LICENSE view
@@ -1,30 +1,30 @@-Copyright (c)2011, Michael Snoyman - -All rights reserved. - -Redistribution and use in source and binary forms, with or without -modification, are permitted provided that the following conditions are met: - - * Redistributions of source code must retain the above copyright - notice, this list of conditions and the following disclaimer. - - * Redistributions in binary form must reproduce the above - copyright notice, this list of conditions and the following - disclaimer in the documentation and/or other materials provided - with the distribution. - - * Neither the name of Michael Snoyman nor the names of other - contributors may be used to endorse or promote products derived - from this software without specific prior written permission. - -THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS -"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT -LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR -A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT -OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, -SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT -LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, -DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY -THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT -(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE -OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +Copyright (c)2011, Michael Snoyman++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Michael Snoyman nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
Setup.lhs view
@@ -1,7 +1,7 @@-#!/usr/bin/env runhaskell - -> module Main where -> import Distribution.Simple - -> main :: IO () -> main = defaultMain +#!/usr/bin/env runhaskell++> module Main where+> import Distribution.Simple++> main :: IO ()+> main = defaultMain
System/PosixFile.hsc view
@@ -1,57 +1,57 @@-{-# LANGUAGE ForeignFunctionInterface #-} -{-# LANGUAGE GeneralizedNewtypeDeriving #-} -module System.PosixFile - ( openRead - , read - , close - ) where - -import Foreign.C.String (CString, withCString) -import Foreign.Marshal.Alloc (mallocBytes, free) -import Foreign.C.Types (CInt) -import Foreign.C.Error (throwErrno) -import Foreign.Ptr (Ptr) -import Data.Bits (Bits) -import Data.Word (Word8) -import qualified Data.ByteString as S -import qualified Data.ByteString.Unsafe as BU -import Prelude hiding (read) -import Data.Conduit.Types.Source (SourceResult (..)) - -#include <fcntl.h> - -newtype Flag = Flag CInt - deriving (Num, Bits, Show, Eq) - -#{enum Flag, Flag - , oRdonly = O_RDONLY - } - -foreign import ccall "open" - c_open :: CString -> Flag -> IO CInt - -foreign import ccall "read" - c_read :: FD -> Ptr Word8 -> CInt -> IO CInt - -foreign import ccall "close" - close :: FD -> IO () - -newtype FD = FD CInt - -openRead :: FilePath -> IO FD -openRead fp = do - h <- withCString fp $ \str -> c_open str oRdonly - if h < 0 - then throwErrno $ "Could not open file: " ++ fp - else return $ FD h - -read :: FD -> IO (SourceResult S.ByteString) -read fd = do - cstr <- mallocBytes 4096 - len <- c_read fd cstr 4096 - if len == 0 - then free cstr >> return Closed - else fmap Open $ BU.unsafePackCStringFinalizer - cstr - (fromIntegral len) - (free cstr) +{-# LANGUAGE ForeignFunctionInterface #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+module System.PosixFile+ ( openRead+ , read+ , close+ ) where++import Foreign.C.String (CString, withCString)+import Foreign.Marshal.Alloc (mallocBytes, free)+import Foreign.C.Types (CInt)+import Foreign.C.Error (throwErrno)+import Foreign.Ptr (Ptr)+import Data.Bits (Bits)+import Data.Word (Word8)+import qualified Data.ByteString as S+import qualified Data.ByteString.Unsafe as BU+import Prelude hiding (read)+import Data.Conduit.Types.Source (SourceResult (..))++#include <fcntl.h>++newtype Flag = Flag CInt+ deriving (Num, Bits, Show, Eq)++#{enum Flag, Flag+ , oRdonly = O_RDONLY+ }++foreign import ccall "open"+ c_open :: CString -> Flag -> IO CInt++foreign import ccall "read"+ c_read :: FD -> Ptr Word8 -> CInt -> IO CInt++foreign import ccall "close"+ close :: FD -> IO ()++newtype FD = FD CInt++openRead :: FilePath -> IO FD+openRead fp = do+ h <- withCString fp $ \str -> c_open str oRdonly+ if h < 0+ then throwErrno $ "Could not open file: " ++ fp+ else return $ FD h++read :: FD -> IO (SourceResult S.ByteString)+read fd = do+ cstr <- mallocBytes 4096+ len <- c_read fd cstr 4096+ if len == 0+ then free cstr >> return Closed+ else fmap Open $ BU.unsafePackCStringFinalizer+ cstr+ (fromIntegral len)+ (free cstr)
System/Win32File.hsc view
@@ -1,89 +1,89 @@-{-# LANGUAGE ForeignFunctionInterface #-} -{-# LANGUAGE GeneralizedNewtypeDeriving #-} -module System.Win32File - ( openRead - , read - , close - ) where - -import Foreign.C.String (CString) -import Foreign.Marshal.Alloc (mallocBytes, free) -import Foreign.C.Types (CInt) -import Foreign.C.Error (throwErrno) -import Foreign.Ptr (Ptr) -import Data.Bits (Bits, (.|.)) -import qualified Data.ByteString as S -import qualified Data.ByteString.Unsafe as BU -import Data.Text (pack) -import Data.Text.Encoding (encodeUtf16LE) -import Data.Word (Word8) -import Prelude hiding (read) -import Data.Conduit (SourceResult (..)) - -#include <fcntl.h> -#include <Share.h> -#include <SYS/Stat.h> -#include <errno.h> - -newtype OFlag = OFlag CInt - deriving (Num, Bits, Show, Eq) - -#{enum OFlag, OFlag - , oBinary = _O_BINARY - , oRdonly = _O_RDONLY - } - -newtype SHFlag = SHFlag CInt - deriving (Num, Bits, Show, Eq) - -#{enum SHFlag, SHFlag - , shDenyno = _SH_DENYNO - } - -newtype PMode = PMode CInt - deriving (Num, Bits, Show, Eq) - -#{enum PMode, PMode - , pIread = _S_IREAD - } - -foreign import ccall "_wsopen" - c_wsopen :: CString -> OFlag -> SHFlag -> PMode -> IO CInt - -foreign import ccall "_read" - c_read :: FD -> Ptr Word8 -> CInt -> IO CInt - -foreign import ccall "_close" - close :: FD -> IO () - -newtype FD = FD CInt - -openRead :: FilePath -> IO FD -openRead fp = do - -- need to append a null char - -- note that useAsCString is not sufficient, as we need to have two - -- null octets to account for UTF16 encoding - let bs = encodeUtf16LE $ pack $ fp ++ "\0" - h <- BU.unsafeUseAsCString bs $ \str -> - c_wsopen - str - (oBinary .|. oRdonly) - shDenyno - pIread - if h < 0 - then throwErrno $ "Could not open file: " ++ fp - else return $ FD h - -read :: FD -> IO (SourceResult S.ByteString) -read fd = do - cstr <- mallocBytes 4096 - len <- c_read fd cstr 4096 - if len == 0 - then do - free cstr - return Closed - else do - fmap Open $ BU.unsafePackCStringFinalizer - cstr - (fromIntegral len) - (free cstr) +{-# LANGUAGE ForeignFunctionInterface #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+module System.Win32File+ ( openRead+ , read+ , close+ ) where++import Foreign.C.String (CString)+import Foreign.Marshal.Alloc (mallocBytes, free)+import Foreign.C.Types (CInt)+import Foreign.C.Error (throwErrno)+import Foreign.Ptr (Ptr)+import Data.Bits (Bits, (.|.))+import qualified Data.ByteString as S+import qualified Data.ByteString.Unsafe as BU+import Data.Text (pack)+import Data.Text.Encoding (encodeUtf16LE)+import Data.Word (Word8)+import Prelude hiding (read)+import Data.Conduit (SourceResult (..))++#include <fcntl.h>+#include <Share.h>+#include <SYS/Stat.h>+#include <errno.h>++newtype OFlag = OFlag CInt+ deriving (Num, Bits, Show, Eq)++#{enum OFlag, OFlag+ , oBinary = _O_BINARY+ , oRdonly = _O_RDONLY+ }++newtype SHFlag = SHFlag CInt+ deriving (Num, Bits, Show, Eq)++#{enum SHFlag, SHFlag+ , shDenyno = _SH_DENYNO+ }++newtype PMode = PMode CInt+ deriving (Num, Bits, Show, Eq)++#{enum PMode, PMode+ , pIread = _S_IREAD+ }++foreign import ccall "_wsopen"+ c_wsopen :: CString -> OFlag -> SHFlag -> PMode -> IO CInt++foreign import ccall "_read"+ c_read :: FD -> Ptr Word8 -> CInt -> IO CInt++foreign import ccall "_close"+ close :: FD -> IO ()++newtype FD = FD CInt++openRead :: FilePath -> IO FD+openRead fp = do+ -- need to append a null char+ -- note that useAsCString is not sufficient, as we need to have two+ -- null octets to account for UTF16 encoding+ let bs = encodeUtf16LE $ pack $ fp ++ "\0"+ h <- BU.unsafeUseAsCString bs $ \str ->+ c_wsopen+ str+ (oBinary .|. oRdonly)+ shDenyno+ pIread+ if h < 0+ then throwErrno $ "Could not open file: " ++ fp+ else return $ FD h++read :: FD -> IO (SourceResult S.ByteString)+read fd = do+ cstr <- mallocBytes 4096+ len <- c_read fd cstr 4096+ if len == 0+ then do+ free cstr+ return Closed+ else do+ fmap Open $ BU.unsafePackCStringFinalizer+ cstr+ (fromIntegral len)+ (free cstr)
conduit.cabal view
@@ -1,68 +1,75 @@-Name: conduit -Version: 0.0.4 -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>. -License: BSD3 -License-file: LICENSE -Author: Michael Snoyman -Maintainer: michael@snoyman.com -Category: Data, Conduit -Build-type: Simple -Cabal-version: >=1.8 -Homepage: http://github.com/snoyberg/conduit -extra-source-files: test/main.hs, test/random - -flag debug - -flag nohandles - -Library - if os(windows) - cpp-options: -DCABAL_OS_WINDOWS - other-modules: System.Win32File - else - other-modules: System.PosixFile - if flag(nohandles) - cpp-options: -DNO_HANDLES - Exposed-modules: Data.Conduit - Data.Conduit.Binary - 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 - Data.Conduit.Util.Source - Data.Conduit.Util.Sink - Data.Conduit.Util.Conduit - Build-depends: base >= 4.3 && < 5 - , lifted-base >= 0.1 && < 0.2 - , transformers-base >= 0.4.1 && < 0.5 - , monad-control >= 0.3.1 && < 0.4 - , containers - , transformers >= 0.2.2 && < 0.3 - , bytestring >= 0.9 - , text >= 0.11 - ghc-options: -Wall - if flag(debug) - cpp-options: -DDEBUG - -test-suite test - hs-source-dirs: test - main-is: main.hs - type: exitcode-stdio-1.0 - cpp-options: -DTEST - build-depends: conduit - , base - , hspec - , HUnit - , QuickCheck - , bytestring - , transformers - , text - ghc-options: -Wall - -source-repository head - type: git - location: git://github.com/snoyberg/conduit.git +Name: conduit+Version: 0.1.0+Synopsis: Streaming data processing library.+Description:+ Conduits are an approach to the streaming data problem. It is meant as an alternative to enumerators\/iterators, hoping to address the same issues with different trade-offs based on real-world experience with enumerators. For more information, see <http://www.yesodweb.com/book/conduit>.+ .+ Release history:+ .+ * 0.1: @BufferedSource@ is now an abstract type, and has a much more efficient internal representation. The result was a 41% speedup on microbenchmarks (note: do not expect speedups anywhere near that in real usage). In general, we are moving towards @BufferedSource@ being a specific tool used internally as needed, but using @Source@ for all external APIs.+ .+ * 0.0: Initial release.+License: BSD3+License-file: LICENSE+Author: Michael Snoyman+Maintainer: michael@snoyman.com+Category: Data, Conduit+Build-type: Simple+Cabal-version: >=1.8+Homepage: http://github.com/snoyberg/conduit+extra-source-files: test/main.hs, test/random++flag debug++flag nohandles++Library+ if os(windows)+ cpp-options: -DCABAL_OS_WINDOWS+ other-modules: System.Win32File+ else+ other-modules: System.PosixFile+ if flag(nohandles)+ cpp-options: -DNO_HANDLES+ Exposed-modules: Data.Conduit+ Data.Conduit.Binary+ 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+ Data.Conduit.Util.Source+ Data.Conduit.Util.Sink+ Data.Conduit.Util.Conduit+ Build-depends: base >= 4.3 && < 5+ , lifted-base >= 0.1 && < 0.2+ , transformers-base >= 0.4.1 && < 0.5+ , monad-control >= 0.3.1 && < 0.4+ , containers+ , transformers >= 0.2.2 && < 0.3+ , bytestring >= 0.9+ , text >= 0.11+ ghc-options: -Wall+ if flag(debug)+ cpp-options: -DDEBUG++test-suite test+ hs-source-dirs: test+ main-is: main.hs+ type: exitcode-stdio-1.0+ cpp-options: -DTEST+ build-depends: conduit+ , base+ , hspec+ , HUnit+ , QuickCheck+ , bytestring+ , transformers+ , text+ ghc-options: -Wall++source-repository head+ type: git+ location: git://github.com/snoyberg/conduit.git
test/main.hs view
@@ -1,389 +1,390 @@-{-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE CPP #-} -import Test.Hspec.Monadic -import Test.Hspec.HUnit () -import Test.Hspec.QuickCheck (prop) -import Test.HUnit - -import qualified Data.Conduit as C -import qualified Data.Conduit.List as CL -import qualified Data.Conduit.Lazy as CLazy -import qualified Data.Conduit.Binary as CB -import qualified Data.Conduit.Text as CT -import Data.Conduit (runResourceT) -import qualified Data.List as DL -import Control.Monad.ST (runST) -import Data.Monoid -import qualified Data.ByteString as S -import qualified Data.IORef as I -import qualified Data.ByteString.Lazy as L -import Data.ByteString.Lazy.Char8 () -import Control.Monad.Trans.Writer (Writer) -import qualified Data.Text as T -import qualified Data.Text.Lazy as TL -import qualified Data.Text.Lazy.Encoding as TLE -import Control.Monad.Trans.Resource (runExceptionT_, withIO, resourceForkIO) -import Control.Concurrent (threadDelay, killThread) -import Control.Monad.IO.Class (liftIO) -import Control.Applicative (pure, (<$>), (<*>)) - -main :: IO () -main = hspecX $ do - describe "data loss rules" $ do - it "consumes the source to quickly" $ do - x <- runResourceT $ CL.sourceList [1..10 :: Int] C.$$ do - strings <- CL.map show C.=$ CL.take 5 - liftIO $ putStr $ unlines strings - CL.fold (+) 0 - 40 @?= x - - it "correctly consumes a chunked resource" $ do - x <- runResourceT $ (CL.sourceList [1..5 :: Int] `mappend` CL.sourceList [6..10]) C.$$ do - strings <- CL.map show C.=$ CL.take 5 - liftIO $ putStr $ unlines strings - CL.fold (+) 0 - 40 @?= x - - describe "filter" $ do - it "even" $ do - x <- runResourceT $ CL.sourceList [1..10] C.$$ CL.filter even C.=$ CL.consume - x @?= filter even [1..10 :: Int] - - describe "ResourceT" $ do - it "resourceForkIO" $ do - counter <- I.newIORef 0 - let w = withIO - (I.atomicModifyIORef counter $ \i -> - (i + 1, ())) - (const $ I.atomicModifyIORef counter $ \i -> - (i - 1, ())) - runResourceT $ do - _ <- w - _ <- resourceForkIO $ return () - _ <- resourceForkIO $ return () - sequence_ $ replicate 1000 $ do - tid <- resourceForkIO $ return () - liftIO $ killThread tid - _ <- resourceForkIO $ return () - _ <- resourceForkIO $ return () - return () - - -- give enough of a chance to the cleanup code to finish - threadDelay 1000 - res <- I.readIORef counter - res @?= (0 :: Int) - - describe "sum" $ do - it "works for 1..10" $ do - 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)) - == sum list - - describe "Monoid instance for Source" $ do - it "mappend" $ do - x <- runResourceT $ (CL.sourceList [1..5 :: Int] `mappend` CL.sourceList [6..10]) C.$$ CL.fold (+) 0 - x @?= sum [1..10] - it "mconcat" $ do - x <- runResourceT $ mconcat - [ CL.sourceList [1..5 :: Int] - , CL.sourceList [6..10] - , CL.sourceList [11..20] - ] C.$$ CL.fold (+) 0 - x @?= sum [1..20] - - describe "file access" $ do - it "read" $ do - bs <- S.readFile "conduit.cabal" - bss <- runResourceT $ CB.sourceFile "conduit.cabal" C.$$ CL.consume - bs @=? S.concat bss - - it "read range" $ do - S.writeFile "tmp" "0123456789" - bss <- runResourceT $ CB.sourceFileRange "tmp" (Just 2) (Just 3) C.$$ CL.consume - S.concat bss @?= "234" - - it "write" $ do - runResourceT $ CB.sourceFile "conduit.cabal" C.$$ CB.sinkFile "tmp" - bs1 <- S.readFile "conduit.cabal" - bs2 <- S.readFile "tmp" - bs1 @=? bs2 - - it "conduit" $ do - runResourceT $ CB.sourceFile "conduit.cabal" - C.$= CB.conduitFile "tmp" - C.$$ CB.sinkFile "tmp2" - bs1 <- S.readFile "conduit.cabal" - bs2 <- S.readFile "tmp" - bs3 <- S.readFile "tmp2" - bs1 @=? bs2 - bs1 @=? bs3 - - describe "Monad instance for Sink" $ do - it "binding" $ do - x <- runResourceT $ CL.sourceList [1..10] C.$$ do - _ <- CL.take 5 - CL.fold (+) (0 :: Int) - x @?= sum [6..10] - - describe "Applicative instance for Sink" $ do - it "<$> and <*>" $ do - x <- runResourceT $ CL.sourceList [1..10] C.$$ - (+) <$> pure 5 <*> CL.fold (+) (0 :: Int) - x @?= sum [1..10] + 5 - - describe "resumable sources" $ do - it "simple" $ do - (x, y, z) <- runResourceT $ do - bs <- C.bufferSource $ CL.sourceList [1..10 :: Int] - x <- bs C.$$ CL.take 5 - y <- bs C.$$ CL.fold (+) 0 - z <- bs C.$$ CL.consume - return (x, y, z) - x @?= [1..5] :: IO () - y @?= sum [6..10] - z @?= [] - - describe "conduits" $ do - it "map, left" $ do - x <- runResourceT $ - CL.sourceList [1..10] - C.$= CL.map (* 2) - C.$$ CL.fold (+) 0 - x @?= 2 * sum [1..10 :: Int] - - it "map, right" $ do - x <- runResourceT $ - CL.sourceList [1..10] - C.$$ CL.map (* 2) - C.=$ CL.fold (+) 0 - x @?= 2 * sum [1..10 :: Int] - - it "groupBy" $ do - let input = [1::Int, 1, 2, 3, 3, 3, 4, 5, 5] - x <- runResourceT $ CL.sourceList input - C.$$ CL.groupBy (==) - C.=$ CL.consume - x @?= DL.groupBy (==) input - - it "groupBy (nondup begin/end)" $ do - let input = [1::Int, 2, 3, 3, 3, 4, 5] - x <- runResourceT $ CL.sourceList input - C.$$ CL.groupBy (==) - C.=$ CL.consume - x @?= DL.groupBy (==) input - - it "concatMap" $ do - let input = [1, 11, 21] - x <- runResourceT $ CL.sourceList input - C.$$ CL.concatMap (\i -> enumFromTo i (i + 9)) - C.=$ CL.fold (+) (0 :: Int) - x @?= sum [1..30] - - it "bind together" $ do - let conduit = CL.map (+ 5) C.=$= CL.map (* 2) - x <- runResourceT $ CL.sourceList [1..10] C.$= conduit C.$$ CL.fold (+) 0 - x @?= sum (map (* 2) $ map (+ 5) [1..10 :: Int]) - -#if !FAST - describe "isolate" $ do - it "bound to resumable source" $ do - (x, y) <- runResourceT $ do - bsrc <- C.bufferSource $ CL.sourceList [1..10 :: Int] - x <- bsrc C.$= CL.isolate 5 C.$$ CL.consume - y <- bsrc C.$$ CL.consume - return (x, y) - x @?= [1..5] - y @?= [6..10] - - it "bound to sink, non-resumable" $ do - (x, y) <- runResourceT $ do - CL.sourceList [1..10 :: Int] C.$$ do - x <- CL.isolate 5 C.=$ CL.consume - y <- CL.consume - return (x, y) - x @?= [1..5] - y @?= [6..10] - - it "bound to sink, resumable" $ do - (x, y) <- runResourceT $ do - bsrc <- C.bufferSource $ CL.sourceList [1..10 :: Int] - x <- bsrc C.$$ CL.isolate 5 C.=$ CL.consume - y <- bsrc C.$$ CL.consume - return (x, y) - x @?= [1..5] - y @?= [6..10] - - it "consumes all data" $ do - x <- runResourceT $ CL.sourceList [1..10 :: Int] C.$$ do - CL.isolate 5 C.=$ CL.sinkNull - CL.consume - x @?= [6..10] - - describe "lazy" $ do - it' "works inside a ResourceT" $ runResourceT $ do - counter <- liftIO $ I.newIORef 0 - let incr i = C.sourceIO - (liftIO $ I.newIORef $ C.Open (i :: Int)) - (const $ return ()) - (\istate -> do - res <- liftIO $ I.atomicModifyIORef istate - (\state -> (C.Closed, state)) - case res of - C.Closed -> return () - _ -> do - count <- liftIO $ I.atomicModifyIORef counter - (\j -> (j + 1, j + 1)) - liftIO $ count @?= i - return res - ) - nums <- CLazy.lazyConsume $ mconcat $ map incr [1..10] - liftIO $ nums @?= [1..10] - - describe "sequenceSink" $ do - it "simple sink" $ do - let sink () = do - _ <- CL.drop 2 - x <- CL.head - return $ C.Emit () $ maybe [] return x - let conduit = C.sequenceSink () sink - res <- runResourceT $ CL.sourceList [1..10 :: Int] - C.$= conduit - C.$$ CL.consume - res @?= [3, 6, 9] - it "finishes on new state" $ do - let sink () = do - x <- CL.head - return $ C.Emit () $ maybe [] return x - let conduit = C.sequenceSink () sink - res <- runResourceT $ CL.sourceList [1..10 :: Int] - C.$= conduit C.$$ CL.consume - res @?= [1..10] - it "switch to a conduit" $ do - let sink () = do - _ <- CL.drop 4 - return $ C.StartConduit $ CL.filter even - let conduit = C.sequenceSink () sink - res <- runResourceT $ CL.sourceList [1..10 :: Int] - C.$= conduit - C.$$ CL.consume - res @?= [6, 8, 10] -#endif - - describe "peek" $ do - it "works" $ do - (a, b) <- runResourceT $ CL.sourceList [1..10 :: Int] C.$$ do - a <- CL.peek - b <- CL.consume - return (a, b) - (a, b) @?= (Just 1, [1..10]) - - describe "text" $ do - let go enc tenc cenc = do - prop (enc ++ " single chunk") $ \chars -> runST $ runExceptionT_ $ runResourceT $ 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 - 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 - let tss = map T.pack chars - lbs = tenc $ TL.fromChunks tss - src = mconcat $ map (CL.sourceList . return) tss - bss <- src C.$= CT.encode cenc C.$$ CL.consume - return $ L.fromChunks bss == lbs - go "utf8" TLE.encodeUtf8 CT.utf8 - go "utf16_le" TLE.encodeUtf16LE CT.utf16_le - go "utf16_be" TLE.encodeUtf16BE CT.utf16_be - go "utf32_le" TLE.encodeUtf32LE CT.utf32_le - go "utf32_be" TLE.encodeUtf32BE CT.utf32_be - - describe "binary isolate" $ do - it "works" $ do - bss <- runResourceT $ CL.sourceList (replicate 1000 "X") - C.$= CB.isolate 6 - C.$$ CL.consume - S.concat bss @?= "XXXXXX" - describe "unbuffering" $ do - it "works" $ do - x <- runResourceT $ do - bsrc <- C.bufferSource $ CL.sourceList [1..10 :: Int] - bsrc C.$$ CL.drop 5 - let src = C.unbufferSource bsrc - src C.$$ CL.fold (+) 0 - x @?= sum [6..10] - - describe "properly using binary file reading" $ do - it "sourceFile" $ do - x <- runResourceT $ CB.sourceFile "test/random" C.$$ CL.consume - lbs <- L.readFile "test/random" - L.fromChunks x @?= lbs - - describe "binary head" $ do - let go lbs = do - x <- CB.head - case (x, L.uncons lbs) of - (Nothing, Nothing) -> return True - (Just y, Just (z, lbs')) - | y == z -> go lbs' - _ -> return False - - prop "works" $ \bss' -> - let bss = map S.pack bss' - in runST $ runResourceT $ - 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 - 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 - bss2 <- CL.sourceList bss C.$$ do - CB.dropWhile (< 5) - CL.consume - return $ L.fromChunks bss2 == L.dropWhile (< 5) (L.fromChunks bss) - - describe "binary take" $ do - let go n l = CL.sourceList l C.$$ do - a <- CB.take n - b <- CL.consume - return (a, b) - - -- Taking nothing should result in an empty Bytestring - it "nothing" $ do - (a, b) <- runResourceT $ go 0 ["abc", "defg"] - a @?= L.empty - L.fromChunks b @?= "abcdefg" - - it "normal" $ do - (a, b) <- runResourceT $ go 4 ["abc", "defg"] - a @?= "abcd" - L.fromChunks b @?= "efg" - - -- Taking exactly the data that is available should result in no - -- leftover. - it "all" $ do - (a, b) <- runResourceT $ go 7 ["abc", "defg"] - a @?= "abcdefg" - b @?= [] - - -- Take as much as possible. - it "more" $ do - (a, b) <- runResourceT $ go 10 ["abc", "defg"] - a @?= "abcdefg" - b @?= [] - -it' :: String -> IO () -> Writer [Spec] () -it' = it +{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE CPP #-}+import Test.Hspec.Monadic+import Test.Hspec.HUnit ()+import Test.Hspec.QuickCheck (prop)+import Test.HUnit++import qualified Data.Conduit as C+import qualified Data.Conduit.List as CL+import qualified Data.Conduit.Lazy as CLazy+import qualified Data.Conduit.Binary as CB+import qualified Data.Conduit.Text as CT+import Data.Conduit (runResourceT)+import qualified Data.List as DL+import Control.Monad.ST (runST)+import Data.Monoid+import qualified Data.ByteString as S+import qualified Data.IORef as I+import qualified Data.ByteString.Lazy as L+import Data.ByteString.Lazy.Char8 ()+import Control.Monad.Trans.Writer (Writer)+import qualified Data.Text as T+import qualified Data.Text.Lazy as TL+import qualified Data.Text.Lazy.Encoding as TLE+import Control.Monad.Trans.Resource (runExceptionT_, withIO, resourceForkIO)+import Control.Concurrent (threadDelay, killThread)+import Control.Monad.IO.Class (liftIO)+import Control.Applicative (pure, (<$>), (<*>))++main :: IO ()+main = hspecX $ do+ describe "data loss rules" $ do+ it "consumes the source to quickly" $ do+ x <- runResourceT $ CL.sourceList [1..10 :: Int] C.$$ do+ strings <- CL.map show C.=$ CL.take 5+ liftIO $ putStr $ unlines strings+ CL.fold (+) 0+ 40 @?= x++ it "correctly consumes a chunked resource" $ do+ x <- runResourceT $ (CL.sourceList [1..5 :: Int] `mappend` CL.sourceList [6..10]) C.$$ do+ strings <- CL.map show C.=$ CL.take 5+ liftIO $ putStr $ unlines strings+ CL.fold (+) 0+ 40 @?= x++ describe "filter" $ do+ it "even" $ do+ x <- runResourceT $ CL.sourceList [1..10] C.$$ CL.filter even C.=$ CL.consume+ x @?= filter even [1..10 :: Int]++ describe "ResourceT" $ do+ it "resourceForkIO" $ do+ counter <- I.newIORef 0+ let w = withIO+ (I.atomicModifyIORef counter $ \i ->+ (i + 1, ()))+ (const $ I.atomicModifyIORef counter $ \i ->+ (i - 1, ()))+ runResourceT $ do+ _ <- w+ _ <- resourceForkIO $ return ()+ _ <- resourceForkIO $ return ()+ sequence_ $ replicate 1000 $ do+ tid <- resourceForkIO $ return ()+ liftIO $ killThread tid+ _ <- resourceForkIO $ return ()+ _ <- resourceForkIO $ return ()+ return ()++ -- give enough of a chance to the cleanup code to finish+ threadDelay 1000+ res <- I.readIORef counter+ res @?= (0 :: Int)++ describe "sum" $ do+ it "works for 1..10" $ do+ 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))+ == sum list++ describe "Monoid instance for Source" $ do+ it "mappend" $ do+ x <- runResourceT $ (CL.sourceList [1..5 :: Int] `mappend` CL.sourceList [6..10]) C.$$ CL.fold (+) 0+ x @?= sum [1..10]+ it "mconcat" $ do+ x <- runResourceT $ mconcat+ [ CL.sourceList [1..5 :: Int]+ , CL.sourceList [6..10]+ , CL.sourceList [11..20]+ ] C.$$ CL.fold (+) 0+ x @?= sum [1..20]++ describe "file access" $ do+ it "read" $ do+ bs <- S.readFile "conduit.cabal"+ bss <- runResourceT $ CB.sourceFile "conduit.cabal" C.$$ CL.consume+ bs @=? S.concat bss++ it "read range" $ do+ S.writeFile "tmp" "0123456789"+ bss <- runResourceT $ CB.sourceFileRange "tmp" (Just 2) (Just 3) C.$$ CL.consume+ S.concat bss @?= "234"++ it "write" $ do+ runResourceT $ CB.sourceFile "conduit.cabal" C.$$ CB.sinkFile "tmp"+ bs1 <- S.readFile "conduit.cabal"+ bs2 <- S.readFile "tmp"+ bs1 @=? bs2++ it "conduit" $ do+ runResourceT $ CB.sourceFile "conduit.cabal"+ C.$= CB.conduitFile "tmp"+ C.$$ CB.sinkFile "tmp2"+ bs1 <- S.readFile "conduit.cabal"+ bs2 <- S.readFile "tmp"+ bs3 <- S.readFile "tmp2"+ bs1 @=? bs2+ bs1 @=? bs3++ describe "Monad instance for Sink" $ do+ it "binding" $ do+ x <- runResourceT $ CL.sourceList [1..10] C.$$ do+ _ <- CL.take 5+ CL.fold (+) (0 :: Int)+ x @?= sum [6..10]++ describe "Applicative instance for Sink" $ do+ it "<$> and <*>" $ do+ x <- runResourceT $ CL.sourceList [1..10] C.$$+ (+) <$> pure 5 <*> CL.fold (+) (0 :: Int)+ x @?= sum [1..10] + 5++ describe "resumable sources" $ do+ it "simple" $ do+ (x, y, z) <- runResourceT $ do+ bs <- C.bufferSource $ CL.sourceList [1..10 :: Int]+ x <- bs C.$$ CL.take 5+ y <- bs C.$$ CL.fold (+) 0+ z <- bs C.$$ CL.consume+ C.bsourceClose bs+ return (x, y, z)+ x @?= [1..5] :: IO ()+ y @?= sum [6..10]+ z @?= []++ describe "conduits" $ do+ it "map, left" $ do+ x <- runResourceT $+ CL.sourceList [1..10]+ C.$= CL.map (* 2)+ C.$$ CL.fold (+) 0+ x @?= 2 * sum [1..10 :: Int]++ it "map, right" $ do+ x <- runResourceT $+ CL.sourceList [1..10]+ C.$$ CL.map (* 2)+ C.=$ CL.fold (+) 0+ x @?= 2 * sum [1..10 :: Int]++ it "groupBy" $ do+ let input = [1::Int, 1, 2, 3, 3, 3, 4, 5, 5]+ x <- runResourceT $ CL.sourceList input+ C.$$ CL.groupBy (==)+ C.=$ CL.consume+ x @?= DL.groupBy (==) input++ it "groupBy (nondup begin/end)" $ do+ let input = [1::Int, 2, 3, 3, 3, 4, 5]+ x <- runResourceT $ CL.sourceList input+ C.$$ CL.groupBy (==)+ C.=$ CL.consume+ x @?= DL.groupBy (==) input++ it "concatMap" $ do+ let input = [1, 11, 21]+ x <- runResourceT $ CL.sourceList input+ C.$$ CL.concatMap (\i -> enumFromTo i (i + 9))+ C.=$ CL.fold (+) (0 :: Int)+ x @?= sum [1..30]++ it "bind together" $ do+ let conduit = CL.map (+ 5) C.=$= CL.map (* 2)+ x <- runResourceT $ CL.sourceList [1..10] C.$= conduit C.$$ CL.fold (+) 0+ x @?= sum (map (* 2) $ map (+ 5) [1..10 :: Int])++#if !FAST+ describe "isolate" $ do+ it "bound to resumable source" $ do+ (x, y) <- runResourceT $ do+ bsrc <- C.bufferSource $ CL.sourceList [1..10 :: Int]+ x <- bsrc C.$= CL.isolate 5 C.$$ CL.consume+ y <- bsrc C.$$ CL.consume+ return (x, y)+ x @?= [1..5]+ y @?= [6..10]++ it "bound to sink, non-resumable" $ do+ (x, y) <- runResourceT $ do+ CL.sourceList [1..10 :: Int] C.$$ do+ x <- CL.isolate 5 C.=$ CL.consume+ y <- CL.consume+ return (x, y)+ x @?= [1..5]+ y @?= [6..10]++ it "bound to sink, resumable" $ do+ (x, y) <- runResourceT $ do+ bsrc <- C.bufferSource $ CL.sourceList [1..10 :: Int]+ x <- bsrc C.$$ CL.isolate 5 C.=$ CL.consume+ y <- bsrc C.$$ CL.consume+ return (x, y)+ x @?= [1..5]+ y @?= [6..10]++ it "consumes all data" $ do+ x <- runResourceT $ CL.sourceList [1..10 :: Int] C.$$ do+ CL.isolate 5 C.=$ CL.sinkNull+ CL.consume+ x @?= [6..10]++ describe "lazy" $ do+ it' "works inside a ResourceT" $ runResourceT $ do+ counter <- liftIO $ I.newIORef 0+ let incr i = C.sourceIO+ (liftIO $ I.newIORef $ C.Open (i :: Int))+ (const $ return ())+ (\istate -> do+ res <- liftIO $ I.atomicModifyIORef istate+ (\state -> (C.Closed, state))+ case res of+ C.Closed -> return ()+ _ -> do+ count <- liftIO $ I.atomicModifyIORef counter+ (\j -> (j + 1, j + 1))+ liftIO $ count @?= i+ return res+ )+ nums <- CLazy.lazyConsume $ mconcat $ map incr [1..10]+ liftIO $ nums @?= [1..10]++ describe "sequenceSink" $ do+ it "simple sink" $ do+ let sink () = do+ _ <- CL.drop 2+ x <- CL.head+ return $ C.Emit () $ maybe [] return x+ let conduit = C.sequenceSink () sink+ res <- runResourceT $ CL.sourceList [1..10 :: Int]+ C.$= conduit+ C.$$ CL.consume+ res @?= [3, 6, 9]+ it "finishes on new state" $ do+ let sink () = do+ x <- CL.head+ return $ C.Emit () $ maybe [] return x+ let conduit = C.sequenceSink () sink+ res <- runResourceT $ CL.sourceList [1..10 :: Int]+ C.$= conduit C.$$ CL.consume+ res @?= [1..10]+ it "switch to a conduit" $ do+ let sink () = do+ _ <- CL.drop 4+ return $ C.StartConduit $ CL.filter even+ let conduit = C.sequenceSink () sink+ res <- runResourceT $ CL.sourceList [1..10 :: Int]+ C.$= conduit+ C.$$ CL.consume+ res @?= [6, 8, 10]+#endif++ describe "peek" $ do+ it "works" $ do+ (a, b) <- runResourceT $ CL.sourceList [1..10 :: Int] C.$$ do+ a <- CL.peek+ b <- CL.consume+ return (a, b)+ (a, b) @?= (Just 1, [1..10])++ describe "text" $ do+ let go enc tenc cenc = do+ prop (enc ++ " single chunk") $ \chars -> runST $ runExceptionT_ $ runResourceT $ 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+ 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+ let tss = map T.pack chars+ lbs = tenc $ TL.fromChunks tss+ src = mconcat $ map (CL.sourceList . return) tss+ bss <- src C.$= CT.encode cenc C.$$ CL.consume+ return $ L.fromChunks bss == lbs+ go "utf8" TLE.encodeUtf8 CT.utf8+ go "utf16_le" TLE.encodeUtf16LE CT.utf16_le+ go "utf16_be" TLE.encodeUtf16BE CT.utf16_be+ go "utf32_le" TLE.encodeUtf32LE CT.utf32_le+ go "utf32_be" TLE.encodeUtf32BE CT.utf32_be++ describe "binary isolate" $ do+ it "works" $ do+ bss <- runResourceT $ CL.sourceList (replicate 1000 "X")+ C.$= CB.isolate 6+ C.$$ CL.consume+ S.concat bss @?= "XXXXXX"+ describe "unbuffering" $ do+ it "works" $ do+ x <- runResourceT $ do+ bsrc <- C.bufferSource $ CL.sourceList [1..10 :: Int]+ bsrc C.$$ CL.drop 5+ let src = C.unbufferSource bsrc+ src C.$$ CL.fold (+) 0+ x @?= sum [6..10]++ describe "properly using binary file reading" $ do+ it "sourceFile" $ do+ x <- runResourceT $ CB.sourceFile "test/random" C.$$ CL.consume+ lbs <- L.readFile "test/random"+ L.fromChunks x @?= lbs++ describe "binary head" $ do+ let go lbs = do+ x <- CB.head+ case (x, L.uncons lbs) of+ (Nothing, Nothing) -> return True+ (Just y, Just (z, lbs'))+ | y == z -> go lbs'+ _ -> return False++ prop "works" $ \bss' ->+ let bss = map S.pack bss'+ in runST $ runResourceT $+ 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+ 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+ bss2 <- CL.sourceList bss C.$$ do+ CB.dropWhile (< 5)+ CL.consume+ return $ L.fromChunks bss2 == L.dropWhile (< 5) (L.fromChunks bss)++ describe "binary take" $ do+ let go n l = CL.sourceList l C.$$ do+ a <- CB.take n+ b <- CL.consume+ return (a, b)++ -- Taking nothing should result in an empty Bytestring+ it "nothing" $ do+ (a, b) <- runResourceT $ go 0 ["abc", "defg"]+ a @?= L.empty+ L.fromChunks b @?= "abcdefg"++ it "normal" $ do+ (a, b) <- runResourceT $ go 4 ["abc", "defg"]+ a @?= "abcd"+ L.fromChunks b @?= "efg"++ -- Taking exactly the data that is available should result in no+ -- leftover.+ it "all" $ do+ (a, b) <- runResourceT $ go 7 ["abc", "defg"]+ a @?= "abcdefg"+ b @?= []++ -- Take as much as possible.+ it "more" $ do+ (a, b) <- runResourceT $ go 10 ["abc", "defg"]+ a @?= "abcdefg"+ b @?= []++it' :: String -> IO () -> Writer [Spec] ()+it' = it