conduit (empty) → 0.0.0
raw patch · 16 files changed
+2663/−0 lines, 16 filesdep +HUnitdep +QuickCheckdep +basesetup-changed
Dependencies added: HUnit, QuickCheck, base, bytestring, conduit, containers, hspec, lifted-base, monad-control, text, transformers, transformers-base
Files
- Control/Monad/Trans/Resource.hs +519/−0
- Data/Conduit.hs +236/−0
- Data/Conduit/Binary.hs +116/−0
- Data/Conduit/Lazy.hs +25/−0
- Data/Conduit/List.hs +230/−0
- Data/Conduit/Text.hs +295/−0
- Data/Conduit/Types/Conduit.hs +46/−0
- Data/Conduit/Types/Sink.hs +192/−0
- Data/Conduit/Types/Source.hs +207/−0
- Data/Conduit/Util/Conduit.hs +189/−0
- Data/Conduit/Util/Sink.hs +101/−0
- Data/Conduit/Util/Source.hs +100/−0
- LICENSE +30/−0
- Setup.lhs +7/−0
- conduit.cabal +59/−0
- test/main.hs +311/−0
+ Control/Monad/Trans/Resource.hs view
@@ -0,0 +1,519 @@+{-# 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'++-- | Read a value from a reference.+readRef :: Resource m => Ref (Base m) a -> ResourceT m a+readRef = lift . resourceLiftBase . readRef'++-- | Write a value to a reference.+writeRef :: Resource m => Ref (Base m) a -> a -> ResourceT m ()+writeRef r = lift . resourceLiftBase . writeRef' r++-- | 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++-- | 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+ modifyRef' = I.atomicModifyIORef+ readRef' = I.readIORef+ writeRef' = I.writeIORef+ mask = E.mask+ mask_ = E.mask_+ try = E.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
@@ -0,0 +1,236 @@+{-# 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'
+ Data/Conduit/Binary.hs view
@@ -0,0 +1,116 @@+{-# LANGUAGE FlexibleContexts #-}+-- | Functions for interacting with bytes.+module Data.Conduit.Binary+ ( sourceFile+ , sourceFileRange+ , sinkFile+ , conduitFile+ , isolate+ ) where++import qualified Data.ByteString as S+import Data.Conduit+import Control.Exception (assert)+import Control.Monad.IO.Class (liftIO)+import qualified System.IO as IO+import Control.Monad.Trans.Resource (withIO, release, newRef, readRef, writeRef)++-- | Stream the contents of a file as binary data.+sourceFile :: ResourceIO m+ => FilePath+ -> Source m S.ByteString+sourceFile fp = sourceIO+ (IO.openFile fp IO.ReadMode)+ IO.hClose+ (\handle -> do+ bs <- liftIO $ S.hGetSome handle 4096+ if S.null bs+ then return Closed+ else return $ Open bs)++-- | Stream the contents of a file as binary data, starting from a certain+-- offset and only consuming up to a certain number of bytes.+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.openFile 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.+sinkFile :: ResourceIO m+ => FilePath+ -> Sink S.ByteString m ()+sinkFile fp = sinkIO+ (IO.openFile 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@.+conduitFile :: ResourceIO m+ => FilePath+ -> Conduit S.ByteString m S.ByteString+conduitFile fp = conduitIO+ (IO.openFile 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.+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 []
+ Data/Conduit/Lazy.hs view
@@ -0,0 +1,25 @@+{-# 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++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
@@ -0,0 +1,230 @@+{-# 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+ , map+ , concatMap+ , head+ , peek+ , consume+ , sinkNull+ -- ** Monadic+ , foldM+ , mapM_+ , concatMapM+ -- Conduits+ -- ** Pure+ , isolate+ , filter+ -- ** Monadic+ , mapM+ ) 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.+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.+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.+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.+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.+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+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.+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.+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.+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_'.+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.+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.+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".+consume :: Resource m => Sink a m [a]+consume = sinkState+ id+ (\front input -> return (front . (input :), Processing))+ (\front -> return $ front [])++-- | 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+-- > ...+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.+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'.+sinkNull :: Resource m => Sink a m ()+sinkNull = Sink $ return $ SinkData+ (\_ -> return Processing)+ (return ())
+ Data/Conduit/Text.hs view
@@ -0,0 +1,295 @@+{-# 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.+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.+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.+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:)++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)++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++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)++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++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++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++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)++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
@@ -0,0 +1,46 @@+-- | 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.+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.+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.+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
@@ -0,0 +1,192 @@+{-# 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.+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.+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.+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
@@ -0,0 +1,207 @@+{-# 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 (..)+ ) 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@.+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.+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'.+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.+data BufferedSource m a = BufferedSource+ { bsourcePull :: ResourceT m (SourceResult a)+ , bsourceUnpull :: a -> ResourceT m ()+ , bsourceClose :: ResourceT m ()+ }++data SourceInvariantException = PullAfterEOF String+ deriving (Show, Typeable)+instance Exception SourceInvariantException++-- | This typeclass allows us to unify operators on 'Source' and 'BufferedSource'.+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
+ Data/Conduit/Util/Conduit.hs view
@@ -0,0 +1,189 @@+{-# 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.+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) <- 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'.+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.+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'.+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.+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'.+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
@@ -0,0 +1,101 @@+{-# 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.+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) <- 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.+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.+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
@@ -0,0 +1,100 @@+{-# 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.+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.+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.+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
@@ -0,0 +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.
+ Setup.lhs view
@@ -0,0 +1,7 @@+#!/usr/bin/env runhaskell++> module Main where+> import Distribution.Simple++> main :: IO ()+> main = defaultMain
+ conduit.cabal view
@@ -0,0 +1,59 @@+Name: conduit+Version: 0.0.0+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/yesodweb/conduit+extra-source-files: test/main.hs++flag debug++Library+ 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 && < 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
@@ -0,0 +1,311 @@+{-# 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 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+ {- FIXME+ it "sink yield" $ do+ x <- runResourceT $ CL.sourceList (map return [1..10 :: Int]) C.$$ do+ CL.drop 5+ C.yield [11..15] ()+ return ()+ CL.consume+ x @?= [11..15] ++ [6..10]+ -}++ 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 "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"++it' :: String -> IO () -> Writer [Spec] ()+it' = it