conduit 0.0.3 → 0.0.4
raw patch · 18 files changed
+3197/−3172 lines, 18 filessetup-changedPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
+ Data.Conduit: unsafeBufferSource :: (BufferSource s, Resource m) => s m a -> ResourceT m (BufferedSource m a)
+ Data.Conduit.List: sourceNull :: Resource m => Source m a
Files
- Control/Monad/Trans/Resource.hs +530/−530
- Data/Conduit.hs +239/−236
- Data/Conduit/Binary.hs +242/−242
- Data/Conduit/Lazy.hs +28/−28
- Data/Conduit/List.hs +290/−281
- 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 +247/−234
- 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 +68/−68
- test/main.hs +389/−389
Control/Monad/Trans/Resource.hs view
@@ -1,530 +1,530 @@-{-# 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 + , 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))
Data/Conduit.hs view
@@ -1,236 +1,239 @@-{-# 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-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 <- bufferSource 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--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 <- bufferSource 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 + -- ** 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'
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,281 +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- -- * 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)---- | 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 ())+{-# 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,234 +1,247 @@-{-# 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)---- | 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- }--instance BufferSource Source where- bufferSource (Source msrc) = msrc >>= bufferSource---- | 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 (..) + , 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
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,68 @@-Name: conduit-Version: 0.0.3-Synopsis: A pull-based approach to streaming data.-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/blog/2011/12/conduits>.-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.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
test/main.hs view
@@ -1,389 +1,389 @@-{-# 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 + 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