conduit 0.5.6 → 1.0.0
raw patch · 12 files changed
+407/−734 lines, 12 files
Files
- Data/Conduit.hs +120/−115
- Data/Conduit/Binary.hs +20/−20
- Data/Conduit/Internal.hs +88/−59
- Data/Conduit/Lazy.hs +16/−13
- Data/Conduit/List.hs +37/−35
- Data/Conduit/Text.hs +10/−10
- Data/Conduit/Util.hs +22/−28
- Data/Conduit/Util/Conduit.hs +0/−176
- Data/Conduit/Util/Sink.hs +0/−81
- Data/Conduit/Util/Source.hs +0/−93
- conduit.cabal +7/−13
- test/main.hs +87/−91
Data/Conduit.hs view
@@ -1,84 +1,5 @@-module Data.Conduit- ( -- * Conduit interface- -- $conduitInterface- Source- , Conduit- , Sink- -- ** Connect/fuse- -- $lifecycle- , ($$)- , ($=)- , (=$)- , (=$=)-- -- * Pipe interface- -- $pipeInterface- , Pipe- , (>+>)- , (<+<)- , runPipe- , injectLeftovers-- -- * Primitives- -- $primitives- , await- , awaitE- , awaitForever- , yield- , yieldOr- , leftover- -- ** Finalization- , bracketP- , addCleanup-- -- * Connect-and-resume- -- $connectAndResume- , ResumableSource- , ($$+)- , ($$++)- , ($$+-)- , unwrapResumable-- -- * Utility functions- , transPipe- , mapOutput- , mapOutputMaybe- , mapInput- , withUpstream-- -- * Generalized conduit types- -- $generalizedConduitTypes- , GSource- , GSink- , GLSink- , GInfSink- , GLInfSink- , GConduit- , GLConduit- , GInfConduit- , GLInfConduit-- -- * Flushing- , Flush (..)-- -- * Convenience re-exports- , ResourceT- , MonadResource- , MonadThrow (..)- , MonadUnsafeIO (..)- , runResourceT- , ExceptionT (..)- , runExceptionT_- , runException- , runException_- , MonadBaseControl- ) where--import Control.Monad.Trans.Resource-import Data.Conduit.Internal-import Data.Void (Void)--{- $conduitInterface+{-# LANGUAGE RankNTypes #-}+{- | Let's start off with a few simple examples of @conduit@ usage. First, a file copy utility:@@ -109,7 +30,7 @@ @sourceList@ is a convenience function for turning a list into a @Source@. @fold@ implements a strict left fold for consuming the input stream. -The next major aspect to the @conduit@ library: the @Conduit@ type.+The next major aspect to the @conduit@ library is the @Conduit@ type. This type represents a stream /transformer/. In order to use a @Conduit@, we must /fuse/ it with either a @Source@ or @Sink@. For example: @@ -157,7 +78,71 @@ is resource finalization, which will also be covered below. -}+module Data.Conduit+ ( -- * Conduit interface+ Source+ , Conduit+ , Sink+ -- ** Connect/fuse+ -- $lifecycle+ , ($$)+ , ($=)+ , (=$)+ , (=$=) + -- * Primitives+ -- $primitives+ , await+ , awaitForever+ , yield+ , yieldOr+ , leftover+ -- ** Finalization+ , bracketP+ , addCleanup++ -- * Connect-and-resume+ -- $connectAndResume+ , ResumableSource+ , ($$+)+ , ($$++)+ , ($$+-)+ , unwrapResumable++ -- * Utility functions+ , transPipe+ , mapOutput+ , mapOutputMaybe+ , mapInput++ -- * Generalized conduit types+ -- $generalizedConduitTypes+ , Producer+ , Consumer+ , toProducer+ , toConsumer++ -- * Flushing+ , Flush (..)++ -- * Convenience re-exports+ , ResourceT+ , MonadResource+ , MonadThrow (..)+ , MonadUnsafeIO (..)+ , runResourceT+ , ExceptionT (..)+ , runExceptionT_+ , runException+ , runException_+ , MonadBaseControl+ ) where++import Control.Monad.Trans.Resource+import Data.Conduit.Internal hiding (await, awaitForever, yield, yieldOr, leftover, bracketP, addCleanup, transPipe, mapOutput, mapOutputMaybe, mapInput)+import qualified Data.Conduit.Internal as CI++ {- $lifecycle It is important to understand the lifecycle of our components. Notice that we@@ -433,8 +418,6 @@ infixl 1 $= infixr 2 =$ infixr 2 =$=-infixr 9 <+<-infixl 9 >+> infixr 0 $$+ infixr 0 $$++ infixr 0 $$+-@@ -461,7 +444,7 @@ -- -- Since 0.4.0 ($=) :: Monad m => Source m a -> Conduit a m b -> Source m b-($=) = pipeL+ConduitM src $= ConduitM con = ConduitM $ pipeL src con {-# INLINE ($=) #-} -- | Right fuse, combining a conduit and a sink together into a new sink.@@ -473,7 +456,7 @@ -- -- Since 0.4.0 (=$) :: Monad m => Conduit a m b -> Sink b m c -> Sink a m c-(=$) = pipeL+ConduitM con =$ ConduitM sink = ConduitM $ pipeL con sink {-# INLINE (=$) #-} -- | Fusion operator, combining two @Conduit@s together into a new @Conduit@.@@ -483,37 +466,10 @@ -- Leftover data returned from the right @Conduit@ will be discarded. -- -- Since 0.4.0-(=$=) :: Monad m => Conduit a m b -> Conduit b m c -> Conduit a m c-(=$=) = pipeL+(=$=) :: Monad m => Conduit a m b -> ConduitM b c m r -> ConduitM a c m r+ConduitM left =$= ConduitM right = ConduitM $ pipeL left right {-# INLINE (=$=) #-} ---- | Fuse together two @Pipe@s, connecting the output from the left to the--- input of the right.------ Notice that the /leftover/ parameter for the @Pipe@s must be @Void@. This--- ensures that there is no accidental data loss of leftovers during fusion. If--- you have a @Pipe@ with leftovers, you must first call 'injectLeftovers'. For--- example:------ >>> :load Data.Conduit.List--- >>> :set -XNoMonomorphismRestriction--- >>> let pipe = peek >>= \x -> fold (Prelude.+) 0 >>= \y -> return (x, y)--- >>> runPipe $ sourceList [1..10] >+> injectLeftovers pipe--- (Just 1,55)------ Since 0.5.0-(>+>) :: Monad m => Pipe l a b r0 m r1 -> Pipe Void b c r1 m r2 -> Pipe l a c r0 m r2-(>+>) = pipe-{-# INLINE (>+>) #-}---- | Same as '>+>', but reverse the order of the arguments.------ Since 0.5.0-(<+<) :: Monad m => Pipe Void b c r1 m r2 -> Pipe l a b r0 m r1 -> Pipe l a c r0 m r2-(<+<) = flip pipe-{-# INLINE (<+<) #-}- -- | The connect-and-resume operator. This does not close the @Source@, but -- instead returns it to be used again. This allows a @Source@ to be used -- incrementally in a large program, without forcing the entire program to live@@ -557,3 +513,52 @@ instance Functor Flush where fmap _ Flush = Flush fmap f (Chunk a) = Chunk (f a)++await :: Monad m => Consumer i m (Maybe i)+await = ConduitM CI.await++mapOutput :: Monad m => (o1 -> o2) -> ConduitM i o1 m r -> ConduitM i o2 m r+mapOutput f (ConduitM p) = ConduitM $ CI.mapOutput f p++mapOutputMaybe :: Monad m => (o1 -> Maybe o2) -> ConduitM i o1 m r -> ConduitM i o2 m r+mapOutputMaybe f (ConduitM p) = ConduitM $ CI.mapOutputMaybe f p++mapInput :: Monad m+ => (i1 -> i2) -- ^ map initial input to new input+ -> (i2 -> Maybe i1) -- ^ map new leftovers to initial leftovers+ -> ConduitM i2 o m r+ -> ConduitM i1 o m r+mapInput f g (ConduitM p) = ConduitM $ CI.mapInput f g p++transPipe :: Monad m => (forall a. m a -> n a) -> ConduitM i o m r -> ConduitM i o n r+transPipe f = ConduitM . CI.transPipe f . unConduitM++addCleanup :: Monad m+ => (Bool -> m ()) -- ^ @True@ if @Pipe@ ran to completion, @False@ for early termination.+ -> ConduitM i o m r+ -> ConduitM i o m r+addCleanup f = ConduitM . CI.addCleanup f . unConduitM++bracketP :: MonadResource m+ => IO a+ -> (a -> IO ())+ -> (a -> ConduitM i o m r)+ -> ConduitM i o m r+bracketP alloc free inside = ConduitM $ CI.bracketP alloc free $ unConduitM . inside++leftover :: i -> ConduitM i o m ()+leftover = ConduitM . CI.leftover++yield :: Monad m+ => o -- ^ output value+ -> ConduitM i o m ()+yield = ConduitM . CI.yield++yieldOr :: Monad m+ => o+ -> m () -- ^ finalizer+ -> ConduitM i o m ()+yieldOr o m = ConduitM $ CI.yieldOr o m++awaitForever :: Monad m => (i -> ConduitM i o m r) -> ConduitM i o m ()+awaitForever f = ConduitM $ CI.awaitForever (unConduitM . f)
Data/Conduit/Binary.hs view
@@ -35,7 +35,7 @@ import Prelude hiding (head, take, drop, takeWhile, dropWhile) import qualified Data.ByteString as S import qualified Data.ByteString.Lazy as L-import Data.Conduit hiding (Source, Conduit, Sink, Pipe)+import Data.Conduit import Data.Conduit.List (sourceList) import Control.Exception (assert) import Control.Monad (unless)@@ -54,7 +54,7 @@ -- Since 0.3.0 sourceFile :: MonadResource m => FilePath- -> GSource m S.ByteString+ -> Producer m S.ByteString sourceFile fp = #if CABAL_OS_WINDOWS || NO_HANDLES bracketP@@ -74,7 +74,7 @@ -- Since 0.3.0 sourceHandle :: MonadIO m => IO.Handle- -> GSource m S.ByteString+ -> Producer m S.ByteString sourceHandle h = loop where@@ -92,7 +92,7 @@ -- Since 0.3.0 sourceIOHandle :: MonadResource m => IO IO.Handle- -> GSource m S.ByteString+ -> Producer m S.ByteString sourceIOHandle alloc = bracketP alloc IO.hClose sourceHandle -- | Stream all incoming data to the given 'IO.Handle'. Note that this function@@ -101,7 +101,7 @@ -- Since 0.3.0 sinkHandle :: MonadIO m => IO.Handle- -> GInfSink S.ByteString m+ -> Consumer S.ByteString m () sinkHandle h = awaitForever $ liftIO . S.hPut h -- | An alternative to 'sinkHandle'.@@ -112,7 +112,7 @@ -- Since 0.3.0 sinkIOHandle :: MonadResource m => IO IO.Handle- -> GInfSink S.ByteString m+ -> Consumer S.ByteString m () sinkIOHandle alloc = bracketP alloc IO.hClose sinkHandle -- | Stream the contents of a file as binary data, starting from a certain@@ -123,7 +123,7 @@ => FilePath -> Maybe Integer -- ^ Offset -> Maybe Integer -- ^ Maximum count- -> GSource m S.ByteString+ -> Producer m S.ByteString sourceFileRange fp offset count = bracketP (IO.openBinaryFile fp IO.ReadMode) IO.hClose@@ -160,7 +160,7 @@ -- Since 0.3.0 sinkFile :: MonadResource m => FilePath- -> GInfSink S.ByteString m+ -> Consumer S.ByteString m () sinkFile fp = sinkIOHandle (IO.openBinaryFile fp IO.WriteMode) -- | Stream the contents of the input to a file, and also send it along the@@ -169,7 +169,7 @@ -- Since 0.3.0 conduitFile :: MonadResource m => FilePath- -> GInfConduit S.ByteString m S.ByteString+ -> Conduit S.ByteString m S.ByteString conduitFile fp = bracketP (IO.openBinaryFile fp IO.WriteMode) IO.hClose@@ -184,7 +184,7 @@ -- Since 0.3.0 isolate :: Monad m => Int- -> GLConduit S.ByteString m S.ByteString+ -> Conduit S.ByteString m S.ByteString isolate = loop where@@ -204,7 +204,7 @@ -- | Return the next byte from the stream, if available. -- -- Since 0.3.0-head :: Monad m => GLSink S.ByteString m (Maybe Word8)+head :: Monad m => Consumer S.ByteString m (Maybe Word8) head = do mbs <- await case mbs of@@ -217,7 +217,7 @@ -- | Return all bytes while the predicate returns @True@. -- -- Since 0.3.0-takeWhile :: Monad m => (Word8 -> Bool) -> GLConduit S.ByteString m S.ByteString+takeWhile :: Monad m => (Word8 -> Bool) -> Conduit S.ByteString m S.ByteString takeWhile p = loop where@@ -233,7 +233,7 @@ -- | Ignore all bytes while the predicate returns @True@. -- -- Since 0.3.0-dropWhile :: Monad m => (Word8 -> Bool) -> GLSink S.ByteString m ()+dropWhile :: Monad m => (Word8 -> Bool) -> Consumer S.ByteString m () dropWhile p = loop where@@ -248,7 +248,7 @@ -- | Take the given number of bytes, if available. -- -- Since 0.3.0-take :: Monad m => Int -> GLSink S.ByteString m L.ByteString+take :: Monad m => Int -> Consumer S.ByteString m L.ByteString take n0 = go n0 id where@@ -266,7 +266,7 @@ -- | Drop up to the given number of bytes. -- -- Since 0.5.0-drop :: Monad m => Int -> GLSink S.ByteString m ()+drop :: Monad m => Int -> Consumer S.ByteString m () drop = go where@@ -285,15 +285,15 @@ -- (10), and strip it from the output. -- -- Since 0.3.0-lines :: Monad m => GInfConduit S.ByteString m S.ByteString+lines :: Monad m => Conduit S.ByteString m S.ByteString lines = loop id where- loop front = awaitE >>= either (finish front) (go front)+ loop front = await >>= maybe (finish front) (go front) - finish front r =+ finish front = let final = front S.empty- in unless (S.null final) (yield final) >> return r+ in unless (S.null final) (yield final) go sofar more = case S.uncons second of@@ -307,5 +307,5 @@ -- | Stream the chunks from a lazy bytestring. -- -- Since 0.5.0-sourceLbs :: Monad m => L.ByteString -> GSource m S.ByteString+sourceLbs :: Monad m => L.ByteString -> Producer m S.ByteString sourceLbs = sourceList . L.toChunks
Data/Conduit/Internal.hs view
@@ -4,21 +4,16 @@ {-# LANGUAGE UndecidableInstances #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE TupleSections #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} module Data.Conduit.Internal ( -- * Types Pipe (..)+ , ConduitM (..) , Source- , GSource+ , Producer , Sink- , GSink- , GLSink- , GInfSink- , GLInfSink+ , Consumer , Conduit- , GConduit- , GLConduit- , GInfConduit- , GLInfConduit , ResumableSource (..) -- * Primitives , await@@ -37,10 +32,14 @@ , connectResume , runPipe , injectLeftovers+ , (>+>)+ , (<+<) -- * Generalizing , sourceToPipe , sinkToPipe , conduitToPipe+ , toProducer+ , toConsumer -- * Utilities , transPipe , mapOutput@@ -140,68 +139,41 @@ instance MonadResource m => MonadResource (Pipe l i o u m) where liftResourceT = lift . liftResourceT +newtype ConduitM i o m r = ConduitM { unConduitM :: Pipe i i o () m r }+ deriving (Functor, Applicative, Monad, MonadIO, MonadTrans, MonadThrow, MonadActive, MonadResource)+instance MonadBase base m => MonadBase base (ConduitM i o m) where+ liftBase = lift . liftBase+instance Monad m => Monoid (ConduitM i o m ()) where+ mempty = return ()+ mappend = (>>)+ -- | Provides a stream of output values, without consuming any input or -- producing a final result. -- -- Since 0.5.0-type Source m o = Pipe () () o () m ()+type Source m o = ConduitM () o m () -- | Generalized 'Source'. -- -- Since 0.5.0-type GSource m o = forall l i u. Pipe l i o u m ()+type Producer m o = forall i. ConduitM i o m () -- | Consumes a stream of input values and produces a final result, without -- producing any output. -- -- Since 0.5.0-type Sink i m r = Pipe i i Void () m r+type Sink i m r = ConduitM i Void m r -- | Generalized 'Sink' without leftovers. -- -- Since 0.5.0-type GSink i m r = forall l o u. Pipe l i o u m r---- | Generalized 'Sink' with leftovers.------ Since 0.5.0-type GLSink i m r = forall o u. Pipe i i o u m r---- | Generalized 'Sink' without leftovers returning upstream result.------ Since 0.5.0-type GInfSink i m = forall l o r. Pipe l i o r m r---- | Generalized 'Sink' with leftovers returning upstream result.------ Since 0.5.0-type GLInfSink i m = forall o r. Pipe i i o r m r+type Consumer i m r = forall o. ConduitM i o m r -- | Consumes a stream of input values and produces a stream of output values, -- without producing a final result. -- -- Since 0.5.0-type Conduit i m o = Pipe i i o () m ()---- | Generalized conduit without leftovers.------ Since 0.5.0-type GConduit i m o = forall l u. Pipe l i o u m ()---- | Generalized conduit with leftovers.------ Since 0.5.0-type GLConduit i m o = forall u. Pipe i i o u m ()---- | Generalized conduit without leftovers returning upstream result.------ Since 0.5.0-type GInfConduit i m o = forall l r. Pipe l i o r m r---- | Generalized conduit with leftovers returning upstream result.------ Since 0.5.0-type GLInfConduit i m o = forall r. Pipe i i o r m r+type Conduit i m o = ConduitM i o m () -- | A @Source@ which has been started, but has not yet completed. --@@ -390,12 +362,12 @@ => ResumableSource m o -> Sink o m r -> m (ResumableSource m o, r)-connectResume (ResumableSource left0 leftFinal0) =- go leftFinal0 left0+connectResume (ResumableSource (ConduitM left0) leftFinal0) =+ go leftFinal0 left0 . unConduitM where go leftFinal left right = case right of- Done r2 -> return (ResumableSource left leftFinal, r2)+ Done r2 -> return (ResumableSource (ConduitM left) leftFinal, r2) PipeM mp -> mp >>= go leftFinal left HaveOutput _ _ o -> absurd o Leftover p i -> go leftFinal (HaveOutput left leftFinal i) p@@ -524,15 +496,18 @@ #-} sourceToPipe :: Monad m => Source m o -> Pipe l i o u m ()-sourceToPipe (Done ()) = Done ()-sourceToPipe (PipeM mp) = PipeM (liftM sourceToPipe mp)-sourceToPipe (NeedInput _ c) = sourceToPipe $ c ()-sourceToPipe (HaveOutput p c o) = HaveOutput (sourceToPipe p) c o-sourceToPipe (Leftover p ()) = sourceToPipe p+sourceToPipe =+ go . unConduitM+ where+ go (Done ()) = Done ()+ go (PipeM mp) = PipeM (liftM go mp)+ go (NeedInput _ c) = go $ c ()+ go (HaveOutput p c o) = HaveOutput (go p) c o+ go (Leftover p ()) = go p sinkToPipe :: Monad m => Sink i m r -> Pipe l i o u m r sinkToPipe =- go . injectLeftovers+ go . injectLeftovers . unConduitM where go (Done r) = Done r go (PipeM mp) = PipeM (liftM go mp)@@ -542,7 +517,7 @@ conduitToPipe :: Monad m => Conduit i m o -> Pipe l i o u m () conduitToPipe =- go . injectLeftovers+ go . injectLeftovers . unConduitM where go (Done ()) = Done () go (PipeM mp) = PipeM (liftM go mp)@@ -587,3 +562,57 @@ x <- liftIO $ I.readIORef ref when x final return (liftIO (I.writeIORef ref False) >> src, final')+infixr 9 <+<+infixl 9 >+>++-- | Fuse together two @Pipe@s, connecting the output from the left to the+-- input of the right.+--+-- Notice that the /leftover/ parameter for the @Pipe@s must be @Void@. This+-- ensures that there is no accidental data loss of leftovers during fusion. If+-- you have a @Pipe@ with leftovers, you must first call 'injectLeftovers'. For+-- example:+--+-- > :load Data.Conduit.List+-- > :set -XNoMonomorphismRestriction+-- > let pipe = peek >>= \x -> fold (Prelude.+) 0 >>= \y -> return (x, y)+-- > runPipe $ sourceList [1..10] >+> injectLeftovers pipe+-- (Just 1,55)+--+-- Since 0.5.0+(>+>) :: Monad m => Pipe l a b r0 m r1 -> Pipe Void b c r1 m r2 -> Pipe l a c r0 m r2+(>+>) = pipe+{-# INLINE (>+>) #-}++-- | Same as '>+>', but reverse the order of the arguments.+--+-- Since 0.5.0+(<+<) :: Monad m => Pipe Void b c r1 m r2 -> Pipe l a b r0 m r1 -> Pipe l a c r0 m r2+(<+<) = flip pipe+{-# INLINE (<+<) #-}++-- | Convert a 'Source' into a 'Producer'.+--+-- Since 1.0.0+toProducer :: Monad m => Source m a -> Producer m a+toProducer =+ ConduitM . go . unConduitM+ where+ go (HaveOutput p c o) = HaveOutput (go p) c o+ go (NeedInput _ c) = go (c ())+ go (Done r) = Done r+ go (PipeM mp) = PipeM (liftM go mp)+ go (Leftover p ()) = go p++-- | Convert a 'Sink' into a 'Consumer'.+--+-- Since 1.0.0+toConsumer :: Monad m => Sink a m b -> Consumer a m b+toConsumer =+ ConduitM . go . unConduitM+ where+ go (HaveOutput _ _ o) = absurd o+ go (NeedInput p c) = NeedInput (go . p) (go . c)+ go (Done r) = Done r+ go (PipeM mp) = PipeM (liftM go mp)+ go (Leftover p l) = Leftover (go p) l
Data/Conduit/Lazy.hs view
@@ -8,7 +8,7 @@ ) where import Data.Conduit-import Data.Conduit.Internal (Pipe (..))+import Data.Conduit.Internal (Pipe (..), unConduitM) import System.IO.Unsafe (unsafeInterleaveIO) import Control.Monad.Trans.Control (liftBaseOp_) import Control.Monad.Trans.Resource (MonadActive (monadActive))@@ -19,15 +19,18 @@ -- state has been closed. -- -- Since 0.3.0-lazyConsume :: (MonadBaseControl IO m, MonadActive m) => Pipe l i a () m r -> m [a]-lazyConsume (Done _) = return []-lazyConsume (HaveOutput src _ x) = do- xs <- lazyConsume src- return $ x : xs-lazyConsume (PipeM msrc) = liftBaseOp_ unsafeInterleaveIO $ do- a <- monadActive- if a- then msrc >>= lazyConsume- else return []-lazyConsume (NeedInput _ c) = lazyConsume (c ())-lazyConsume (Leftover p _) = lazyConsume p+lazyConsume :: (MonadBaseControl IO m, MonadActive m) => Source m a -> m [a]+lazyConsume =+ go . unConduitM+ where+ go (Done _) = return []+ go (HaveOutput src _ x) = do+ xs <- go src+ return $ x : xs+ go (PipeM msrc) = liftBaseOp_ unsafeInterleaveIO $ do+ a <- monadActive+ if a+ then msrc >>= go+ else return []+ go (NeedInput _ c) = go (c ())+ go (Leftover p _) = go p
Data/Conduit/List.hs view
@@ -62,8 +62,7 @@ , either ) import Data.Monoid (Monoid, mempty, mappend)-import Data.Conduit hiding (Source, Sink, Conduit, Pipe)-import Data.Conduit.Internal (sourceList, pipeL, pipe)+import Data.Conduit import Control.Monad (when, (<=<)) import Control.Monad.Trans.Class (lift) @@ -73,7 +72,7 @@ unfold :: Monad m => (b -> Maybe (a, b)) -> b- -> GSource m a+ -> Producer m a unfold f = go where@@ -82,6 +81,9 @@ Just (a, seed') -> yield a >> go seed' Nothing -> return () +sourceList :: Monad m => [a] -> Producer m a+sourceList = Prelude.mapM_ yield+ -- | Enumerate from a value to a final value, inclusive, via 'succ'. -- -- This is generally more efficient than using @Prelude@\'s @enumFromTo@ and@@ -92,7 +94,7 @@ enumFromTo :: (Enum a, Eq a, Monad m) => a -> a- -> GSource m a+ -> Producer m a enumFromTo start stop = go start where@@ -101,7 +103,7 @@ | otherwise = yield i >> go (succ i) -- | Produces an infinite stream of repeated applications of f to x.-iterate :: Monad m => (a -> a) -> a -> GSource m a+iterate :: Monad m => (a -> a) -> a -> Producer m a iterate f = go where@@ -113,7 +115,7 @@ fold :: Monad m => (b -> a -> b) -> b- -> GSink a m b+ -> Consumer a m b fold f = loop where@@ -130,7 +132,7 @@ foldM :: Monad m => (b -> a -> m b) -> b- -> GSink a m b+ -> Consumer a m b foldM f = loop where@@ -146,7 +148,7 @@ -- Since 0.5.3 foldMap :: (Monad m, Monoid b) => (a -> b)- -> GSink a m b+ -> Consumer a m b foldMap f = fold combiner mempty where@@ -157,7 +159,7 @@ -- Since 0.3.0 mapM_ :: Monad m => (a -> m ())- -> GInfSink a m+ -> Consumer a m () mapM_ f = awaitForever $ lift . f -- | Ignore a certain number of values in the stream. This function is@@ -171,7 +173,7 @@ -- Since 0.3.0 drop :: Monad m => Int- -> GSink a m ()+ -> Consumer a m () drop = loop where@@ -187,7 +189,7 @@ -- Since 0.3.0 take :: Monad m => Int- -> GSink a m [a]+ -> Consumer a m [a] take = loop id where@@ -199,20 +201,20 @@ -- | Take a single value from the stream, if available. -- -- Since 0.3.0-head :: Monad m => GSink a m (Maybe a)+head :: Monad m => Consumer a m (Maybe a) head = await -- | Look at the next value in the stream, if available. This function will not -- change the state of the stream. -- -- Since 0.3.0-peek :: Monad m => GLSink a m (Maybe a)+peek :: Monad m => Consumer a m (Maybe a) peek = await >>= maybe (return Nothing) (\x -> leftover x >> return (Just x)) -- | Apply a transformation to all values in a stream. -- -- Since 0.3.0-map :: Monad m => (a -> b) -> GInfConduit a m b+map :: Monad m => (a -> b) -> Conduit a m b map f = awaitForever $ yield . f {-@@ -243,7 +245,7 @@ -- side-effects of running the action, see 'mapM_'. -- -- Since 0.3.0-mapM :: Monad m => (a -> m b) -> GInfConduit a m b+mapM :: Monad m => (a -> m b) -> Conduit a m b mapM f = awaitForever $ yield <=< lift . f -- | Apply a monadic action on all values in a stream.@@ -254,52 +256,52 @@ -- > iterM f = mapM (\a -> f a >>= \() -> return a) -- -- Since 0.5.6-iterM :: Monad m => (a -> m ()) -> GInfConduit a m a+iterM :: Monad m => (a -> m ()) -> Conduit a m a iterM f = awaitForever $ \a -> lift (f a) >> yield a -- | Apply a transformation that may fail to all values in a stream, discarding -- the failures. -- -- Since 0.5.1-mapMaybe :: Monad m => (a -> Maybe b) -> GInfConduit a m b+mapMaybe :: Monad m => (a -> Maybe b) -> Conduit a m b mapMaybe f = awaitForever $ maybe (return ()) yield . f -- | Apply a monadic transformation that may fail to all values in a stream, -- discarding the failures. -- -- Since 0.5.1-mapMaybeM :: Monad m => (a -> m (Maybe b)) -> GInfConduit a m b+mapMaybeM :: Monad m => (a -> m (Maybe b)) -> Conduit a m b mapMaybeM f = awaitForever $ maybe (return ()) yield <=< lift . f -- | Filter the @Just@ values from a stream, discarding the @Nothing@ values. -- -- Since 0.5.1-catMaybes :: Monad m => GInfConduit (Maybe a) m a+catMaybes :: Monad m => Conduit (Maybe a) m a catMaybes = awaitForever $ maybe (return ()) yield -- | Apply a transformation to all values in a stream, concatenating the output -- values. -- -- Since 0.3.0-concatMap :: Monad m => (a -> [b]) -> GInfConduit a m b+concatMap :: Monad m => (a -> [b]) -> Conduit a m b concatMap f = awaitForever $ sourceList . f -- | Apply a monadic transformation to all values in a stream, concatenating -- the output values. -- -- Since 0.3.0-concatMapM :: Monad m => (a -> m [b]) -> GInfConduit a m b+concatMapM :: Monad m => (a -> m [b]) -> Conduit a m b concatMapM f = awaitForever $ sourceList <=< lift . f -- | 'concatMap' with an accumulator. -- -- Since 0.3.0-concatMapAccum :: Monad m => (a -> accum -> (accum, [b])) -> accum -> GInfConduit a m b+concatMapAccum :: Monad m => (a -> accum -> (accum, [b])) -> accum -> Conduit a m b concatMapAccum f = loop where loop accum =- awaitE >>= either return go+ await >>= maybe (return ()) go where go a = do let (accum', bs) = f a accum@@ -309,12 +311,12 @@ -- | 'concatMapM' with an accumulator. -- -- Since 0.3.0-concatMapAccumM :: Monad m => (a -> accum -> m (accum, [b])) -> accum -> GInfConduit a m b+concatMapAccumM :: Monad m => (a -> accum -> m (accum, [b])) -> accum -> Conduit a m b concatMapAccumM f = loop where loop accum = do- awaitE >>= either return go+ await >>= maybe (return ()) go where go a = do (accum', bs) <- lift $ f a accum@@ -326,7 +328,7 @@ -- "Data.Conduit.Lazy". -- -- Since 0.3.0-consume :: Monad m => GSink a m [a]+consume :: Monad m => Consumer a m [a] consume = loop id where@@ -335,14 +337,14 @@ -- | Grouping input according to an equality function. -- -- Since 0.3.0-groupBy :: Monad m => (a -> a -> Bool) -> GInfConduit a m [a]+groupBy :: Monad m => (a -> a -> Bool) -> Conduit a m [a] groupBy f = start where- start = awaitE >>= either return (loop id)+ start = await >>= maybe (return ()) (loop id) loop rest x =- awaitE >>= either (\r -> yield (x : rest []) >> return r) go+ await >>= maybe (yield (x : rest [])) go where go y | f x y = loop (rest . (y:)) x@@ -361,7 +363,7 @@ -- > ... -- -- Since 0.3.0-isolate :: Monad m => Int -> GConduit a m a+isolate :: Monad m => Int -> Conduit a m a isolate = loop where@@ -371,21 +373,21 @@ -- | Keep only values in the stream passing a given predicate. -- -- Since 0.3.0-filter :: Monad m => (a -> Bool) -> GInfConduit a m a+filter :: Monad m => (a -> Bool) -> Conduit a m a filter f = awaitForever $ \i -> when (f i) (yield i) -- | Ignore the remainder of values in the source. Particularly useful when -- combined with 'isolate'. -- -- Since 0.3.0-sinkNull :: Monad m => GInfSink a m+sinkNull :: Monad m => Consumer a m () sinkNull = awaitForever $ \_ -> return () -- | A source that outputs no values. Note that this is just a type-restricted -- synonym for 'mempty'. -- -- Since 0.3.0-sourceNull :: Monad m => GSource m a+sourceNull :: Monad m => Producer m a sourceNull = return () -- | Run a @Pipe@ repeatedly, and output its result value downstream. Stops@@ -393,8 +395,8 @@ -- -- Since 0.5.0 sequence :: Monad m- => GLSink i m o -- ^ @Pipe@ to run repeatedly- -> GLInfConduit i m o+ => Consumer i m o -- ^ @Pipe@ to run repeatedly+ -> Conduit i m o sequence sink = self where
Data/Conduit/Text.hs view
@@ -40,7 +40,7 @@ import System.IO.Unsafe (unsafePerformIO) import Data.Typeable (Typeable) -import Data.Conduit hiding (Source, Conduit, Sink, Pipe)+import Data.Conduit import qualified Data.Conduit.List as CL import Control.Monad.Trans.Class (lift) import Control.Monad (unless)@@ -67,15 +67,15 @@ -- | Emit each line separately -- -- Since 0.4.1-lines :: Monad m => GInfConduit T.Text m T.Text+lines :: Monad m => Conduit T.Text m T.Text lines = loop id where- loop front = awaitE >>= either (finish front) (go front)+ loop front = await >>= maybe (finish front) (go front) - finish front r =+ finish front = let final = front T.empty- in unless (T.null final) (yield final) >> return r+ in unless (T.null final) (yield final) go sofar more = case T.uncons second of@@ -90,7 +90,7 @@ -- not capable of representing an input character, an exception will be thrown. -- -- Since 0.3.0-encode :: MonadThrow m => Codec -> GInfConduit T.Text m B.ByteString+encode :: MonadThrow m => Codec -> Conduit T.Text m B.ByteString encode codec = CL.mapM $ \t -> do let (bs, mexc) = codecEncode codec t maybe (return bs) (monadThrow . fst) mexc@@ -100,15 +100,15 @@ -- not capable of decoding an input byte sequence, an exception will be thrown. -- -- Since 0.3.0-decode :: MonadThrow m => Codec -> GInfConduit B.ByteString m T.Text+decode :: MonadThrow m => Codec -> Conduit B.ByteString m T.Text decode codec = loop id where- loop front = awaitE >>= either (finish front) (go front)+ loop front = await >>= maybe (finish front) (go front) - finish front r =+ finish front = case B.uncons $ front B.empty of- Nothing -> return r+ Nothing -> return () Just (w, _) -> lift $ monadThrow $ DecodeException codec w go front bs' =
Data/Conduit/Util.hs view
@@ -2,43 +2,37 @@ -- considered deprecated, as there are now easier ways to handle their use -- cases. This module is provided solely for backwards compatibility. module Data.Conduit.Util- ( -- * Source- module Data.Conduit.Util.Source- -- * Sink- , module Data.Conduit.Util.Sink- -- * Conduit- , module Data.Conduit.Util.Conduit- -- * Misc- , zip+ ( -- * Misc+ zip , zipSinks ) where import Prelude hiding (zip) import Control.Monad (liftM, liftM2)-import Data.Conduit.Internal (Pipe (..), Source, Sink, injectLeftovers)+import Data.Conduit.Internal (Pipe (..), Source, Sink, injectLeftovers, ConduitM (..)) import Data.Void (Void, absurd)-import Data.Conduit.Util.Source-import Data.Conduit.Util.Sink-import Data.Conduit.Util.Conduit -- | Combines two sources. The new source will stop producing once either -- source has been exhausted. -- -- Since 0.3.0 zip :: Monad m => Source m a -> Source m b -> Source m (a, b)-zip (Leftover left ()) right = zip left right-zip left (Leftover right ()) = zip left right-zip (Done ()) (Done ()) = Done ()-zip (Done ()) (HaveOutput _ close _) = PipeM (close >> return (Done ()))-zip (HaveOutput _ close _) (Done ()) = PipeM (close >> return (Done ()))-zip (Done ()) (PipeM _) = Done ()-zip (PipeM _) (Done ()) = Done ()-zip (PipeM mx) (PipeM my) = PipeM (liftM2 zip mx my)-zip (PipeM mx) y@HaveOutput{} = PipeM (liftM (\x -> zip x y) mx)-zip x@HaveOutput{} (PipeM my) = PipeM (liftM (zip x) my)-zip (HaveOutput srcx closex x) (HaveOutput srcy closey y) = HaveOutput (zip srcx srcy) (closex >> closey) (x, y)-zip (NeedInput _ c) right = zip (c ()) right-zip left (NeedInput _ c) = zip left (c ())+zip (ConduitM left0) (ConduitM right0) =+ ConduitM $ go left0 right0+ where+ go (Leftover left ()) right = go left right+ go left (Leftover right ()) = go left right+ go (Done ()) (Done ()) = Done ()+ go (Done ()) (HaveOutput _ close _) = PipeM (close >> return (Done ()))+ go (HaveOutput _ close _) (Done ()) = PipeM (close >> return (Done ()))+ go (Done ()) (PipeM _) = Done ()+ go (PipeM _) (Done ()) = Done ()+ go (PipeM mx) (PipeM my) = PipeM (liftM2 go mx my)+ go (PipeM mx) y@HaveOutput{} = PipeM (liftM (\x -> go x y) mx)+ go x@HaveOutput{} (PipeM my) = PipeM (liftM (go x) my)+ go (HaveOutput srcx closex x) (HaveOutput srcy closey y) = HaveOutput (go srcx srcy) (closex >> closey) (x, y)+ go (NeedInput _ c) right = go (c ()) right+ go left (NeedInput _ c) = go left (c ()) -- | Combines two sinks. The new sink will complete when both input sinks have -- completed.@@ -47,10 +41,10 @@ -- -- Since 0.4.1 zipSinks :: Monad m => Sink i m r -> Sink i m r' -> Sink i m (r, r')-zipSinks x0 y0 =- injectLeftovers x0 >< injectLeftovers y0+zipSinks (ConduitM x0) (ConduitM y0) =+ ConduitM $ injectLeftovers x0 >< injectLeftovers y0 where- (><) :: Monad m => Pipe Void i Void () m r1 -> Pipe Void i Void () m r2 -> Sink i m (r1, r2)+ (><) :: Monad m => Pipe Void i Void () m r1 -> Pipe Void i Void () m r2 -> Pipe l i o () m (r1, r2) Leftover _ i >< _ = absurd i _ >< Leftover _ i = absurd i
− Data/Conduit/Util/Conduit.hs
@@ -1,176 +0,0 @@--- | Utilities for constructing and covnerting conduits. Please see--- "Data.Conduit.Types.Conduit" for more information on the base types.-module Data.Conduit.Util.Conduit- ( haveMore- , conduitState- , ConduitStateResult (..)- , conduitIO- , ConduitIOResult (..)- -- *** Sequencing- , SequencedSink- , sequenceSink- , SequencedSinkResponse (..)- ) where--import Prelude hiding (sequence)-import Control.Monad.Trans.Resource-import Data.Conduit.Internal hiding (leftover)-import Control.Monad (liftM)---- | A helper function for returning a list of values from a @Conduit@.------ Since 0.3.0-haveMore :: Conduit a m b -- ^ The next @Conduit@ to return after the list has been exhausted.- -> m () -- ^ A close action for early termination.- -> [b] -- ^ The values to send down the stream.- -> Conduit a m b-haveMore res _ [] = res-haveMore res close (x:xs) = HaveOutput (haveMore res close xs) close x---- | A helper type for @conduitState@, indicating the result of being pushed--- to. It can either indicate that processing is done, or to continue with the--- updated state.------ Since 0.3.0-data ConduitStateResult state input output =- StateFinished (Maybe input) [output]- | StateProducing state [output]--instance Functor (ConduitStateResult state input) where- fmap f (StateFinished a b) = StateFinished a (map f b)- fmap f (StateProducing a b) = StateProducing a (map f b)---- | Construct a 'Conduit' with some stateful functions. This function addresses--- threading the state value for you.------ Since 0.3.0-conduitState- :: Monad m- => state -- ^ initial state- -> (state -> input -> m (ConduitStateResult state input output)) -- ^ Push function.- -> (state -> m [output]) -- ^ Close function. The state need not be returned, since it will not be used again.- -> Conduit input m output-conduitState state0 push0 close0 =- NeedInput (push state0) (\() -> close state0)- where- push state input = PipeM (liftM goRes' $ state `seq` push0 state input)-- close state = PipeM (do- os <- close0 state- return $ sourceList os)-- goRes' (StateFinished leftover output) = maybe id pipePush leftover $ haveMore- (Done ())- (return ())- output- goRes' (StateProducing state output) = haveMore- (NeedInput (push state) (\() -> close state))- (return ())- output---- | A helper type for @conduitIO@, indicating the result of being pushed to.--- It can either indicate that processing is done, or to continue.------ Since 0.3.0-data ConduitIOResult input output =- IOFinished (Maybe input) [output]- | IOProducing [output]--instance Functor (ConduitIOResult input) where- fmap f (IOFinished a b) = IOFinished a (map f b)- fmap f (IOProducing b) = IOProducing (map f b)---- | Construct a 'Conduit'.------ Since 0.3.0-conduitIO :: MonadResource m- => IO state -- ^ resource and/or state allocation- -> (state -> IO ()) -- ^ resource and/or state cleanup- -> (state -> input -> m (ConduitIOResult input output)) -- ^ Push function. Note that this need not explicitly perform any cleanup.- -> (state -> m [output]) -- ^ Close function. Note that this need not explicitly perform any cleanup.- -> Conduit input m output-conduitIO alloc cleanup push0 close0 = NeedInput- (\input -> PipeM $ do- (key, state) <- allocate alloc cleanup- push key state input)- (\() -> PipeM $ do- (key, state) <- allocate alloc cleanup- os <- close0 state- release key- return $ sourceList os)- where- push key state input = do- res <- push0 state input- case res of- IOProducing output -> return $ haveMore- (NeedInput (PipeM . push key state) (\() -> close key state))- (release key)- output- IOFinished leftover output -> do- release key- return $ maybe id pipePush leftover $ haveMore- (Done ())- (return ())- output-- close key state = PipeM $ do- output <- close0 state- release key- return $ sourceList output---- | Return value from a 'SequencedSink'.------ Since 0.3.0-data SequencedSinkResponse state input m output =- Emit state [output] -- ^ Set a new state, and emit some new output.- | Stop -- ^ End the conduit.- | 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.3.0-type SequencedSink state input m output =- state -> Sink input m (SequencedSinkResponse state input m output)---- | Convert a 'SequencedSink' into a 'Conduit'.------ Since 0.3.0-sequenceSink- :: Monad m- => state -- ^ initial state- -> SequencedSink state input m output- -> Conduit input m output-sequenceSink state0 fsink = do- x <- hasInput- if x- then do- res <- sinkToPipe $ fsink state0- case res of- Emit state os -> do- sourceList os- sequenceSink state fsink- Stop -> return ()- StartConduit c -> c- else return ()--pipePush :: Monad m => i -> Pipe i i o u m r -> Pipe i i o u m r-pipePush i (HaveOutput p c o) = HaveOutput (pipePush i p) c o-pipePush i (NeedInput p _) =- case p i of- Leftover p' i' -> pipePush i' p'- p' -> p'-pipePush i (Done r) = Leftover (Done r) i-pipePush i (PipeM mp) = PipeM (pipePush i `liftM` mp)-pipePush i (Leftover p i') =- case pipePush i' p of- Leftover p'' i'' -> Leftover (Leftover p'' i'') i- p' -> pipePush i p'---- | Check if input is available from upstream. Will not remove the data from--- the stream.------ Since 0.4.0-hasInput :: Pipe i i o u m Bool -- FIXME consider removing-hasInput = NeedInput (Leftover (Done True)) (const $ Done False)
− Data/Conduit/Util/Sink.hs
@@ -1,81 +0,0 @@--- | Utilities for constructing 'Sink's. Please see "Data.Conduit.Types.Sink"--- for more information on the base types.-module Data.Conduit.Util.Sink- ( sinkState- , SinkStateResult (..)- , sinkIO- , SinkIOResult (..)- ) where--import Control.Monad.Trans.Resource-import Control.Monad.Trans.Class (lift)-import Data.Conduit.Internal---- | A helper type for @sinkState@, indicating the result of being pushed to.--- It can either indicate that processing is done, or to continue with the--- updated state.------ Since 0.3.0-data SinkStateResult state input output =- StateDone (Maybe input) output- | StateProcessing state---- | Construct a 'Sink' with some stateful functions. This function addresses--- threading the state value for you.------ Since 0.3.0-sinkState- :: Monad m- => state -- ^ initial state- -> (state -> input -> m (SinkStateResult state input output)) -- ^ push- -> (state -> m output) -- ^ Close. Note that the state is not returned, as it is not needed.- -> Sink input m output-sinkState state0 push0 close0 =- NeedInput (push state0) (\() -> close state0)- where- push state input = PipeM- (do- res <- state `seq` push0 state input- case res of- StateProcessing state' -> return $ NeedInput (push state') (\() -> close state')- StateDone mleftover output -> return $ maybe id (flip Leftover) mleftover $ Done output)-- close = lift . close0---- | A helper type for @sinkIO@, indicating the result of being pushed to. It--- can either indicate that processing is done, or to continue.------ Since 0.3.0-data SinkIOResult input output = IODone (Maybe input) output | IOProcessing---- | Construct a 'Sink'. Note that your push and close functions need not--- explicitly perform any cleanup.------ Since 0.3.0-sinkIO :: MonadResource m- => IO state -- ^ resource and/or state allocation- -> (state -> IO ()) -- ^ resource and/or state cleanup- -> (state -> input -> m (SinkIOResult input output)) -- ^ push- -> (state -> m output) -- ^ close- -> Sink input m output-sinkIO alloc cleanup push0 close0 = NeedInput- (\input -> PipeM $ do- (key, state) <- allocate alloc cleanup- push key state input)- (\() -> do- (key, state) <- lift $ allocate alloc cleanup- lift $ close key state)- where- push key state input = do- res <- push0 state input- case res of- IODone a b -> do- release key- return $ maybe id (flip Leftover) a $ Done b- IOProcessing -> return $ NeedInput- (PipeM . push key state)- (\() -> lift $ close key state)- close key state = do- res <- close0 state- release key- return res
− Data/Conduit/Util/Source.hs
@@ -1,93 +0,0 @@--- | 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- , sourceStateIO- , SourceStateResult (..)- , sourceIO- , SourceIOResult (..)- ) where--import Control.Monad.Trans.Resource-import Data.Conduit.Internal---- | The return value when pulling in the @sourceState@ function. Either--- indicates no more data, or the next value and an updated state.------ Since 0.3.0-data SourceStateResult state output = StateOpen state output | StateClosed---- | Construct a 'Source' with some stateful functions. This function addresses--- threading the state value for you.------ Since 0.3.0-sourceState- :: Monad m- => state -- ^ Initial state- -> (state -> m (SourceStateResult state output)) -- ^ Pull function- -> Source m output-sourceState state0 pull0 =- src state0- where- src state = PipeM (pull state)-- pull state = do- res <- pull0 state- return $ case res of- StateOpen state' val -> HaveOutput (src state') (return ()) val- StateClosed -> Done ()---- | The return value when pulling in the @sourceIO@ function. Either indicates--- no more data, or the next value.------ Since 0.3.0-data SourceIOResult output = IOOpen output | IOClosed---- | Construct a 'Source' based on some IO actions for alloc/release.------ Since 0.3.0-sourceIO :: MonadResource m- => IO state -- ^ resource and/or state allocation- -> (state -> IO ()) -- ^ resource and/or state cleanup- -> (state -> m (SourceIOResult output)) -- ^ Pull function. Note that this should not perform any cleanup.- -> Source m output-sourceIO alloc cleanup pull0 =- PipeM (do- (key, state) <- allocate alloc cleanup- pull key state)- where- src key state = PipeM (pull key state)-- pull key state = do- res <- pull0 state- case res of- IOClosed -> do- release key- return $ Done ()- IOOpen val -> return $ HaveOutput (src key state) (release key) val---- | A combination of 'sourceIO' and 'sourceState'.------ Since 0.3.0-sourceStateIO :: MonadResource m- => IO state -- ^ resource and/or state allocation- -> (state -> IO ()) -- ^ resource and/or state cleanup- -> (state -> m (SourceStateResult state output)) -- ^ Pull function. Note that this need not explicitly perform any cleanup.- -> Source m output-sourceStateIO alloc cleanup pull0 =- PipeM (do- (key, state) <- allocate alloc cleanup- pull key state)- where- src key state = PipeM (pull key state)-- pull key state = do- res <- pull0 state- case res of- StateClosed -> do- release key- return $ Done ()- StateOpen state' val -> return $ HaveOutput (src key state') (release key) val---- FIXME transPipe
conduit.cabal view
@@ -1,11 +1,13 @@ Name: conduit-Version: 0.5.6+Version: 1.0.0 Synopsis: Streaming data processing library. Description: @conduit@ is a solution to the streaming data problem, allowing for production, transformation, and consumption of streams of data in constant memory. It is an alternative to lazy I\/O which guarantees deterministic resource handling, and fits in the same general solution space as @enumerator@/@iteratee@ and @pipes@. For a brief tutorial, please see the "Data.Conduit" module. . Release history: .+ [1.0] Simplified the user-facing interface back to the Source, Sink, and Conduit types, with Producer and Consumer for generic code. Error messages have been simplified, and optional leftovers and upstream terminators have been removed from the external API. Some long-deprecated functions were finally removed.+ . [0.5] The internals of the package are now separated to the .Internal module, leaving only the higher-level interface in the advertised API. Internally, switched to a @Leftover@ constructor and slightly tweaked the finalization semantics. . [0.4] Inspired by the design of the pipes package: we now have a single unified type underlying @Source@, @Sink@, and @Conduit@. This type is named @Pipe@. There are type synonyms provided for the other three types. Additionally, @BufferedSource@ is no longer provided. Instead, the connect-and-resume operator, @$$+@, can be used for the same purpose.@@ -27,20 +29,17 @@ Homepage: http://github.com/snoyberg/conduit extra-source-files: test/main.hs, test/random -flag debug- default: True- description: Turn on some runtime check to ensure invariants are respected.- flag nohandles+ default: False 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+ if flag(nohandles)+ other-modules: System.PosixFile+ cpp-options: -DNO_HANDLES Exposed-modules: Data.Conduit Data.Conduit.Binary Data.Conduit.Text@@ -48,9 +47,6 @@ Data.Conduit.Lazy Data.Conduit.Internal Data.Conduit.Util- Other-modules: Data.Conduit.Util.Source- Data.Conduit.Util.Sink- Data.Conduit.Util.Conduit Build-depends: base >= 4.3 && < 5 , resourcet >= 0.4.3 && < 0.5 , lifted-base >= 0.1@@ -62,8 +58,6 @@ , text >= 0.11 , void >= 0.5.5 && < 0.6 ghc-options: -Wall- if flag(debug)- cpp-options: -DDEBUG test-suite test hs-source-dirs: test
test/main.hs view
@@ -201,8 +201,9 @@ it "map, left >+>" $ do x <- runResourceT $- CL.sourceList [1..10]- C.>+> CL.map (* 2)+ CI.ConduitM+ (CI.unConduitM (CL.sourceList [1..10])+ CI.>+> CI.injectLeftovers (CI.unConduitM $ CL.map (* 2))) C.$$ CL.fold (+) 0 x `shouldBe` 2 * sum [1..10 :: Int] @@ -298,20 +299,19 @@ describe "lazy" $ do it' "works inside a ResourceT" $ runResourceT $ do counter <- liftIO $ I.newIORef 0- let incr i = C.sourceIO- (liftIO $ I.newIORef $ C.IOOpen (i :: Int))- (const $ return ())- (\istate -> do- res <- liftIO $ I.atomicModifyIORef istate- (\state -> (C.IOClosed, state))- case res of- C.IOClosed -> return ()- _ -> do- count <- liftIO $ I.atomicModifyIORef counter- (\j -> (j + 1, j + 1))- liftIO $ count `shouldBe` i- return res- )+ let incr i = do+ istate <- liftIO $ I.newIORef $ Just (i :: Int)+ let loop = do+ res <- liftIO $ I.atomicModifyIORef istate ((,) Nothing)+ case res of+ Nothing -> return ()+ Just x -> do+ count <- liftIO $ I.atomicModifyIORef counter+ (\j -> (j + 1, j + 1))+ liftIO $ count `shouldBe` i+ C.yield x+ loop+ loop nums <- CLazy.lazyConsume $ mconcat $ map incr [1..10] liftIO $ nums `shouldBe` [1..10] @@ -344,35 +344,6 @@ C.$$ CL.consume res `shouldBe` [3, 5, 7, 9, 11, 13, 15, 17, 19, 21, 11] -- 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 `shouldBe` [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 `shouldBe` [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 `shouldBe` [6, 8, 10] #endif describe "peek" $ do@@ -568,25 +539,25 @@ describe "termination" $ do it "terminates early" $ do- let src = forever $ CI.yield ()+ let src = forever $ C.yield () x <- src C.$$ CL.head x `shouldBe` Just () it "bracket" $ do ref <- I.newIORef (0 :: Int)- let src = CI.bracketP+ let src = C.bracketP (I.modifyIORef ref (+ 1)) (\() -> I.modifyIORef ref (+ 2))- (\() -> forever $ CI.yield (1 :: Int))+ (\() -> forever $ C.yield (1 :: Int)) val <- C.runResourceT $ src C.$$ CL.isolate 10 C.=$ CL.fold (+) 0 val `shouldBe` 10 i <- I.readIORef ref i `shouldBe` 3 it "bracket skipped if not needed" $ do ref <- I.newIORef (0 :: Int)- let src = CI.bracketP+ let src = C.bracketP (I.modifyIORef ref (+ 1)) (\() -> I.modifyIORef ref (+ 2))- (\() -> forever $ CI.yield (1 :: Int))+ (\() -> forever $ C.yield (1 :: Int)) src' = CL.sourceList $ repeat 1 val <- C.runResourceT $ (src' >> src) C.$$ CL.isolate 10 C.=$ CL.fold (+) 0 val `shouldBe` 10@@ -594,20 +565,20 @@ i `shouldBe` 0 it "bracket + toPipe" $ do ref <- I.newIORef (0 :: Int)- let src = CI.bracketP+ let src = C.bracketP (I.modifyIORef ref (+ 1)) (\() -> I.modifyIORef ref (+ 2))- (\() -> forever $ CI.yield (1 :: Int))+ (\() -> forever $ C.yield (1 :: Int)) val <- C.runResourceT $ src C.$$ CL.isolate 10 C.=$ CL.fold (+) 0 val `shouldBe` 10 i <- I.readIORef ref i `shouldBe` 3 it "bracket skipped if not needed" $ do ref <- I.newIORef (0 :: Int)- let src = CI.bracketP+ let src = C.bracketP (I.modifyIORef ref (+ 1)) (\() -> I.modifyIORef ref (+ 2))- (\() -> forever $ CI.yield (1 :: Int))+ (\() -> forever $ C.yield (1 :: Int)) src' = CL.sourceList $ repeat 1 val <- C.runResourceT $ (src' >> src) C.$$ CL.isolate 10 C.=$ CL.fold (+) 0 val `shouldBe` 10@@ -618,20 +589,21 @@ it "leftovers without input" $ do ref <- I.newIORef [] let add x = I.modifyIORef ref (x:)- adder = CI.NeedInput (\a -> liftIO (add a) >> adder) return- residue x = CI.Leftover (CI.Done ()) x+ adder' = CI.NeedInput (\a -> liftIO (add a) >> adder') return+ adder = CI.ConduitM adder'+ residue x = CI.ConduitM $ CI.Leftover (CI.Done ()) x - _ <- CI.yield 1 C.$$ adder+ _ <- C.yield 1 C.$$ adder x <- I.readIORef ref x `shouldBe` [1 :: Int] I.writeIORef ref [] - _ <- CI.yield 1 C.$$ (residue 2 >> residue 3) >> adder+ _ <- C.yield 1 C.$$ (residue 2 >> residue 3) >> adder y <- I.readIORef ref y `shouldBe` [1, 2, 3] I.writeIORef ref [] - _ <- CI.yield 1 C.$$ residue 2 >> (residue 3 >> adder)+ _ <- C.yield 1 C.$$ residue 2 >> (residue 3 >> adder) z <- I.readIORef ref z `shouldBe` [1, 2, 3] I.writeIORef ref []@@ -640,12 +612,12 @@ it' "yield terminates" $ do let is = [1..10] ++ undefined src [] = return ()- src (x:xs) = CI.yield x >> src xs+ src (x:xs) = C.yield x >> src xs x <- src is C.$$ CL.take 10 x `shouldBe` [1..10 :: Int] it' "yield terminates (2)" $ do let is = [1..10] ++ undefined- x <- mapM_ CI.yield is C.$$ CL.take 10+ x <- mapM_ C.yield is C.$$ CL.take 10 x `shouldBe` [1..10 :: Int] it' "yieldOr finalizer called" $ do iref <- I.newIORef (0 :: Int)@@ -656,23 +628,23 @@ describe "upstream results" $ do it' "works" $ do- let foldUp :: (b -> a -> b) -> b -> C.Pipe l a Void u IO (u, b)- foldUp f b = C.awaitE >>= either (\u -> return (u, b)) (\a -> let b' = f b a in b' `seq` foldUp f b')- passFold :: (b -> a -> b) -> b -> C.Pipe l a a () IO b- passFold f b = C.await >>= maybe (return b) (\a -> let b' = f b a in b' `seq` C.yield a >> passFold f b')- (x, y) <- CI.runPipe $ CL.sourceList [1..10 :: Int] C.>+> passFold (+) 0 C.>+> foldUp (*) 1+ let foldUp :: (b -> a -> b) -> b -> CI.Pipe l a Void u IO (u, b)+ foldUp f b = CI.awaitE >>= either (\u -> return (u, b)) (\a -> let b' = f b a in b' `seq` foldUp f b')+ passFold :: (b -> a -> b) -> b -> CI.Pipe l a a () IO b+ passFold f b = CI.await >>= maybe (return b) (\a -> let b' = f b a in b' `seq` CI.yield a >> passFold f b')+ (x, y) <- CI.runPipe $ mapM_ CI.yield [1..10 :: Int] CI.>+> passFold (+) 0 CI.>+> foldUp (*) 1 (x, y) `shouldBe` (sum [1..10], product [1..10]) describe "input/output mapping" $ do it' "mapOutput" $ do- x <- CI.mapOutput (+ 1) (CL.sourceList [1..10 :: Int]) C.$$ CL.fold (+) 0+ x <- C.mapOutput (+ 1) (CL.sourceList [1..10 :: Int]) C.$$ CL.fold (+) 0 x `shouldBe` sum [2..11] it' "mapOutputMaybe" $ do- x <- CI.mapOutputMaybe (\i -> if even i then Just i else Nothing) (CL.sourceList [1..10 :: Int]) C.$$ CL.fold (+) 0+ x <- C.mapOutputMaybe (\i -> if even i then Just i else Nothing) (CL.sourceList [1..10 :: Int]) C.$$ CL.fold (+) 0 x `shouldBe` sum [2, 4..10] it' "mapInput" $ do xyz <- (CL.sourceList $ map show [1..10 :: Int]) C.$$ do- (x, y) <- CI.mapInput read (Just . show) $ ((do+ (x, y) <- C.mapInput read (Just . show) $ ((do x <- CL.isolate 5 C.=$ CL.fold (+) 0 y <- CL.peek return (x :: Int, y :: Maybe Int)) :: C.Sink Int IO (Int, Maybe Int))@@ -683,27 +655,36 @@ describe "left/right identity" $ do it' "left identity" $ do- x <- CL.sourceList [1..10 :: Int] C.$$ CI.idP C.=$ CL.fold (+) 0+ x <- CL.sourceList [1..10 :: Int] C.$$ CI.ConduitM CI.idP C.=$ CL.fold (+) 0 y <- CL.sourceList [1..10 :: Int] C.$$ CL.fold (+) 0 x `shouldBe` y it' "right identity" $ do- x <- CI.runPipe $ CL.sourceList [1..10 :: Int] C.>+> CL.fold (+) 0 C.>+> CI.idP- y <- CI.runPipe $ CL.sourceList [1..10 :: Int] C.>+> CL.fold (+) 0+ x <- CI.runPipe $ mapM_ CI.yield [1..10 :: Int] CI.>+> (CI.injectLeftovers $ CI.unConduitM $ CL.fold (+) 0) CI.>+> CI.idP+ y <- CI.runPipe $ mapM_ CI.yield [1..10 :: Int] CI.>+> (CI.injectLeftovers $ CI.unConduitM $ CL.fold (+) 0) x `shouldBe` y describe "generalizing" $ do it' "works" $ do- x <- C.runPipe+ x <- CI.runPipe $ CI.sourceToPipe (CL.sourceList [1..10 :: Int])- C.>+> CI.conduitToPipe (CL.map (+ 1))- C.>+> CI.sinkToPipe (CL.fold (+) 0)+ CI.>+> CI.conduitToPipe (CL.map (+ 1))+ CI.>+> CI.sinkToPipe (CL.fold (+) 0) x `shouldBe` sum [2..11] describe "withUpstream" $ do it' "works" $ do- let src = CL.sourceList [1..10 :: Int] >> return True- sink = C.withUpstream $ CL.fold (+) 0- res <- C.runPipe $ src C.>+> sink+ let src = mapM_ CI.yield [1..10 :: Int] >> return True+ fold f =+ loop+ where+ loop accum =+ CI.await >>= maybe (return accum) go+ where+ go a =+ let accum' = f accum a+ in accum' `seq` loop accum'+ sink = CI.withUpstream $ fold (+) 0+ res <- CI.runPipe $ src CI.>+> sink res `shouldBe` (True, sum [1..10]) describe "iterate" $ do@@ -784,26 +765,32 @@ x3 `shouldBe` 1 describe "injectLeftovers" $ do it "works" $ do- let src = CL.sourceList [1..10 :: Int]- conduit = C.awaitForever $ \i -> do+ let src = mapM_ CI.yield [1..10 :: Int]+ conduit = CI.injectLeftovers $ CI.unConduitM $ C.awaitForever $ \i -> do js <- CL.take 2 mapM_ C.leftover $ reverse js C.yield i- res <- (src C.>+> C.injectLeftovers conduit) C.$$ CL.consume+ res <- CI.ConduitM (src CI.>+> CI.injectLeftovers conduit) C.$$ CL.consume res `shouldBe` [1..10] describe "up-upstream finalizers" $ do- let p1 = C.await >>= maybe (return ()) C.yield- p2 = idMsg "p2-final"- p3 = idMsg "p3-final"- idMsg msg = C.addCleanup (const $ tell [msg]) $ C.awaitForever C.yield- printer = C.awaitForever $ lift . tell . return . show- src = CL.sourceList [1 :: Int ..] it "pipe" $ do- let run p = execWriter $ C.runPipe $ printer C.<+< p C.<+< src- run (p1 C.<+< (p2 C.<+< p3)) `shouldBe` run ((p1 C.<+< p2) C.<+< p3)+ let p1 = CI.await >>= maybe (return ()) CI.yield+ p2 = idMsg "p2-final"+ p3 = idMsg "p3-final"+ idMsg msg = CI.addCleanup (const $ tell [msg]) $ CI.awaitForever CI.yield+ printer = CI.awaitForever $ lift . tell . return . show+ src = mapM_ CI.yield [1 :: Int ..]+ let run' p = execWriter $ CI.runPipe $ printer CI.<+< p CI.<+< src+ run' (p1 CI.<+< (p2 CI.<+< p3)) `shouldBe` run' ((p1 CI.<+< p2) CI.<+< p3) it "conduit" $ do- let run p = execWriter $ src C.$$ p C.=$ printer- run ((p3 C.=$= p2) C.=$= p1) `shouldBe` run (p3 C.=$= (p2 C.=$= p1))+ let p1 = C.await >>= maybe (return ()) C.yield+ p2 = idMsg "p2-final"+ p3 = idMsg "p3-final"+ idMsg msg = C.addCleanup (const $ tell [msg]) $ C.awaitForever C.yield+ printer = C.awaitForever $ lift . tell . return . show+ src = CL.sourceList [1 :: Int ..]+ let run' p = execWriter $ src C.$$ p C.=$ printer+ run' ((p3 C.=$= p2) C.=$= p1) `shouldBe` run' (p3 C.=$= (p2 C.=$= p1)) describe "monad transformer laws" $ do it "transPipe" $ do let source = CL.sourceList $ replicate 10 ()@@ -853,6 +840,15 @@ assert $ c1 == c2 assert $ s1 == s2++ describe "generalizing" $ do+ it "works" $ do+ let src :: Int -> C.Source IO Int+ src i = CL.sourceList [1..i]+ sink :: C.Sink Int IO Int+ sink = CL.fold (+) 0+ res <- C.yield 10 C.$$ C.awaitForever (C.toProducer . src) C.=$ (C.toConsumer sink >>= C.yield) C.=$ C.await+ res `shouldBe` Just (sum [1..10]) it' :: String -> IO () -> Spec it' = it