conduit 0.4.2 → 0.5.0

raw patch · 14 files changed

+1554/−887 lines, 14 filesdep ~basedep ~voidPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: base, void

API changes (from Hackage documentation)

Files

• Data/Conduit.hs +477/−77
• Data/Conduit/Binary.hs +147/−153
• Data/Conduit/Internal.hs +445/−303
• Data/Conduit/Lazy.hs +9/−6
• Data/Conduit/List.hs +126/−203
• Data/Conduit/Text.hs +28/−34
• Data/Conduit/Util.hs +65/−0
• Data/Conduit/Util/Conduit.hs +37/−42
• Data/Conduit/Util/Sink.hs +9/−18
• Data/Conduit/Util/Source.hs +8/−12
• System/PosixFile.hsc +3/−4
• System/Win32File.hsc +7/−4
• conduit.cabal +6/−2
• test/main.hs +187/−29

@@ -1,54 +1,65 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE UndecidableInstances #-}-{-# LANGUAGE DeriveDataTypeable #-}-{-# LANGUAGE FunctionalDependencies #-}--- | The main module, exporting types, utility functions, and fuse and connect--- operators.------ There are three main types in this package: @Source@ (data producer), @Sink@--- (data consumer), and @Conduit@ (data transformer). All three are in fact--- type synonyms for the underlying @Pipe@ data type.------ The typical approach to use of this package is:------ * Compose multiple @Sink@s together using its @Monad@ instance.------ * Left-fuse @Source@s and @Conduit@s into new @Conduit@s.------ * Right-fuse @Conduit@s and @Sink@s into new @Sink@s.------ * Middle-fuse two @Conduit@s into a new @Conduit@.------ * Connect a @Source@ to a @Sink@ to obtain a result. module Data.Conduit-    ( -- * Types-      Pipe (..)-    , Source+    ( -- * Conduit interface+      -- $conduitInterface+      Source     , Conduit     , Sink-      -- * Connect/fuse operators+      -- ** Connect/fuse+      -- $lifecycle     , ($$)-    , ($$+)     , ($=)     , (=$)     , (=$=)-      -- * Utility functions-      -- ** General++      -- * Pipe interface+      -- $pipeInterface+    , Pipe+    , (>+>)+    , (<+<)+    , runPipe+    , injectLeftovers++      -- * Primitives+      -- $primitives     , await+    , awaitE+    , awaitForever     , yield-    , hasInput+    , yieldOr+    , leftover+      -- ** Finalization+    , bracketP+    , addCleanup++      -- * Connect-and-resume+      -- $connectAndResume+    , ResumableSource+    , ($$+)+    , ($$++)+    , ($$+-)++      -- * Utility functions     , transPipe     , mapOutput-      -- ** Source-    , module Data.Conduit.Util.Source-      -- ** Sink-    , module Data.Conduit.Util.Sink-      -- ** Conduit-    , module Data.Conduit.Util.Conduit+    , mapOutputMaybe+    , mapInput+    , withUpstream++      -- * Generalized conduit types+      -- $generalizedConduitTypes+    , GSource+    , GSink+    , GLSink+    , GInfSink+    , GLInfSink+    , GConduit+    , GLConduit+    , GInfConduit+    , GLInfConduit+       -- * Flushing     , Flush (..)+       -- * Convenience re-exports     , ResourceT     , MonadResource@@ -59,47 +70,382 @@  import Control.Monad.Trans.Resource import Data.Conduit.Internal-import Data.Conduit.Util.Source-import Data.Conduit.Util.Sink-import Data.Conduit.Util.Conduit+import Data.Void (Void) --- $typeOverview+{- $conduitInterface +Let's start off with a few simple examples of @conduit@ usage. First, a file+copy utility:++>>> :load Data.Conduit.Binary+>>> runResourceT $ sourceFile "input.txt" $$ sinkFile "output.txt"++@runResourceT@ is a function provided by the @resourcet@ package, and ensures+that resources are properly cleaned up, even in the presence of exceptions. The+type system will enforce that @runResourceT@ is called as needed. The remainder+of this tutorial will not discuss @runResourceT@; please see the documentation+in @resourcet@ for more information.++Looking at the rest of our example, there are three components to understand:+@sourceFile@, @sinkFile@, and the @$$@ operator (called \"connect\"). These+represent the most basic building blocks in @conduit@: a @Source@ produces a+stream of values, a @Sink@ consumes such a stream, and @$$@ will combine these+together.++In the case of file copying, there was no value produced by the @Sink@.+However, often times a @Sink@ will produce some result value. For example:++>>> :load Data.Conduit.List+>>> :module +Prelude+>>> sourceList [1..10] $$ fold (+) 0+55++@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.+This type represents a stream /transformer/. In order to use a @Conduit@, we+must /fuse/ it with either a @Source@ or @Sink@. For example:++>>> :load Data.Conduit.List+>>> :module +Prelude+>>> sourceList [1..10] $= Data.Conduit.List.map (+1) $$ consume+[2,3,4,5,6,7,8,9,10,11]++Notice the addition of the @$=@, or /left fuse/ operator. This combines a+@Source@ and a @Conduit@ into a new @Source@, which can then be connected to a+@Sink@ (in this case, @consume@). We can similarly perform /right fusion/ to+combine a @Conduit@ and @Sink@, or /middle fusion/ to combine two @Conduit@s.++A number of very common functions are provided in the "Data.Conduit.List"+module. Many of these functions correspond very closely to standard Haskell+functions.++In addition to connecting and fusing components together, we can also build up+more sophisticated components through monadic composition. For example, to+create a @Sink@ that ignores the first 3 numbers and returns the sum of the+remaining numbers, we can use:++>>> :load Data.Conduit.List+>>> :module +Prelude+>>> sourceList [1..10] $$ Data.Conduit.List.drop 3 >> fold (+) 0+49++In some cases, we might end up consuming more input than we needed, and want to+provide that input to the next component in our monadic chain. We refer to this+as /leftovers/. The simplest example of this is @peek@.++>>> :load Data.Conduit.List+>>> :set -XNoMonomorphismRestriction+>>> :module +Prelude+>>> let sink = do { first <- peek; total <- fold (+) 0; return (first, total) }+>>> sourceList [1..10] $$ sink+(Just 1,55)++Notice that, although we \"consumed\" the first value from the stream via+@peek@, it was still available to @fold@. This idea becomes even more important+when dealing with chunked data such as @ByteString@s or @Text@.++/Final note/: Notice in the types below that @Source@, @Sink@, and @Conduit@+are just type aliases. This will be explained later. Another important aspect+is resource finalization, which will also be covered below.++-}++{- $lifecycle++It is important to understand the lifecycle of our components. Notice that we+can connect or fuse two components together. When we do this, the component+providing output is called /upstream/, and the component consuming this input+is called /downstream/. We can have arbitrarily long chains of such fusion, so+a single component can simultaneously function as /upstream/ and /downstream/.++Each component can be in one of four states of operation at any given time:++* It hasn't yet started operating.++* It is providing output downstream.++* It is waiting for input from upstream.++* It has completed processing.++Let\'s use @sourceFile@ and @sinkFile@ as an example. When we run @sourceFile+"input" $$ sinkFile "output"@, both components begin in the \"not started\"+state. Next, we start running @sinkFile@ (note: we /always/ begin processing on+the downstream component). @sinkFile@ will open up the file, and then wait for+input from upstream.++Next, we\'ll start running @sourceFile@, which will open the file, read some+data from it, and provide it as output downstream. This will be fed to+@sinkFile@ (which was already waiting). @sinkFile@ will write the data to a+file, then ask for more input. This process will continue until @sourceFile@+reaches the end of the input. It will close the file handle and switch to the+completed state. When this happens, @sinkFile@ is sent a signal that no more+input is available. It will then close its file and return a result.++Now let\'s change things up a bit. Suppose we were instead connecting+@sourceFile@ to @take 1@. We start by running @take 1@, which will wait for+some input. We\'ll then start @sourceFile@, which will open the file, read a+chunk, and send it downstream. @take 1@ will take that single chunk and return+it as a result. Once it does this, it has transitioned to the complete state.++We don\'t want to pull any more data from @sourceFile@, as we do not need it.+So instead, we call @sourceFile@\'s finalizer. Each time upstream provides+output, it also provides a finalizer to be run if downstream finishes+processing.++One final case: suppose we connect @sourceFile@ to @return ()@. The latter does+nothing: it immediately switches to the complete state. In this case, we never+even start running @sourceFile@ (it stays in the \"not yet started\" state),+and so no finalization occurs.++So here are the takeaways from the above discussion:++* When upstream completes before downstream, it cleans up all of its resources+  and sends some termination signal. We never think about upstream again. This+  can only occur while downstream is in the \"waiting for input\" state, since+  that is the only time that upstream is called.++* When downstream completes before upstream, we finalize upstream immediately.+  This can only occur when upstream produces output, because that\'s the only+  time when control is passed back to downstream.++* If downstream never awaits for input before it terminates, upstream was never+  started, and therefore it does not need to be finalized.++Note that all of the discussion above applies equally well to chains of+components. If you have an upstream, middle, and downstream component, and+downstream terminates, then the middle component will be finalized, which in+turn will trigger upstream to be finalized. This setup ensures that we always+have prompt resource finalization.++-}++{- $pipeInterface++We discussed three main types in the @conduit@ package: @Source@, @Sink@, and+@Conduit@. In fact, these are all unified into a single type, @Pipe@. This+greatly simplifies the internal workings of this package, and makes it much+easier to build more powerful components from simpler ones. For example, it is+easy to combine a number of simple @Sink@s together to produce a more powerful+@Conduit@. To create a @Conduit@ which drops 3 input elements and doubles the+rest, we could use:++>>> :load Data.Conduit.List+>>> :set -XNoMonomorphismRestriction+>>> let conduit = do { drop 3; map (Prelude.* 2) }+>>> sourceList [1..10] $$ conduit =$ consume+[8,10,12,14,16,18,20]++If we look again at our examples from above, we'll see a few different aspects+to @Pipe@s:++* @Sink@s and @Conduit@s can consume a stream of /input/ values. Both @map@ and+  @fold@ took a stream of @Int@s, while @sinkFile@ took a stream of+  @ByteString@s.++* @Source@s and @Conduit@s can produce a stream of /output/ values. @sourceFile@+  produced a stream of @ByteString@s, which was then consumed by @sinkFile@. This+  is an important point in @conduit@: the output of the left-hand pipe (a.k.a.,+  /upstream/) must match the input of the right-hand pipe (a.k.a., /downstream/).++* All @Pipe@s have some underlying @Monad@. The @sourceFile@ and @sinkFile@+  functions needed to use @MonadResource@ from @resourcet@ to get exception+  handling, but our other functions could live in any monad. Since @Pipe@+  provides a @MonadTrans@ instance, you can actually lift any action from the+  underlying @Monad@ into your @Pipe@.++* @Sink@s can provide a /result/ type. Our @fold@ returned a final @Int@, while+  @sinkFile@ returned @()@.++A @Pipe@ also exposes two other features as well, not covered by the above+three types:++* Each @Pipe@ has some /leftover/ value. Above, we described a situation where+  the leftover would be identical to the input type. However, @Pipe@ provides a+  type parameter for this instead, so that you can alternatively set the leftover+  type to 'Void', thereby ensuring that a @Pipe@ does not provide any leftover+  values. This is important for ensuring that leftover values aren't accidentally+  discarded.++* Above, we described a situation where only @Sink@s could return results.+  However, sometimes it's advantageous to allow stream producers to also produce+  a result type. We call this the /upstream/ result.++Putting this all together, a @Pipe@ has six type parameters: @Pipe l i o u m r@,+corresponding to each of the bullets above. @Source@, @Conduit@, and @Sink@ are+simply type aliases that restrict one or more of these type parameters to+specific types. For example, both @Source@ and @Conduit@ have @r@ restricted to+@()@, since neither may return a result.++There are two ways that @Pipe@s can be composed: via the @Monad@ instance, and+via fusion. (/Note/: connecting is just a special case of fusion, where the+@Pipe@ is then run. We'll discuss that more later on.) In the @pipes@ package,+these are referred to as /vertical/ and /horizontal/ composition, respectively.+Let's clarify the distinction between these two:++Monadic composition takes two @Pipe@s with the same input and output types, and+combines them into a single @Pipe@. These two @Pipe@s will be run one after the+other, and they will share the same input and output streams. Essentially, the+second @Pipe@ will continue consuming input where the first left off, and the+output streams of each will be concatenated. Any leftover values from the first+@Pipe@ will be fed to the second @Pipe@. Let\'s see a simple example:++>>> :load Data.Conduit.List+>>> sourceList [1..10] $$ do { x <- take 3; y <- take 2; return (x, y) }+([1,2,3],[4,5])++Fusion, on the other hand, will connect the output from an upstream @Pipe@ to+the input of a downstream @Pipe@. The upstream @Pipe@ is required to have a+result type of @()@, since any results it produces are thrown out. This form of+composition produces a new @Pipe@ with the input parameter of the upstream+@Pipe@ and the output and result parameters of the downstream @Pipe@. (For+examples, see the initial examples on this page. Every usage of the connect or+fusion operators is fusion composition.)++/Note/: If you are building a library of @conduit@ functions, it is best to+keep the type signatures as general as possible. For example, even though the+simplest type signature for the @drop@ function would be @Int -> Sink i m ()@,+this would prevent it from being used in construction of @Conduit@s. Instead,+we give it a type signature of @Int -> Pipe l i o u m ()@.++-}++{- $primitives++While @conduit@ provides a number of built-in @Source@s, @Sink@s, and+@Conduit@s, you will almost certainly want to construct some of your own.+Previous versions recommended using the constructors directly. Beginning with+0.5, the recommended approach is to compose existing @Pipe@s into larger ones.++It is certainly possible (and advisable!) to leverage existing @Pipe@s- like+those in "Data.Conduit.List". However, you will often need to go to a lower+level set of @Pipe@s to start your composition. The following few functions+should be sufficient for expressing all constructs besides finalization. Adding+in @bracketP@ and @addCleanup@, you should be able to create any @Pipe@ you+need. (In fact, that's precisely how the remainder of this package is written.)++The three basic operations are /awaiting/, /yielding/, and /leftovers/.+Awaiting asks for a new value from upstream, or returns @Nothing@ if upstream+is done. For example:++>>> :load Data.Conduit.List+>>> sourceList [1..10] $$ await+Just 1++>>> :load Data.Conduit.List+>>> sourceList [] $$ await+Nothing++Similarly, we have a @yield@ function, which provides a value to the downstream+@Pipe@. @yield@ features /auto-termination/: if the downstream @Pipe@ has+already completed processing, the upstream @Pipe@ will stop processing when it+tries to @yield@.++The upshot of this is that you can write code that appears to loop infinitely,+and yet will terminate.++>>> :set -XNoMonomorphismRestriction+>>> let infinite = yield () >> infinite+>>> infinite $$ await+Just ()++Or for something a bit more sophisticated:++>>> let enumFrom' i = yield i >> enumFrom' (succ i)+>>> enumFrom' 1 $$ take 5+[1,2,3,4,5]++The final primitive @Pipe@ is @leftover@. This allows you to return unused+input to be used by the next @Pipe@ in the monadic chain. A simple use case+would be implementing the @peek@ function:++>>> let peek = await >>= maybe (return Nothing) (\x -> leftover x >> return (Just x))+>>> enumFrom' 1 $$ do { mx <- peek; my <- await; mz <- await; return (mx, my, mz) }+(Just 1,Just 1,Just 2)++Note that you should only return leftovers that were previously yielded from+upstream.++-}++{- $connectAndResume++Sometimes, we do not want to force our entire application to live inside the+@Pipe@ monad. It can be convenient to keep normal control flow of our program,+and incrementally apply data from a @Source@ to various @Sink@s. A strong+motivating example for this use case is interleaving multiple @Source@s, such+as combining a @conduit@-powered HTTP server and client into an HTTP proxy.++Normally, when we run a @Pipe@, we get a result and can never run it again.+Connect-and-resume allows us to connect a @Source@ to a @Sink@ until the latter+completes, and then return the current state of the @Source@ to be applied+later. To do so, we introduce three new operators. Let\' start off by+demonstrating them:++>>> :load Data.Conduit.List+>>> (next, x) <- sourceList [1..10] $$+ take 5+>>> Prelude.print x+[1,2,3,4,5]+>>> (next, y) <- next $$++ (isolate 4 =$ fold (Prelude.+) 0)+>>> Prelude.print y+30+>>> next $$+- consume+[10]++-}++{- $generalizedConduitTypes++It's recommended to keep your type signatures as general as possible to+encourage reuse. For example, a theoretical signature for the @head@ function+would be:++> head :: Sink a m (Maybe a)++However, doing so would prevent usage of @head@ from inside a @Conduit@, since+a @Sink@ sets its output type parameter to @Void@. The most general type+signature would instead be:++> head :: Pipe l a o u m (Maybe a)++However, that signature is much more confusing. To bridge this gap, we also+provide some generalized conduit types. They follow a simple naming convention:++* They have the same name as their non-generalized types, with a @G@ prepended.++* If they have leftovers, we add an @L@.++* If they consume the entirety of their input stream and return the upstream+  result, we add @Inf@ to indicate /infinite consumption/.++-}++-- Define fixity of all our operators infixr 0 $$-infixr 0 $$+ infixl 1 $= infixr 2 =$ infixr 2 =$=-+infixr 9 <+<+infixl 9 >+>+infixr 0 $$++infixr 0 $$+++infixr 0 $$+-  -- | The connect operator, which pulls data from a source and pushes to a sink.--- There are two ways this process can terminate:------ 1. If the @Sink@ is a @Done@ constructor, the @Source@ is closed.------ 2. If the @Source@ is a @Done@ constructor, the @Sink@ is closed.------ In other words, both the @Source@ and @Sink@ will always be closed. If you--- would like to keep the @Source@ open to be used for another operations, use--- the connect-and-resume operators '$$+'.+-- When either side closes, the other side will immediately be closed as well.+-- If you would like to keep the @Source@ open to be used for another+-- operations, use the connect-and-resume operator '$$+'. -- -- Since 0.4.0 ($$) :: Monad m => Source m a -> Sink a m b -> m b-src $$ sink = runPipe $ pipe src sink+src $$ sink = do+    (rsrc, res) <- src $$+ sink+    rsrc $$+- return ()+    return res {-# INLINE ($$) #-} --- | The connect-and-resume operator. 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 in the @Sink@--- monad.------ Mnemonic: connect + do more.------ Since 0.4.0-($$+) :: Monad m => Source m a -> Sink a m b -> m (Source m a, b)-src $$+ sink = runPipe $ pipeResume src sink-{-# INLINE ($$+) #-}- -- | Left fuse, combining a source and a conduit together into a new source. -- -- Both the @Source@ and @Conduit@ will be closed when the newly-created@@ -109,7 +455,7 @@ -- -- Since 0.4.0 ($=) :: Monad m => Source m a -> Conduit a m b -> Source m b-($=) = pipe+($=) = pipeL {-# INLINE ($=) #-}  -- | Right fuse, combining a conduit and a sink together into a new sink.@@ -121,23 +467,77 @@ -- -- Since 0.4.0 (=$) :: Monad m => Conduit a m b -> Sink b m c -> Sink a m c-(=$) = pipe+(=$) = pipeL {-# INLINE (=$) #-} --- | Fusion operator, combining two @Pipe@s together into a new @Pipe@.------ Both @Pipe@s will be closed when the newly-created @Pipe@ is closed.+-- | Fusion operator, combining two @Conduit@s together into a new @Conduit@. ----- Leftover data returned from the right @Pipe@ will be discarded.+-- Both @Conduit@s will be closed when the newly-created @Conduit@ is closed. ----- Note: in previous versions, this operator would only fuse together two--- @Conduit@s (known as middle fusion). This operator is generalized to work on--- all @Pipe@s, including @Source@s and @Sink@s.+-- Leftover data returned from the right @Conduit@ will be discarded. -- -- Since 0.4.0-(=$=) :: Monad m => Pipe a b m () -> Pipe b c m r -> Pipe a c m r-(=$=) = pipe+(=$=) :: Monad m => Conduit a m b -> Conduit b m c -> Conduit a m c+(=$=) = pipeL {-# 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+-- in the @Sink@ monad.+--+-- Mnemonic: connect + do more.+--+-- Since 0.5.0+($$+) :: Monad m => Source m a -> Sink a m b -> m (ResumableSource m a, b)+src $$+ sink = connectResume (ResumableSource src (return ())) sink+{-# INLINE ($$+) #-}++-- | Continue processing after usage of @$$+@.+--+-- Since 0.5.0+($$++) :: Monad m => ResumableSource m a -> Sink a m b -> m (ResumableSource m a, b)+($$++) = connectResume+{-# INLINE ($$++) #-}++-- | Complete processing of a @ResumableSource@. This will run the finalizer+-- associated with the @ResumableSource@. In order to guarantee process resource+-- finalization, you /must/ use this operator after using @$$+@ and @$$++@.+--+-- Since 0.5.0+($$+-) :: Monad m => ResumableSource m a -> Sink a m b -> m b+rsrc $$+- sink = do+    (ResumableSource _ final, res) <- connectResume rsrc sink+    final+    return res+{-# INLINE ($$+-) #-}  -- | Provide for a stream of data that can be flushed. --

@@ -1,65 +1,68 @@-{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE CPP #-} -- | Functions for interacting with bytes. module Data.Conduit.Binary-    ( sourceFile+    ( -- * Files and @Handle@s++      -- | Note that most of these functions live in the @MonadResource@ monad+      -- to ensure resource finalization even in the presence of exceptions. In+      -- order to run such code, you will need to use @runResourceT@.++      -- ** Sources+      sourceFile     , sourceHandle     , sourceIOHandle     , sourceFileRange+      -- ** Sinks     , sinkFile     , sinkHandle     , sinkIOHandle+      -- ** Conduits     , conduitFile-    , isolate-    , openFile+      -- * Utilities+      -- ** Sources+    , sourceLbs+      -- ** Sinks     , head-    , takeWhile     , dropWhile     , take+    , drop+      -- ** Conduits+    , isolate+    , takeWhile     , Data.Conduit.Binary.lines     ) where -import Prelude hiding (head, take, takeWhile, dropWhile)+import Prelude hiding (head, take, drop, takeWhile, dropWhile) import qualified Data.ByteString as S import qualified Data.ByteString.Lazy as L-import Data.Conduit+import Data.Conduit hiding (Source, Conduit, Sink, Pipe)+import Data.Conduit.List (sourceList) import Control.Exception (assert)-import Control.Monad (liftM)+import Control.Monad (unless) import Control.Monad.IO.Class (liftIO, MonadIO) import qualified System.IO as IO-import Control.Monad.Trans.Resource (allocate, release) import Data.Word (Word8)-import Data.Monoid (mempty)+import Control.Applicative ((<$>)) #if CABAL_OS_WINDOWS import qualified System.Win32File as F #elif NO_HANDLES import qualified System.PosixFile as F #endif --- | Open a file 'IO.Handle' safely by automatically registering a release--- action.------ While you are not required to call @hClose@ on the resulting handle, you--- should do so as early as possible to free scarce resources.------ Since 0.3.0-openFile :: MonadResource m-         => FilePath-         -> IO.IOMode-         -> m IO.Handle-openFile fp mode = liftM snd $ allocate (IO.openBinaryFile fp mode) IO.hClose- -- | Stream the contents of a file as binary data. -- -- Since 0.3.0 sourceFile :: MonadResource m            => FilePath-           -> Source m S.ByteString+           -> GSource m S.ByteString sourceFile fp = #if CABAL_OS_WINDOWS || NO_HANDLES-    sourceIO (F.openRead fp)-             F.close-             (liftIO . F.read)+    bracketP+        (F.openRead fp)+         F.close+         loop+  where+    loop h = liftIO (F.read h) >>= maybe (return ()) (\bs -> yield bs >> loop h) #else     sourceIOHandle (IO.openBinaryFile fp IO.ReadMode) #endif@@ -71,35 +74,26 @@ -- Since 0.3.0 sourceHandle :: MonadIO m              => IO.Handle-             -> Source m S.ByteString+             -> GSource m S.ByteString sourceHandle h =-    src+    loop   where-    src = PipeM pull close--    pull = do+    loop = do         bs <- liftIO (S.hGetSome h 4096)         if S.null bs-            then return $ Done Nothing ()-            else return $ HaveOutput src close bs--    close = return ()+            then return ()+            else yield bs >> loop  -- | An alternative to 'sourceHandle'. -- Instead of taking a pre-opened 'IO.Handle', it takes an action that opens -- a 'IO.Handle' (in read mode), so that it can open it only when needed--- and close it as soon as possible.+-- and closed it as soon as possible. -- -- Since 0.3.0 sourceIOHandle :: MonadResource m                => IO IO.Handle-               -> Source m S.ByteString-sourceIOHandle alloc = sourceIO alloc IO.hClose-    (\handle -> do-        bs <- liftIO (S.hGetSome handle 4096)-        if S.null bs-            then return IOClosed-            else return $ IOOpen bs)+               -> GSource m S.ByteString+sourceIOHandle alloc = bracketP alloc IO.hClose sourceHandle  -- | Stream all incoming data to the given 'IO.Handle'. Note that this function -- will /not/ automatically close the @Handle@ when processing completes.@@ -107,14 +101,8 @@ -- Since 0.3.0 sinkHandle :: MonadIO m            => IO.Handle-           -> Sink S.ByteString m ()-sinkHandle h =-    NeedInput push close-  where-    push input = PipeM-        (liftIO (S.hPut h input) >> return (NeedInput push close))-        (return ())-    close = return ()+           -> GInfSink S.ByteString m+sinkHandle h = awaitForever $ liftIO . S.hPut h  -- | An alternative to 'sinkHandle'. -- Instead of taking a pre-opened 'IO.Handle', it takes an action that opens@@ -124,10 +112,8 @@ -- Since 0.3.0 sinkIOHandle :: MonadResource m              => IO IO.Handle-             -> Sink S.ByteString m ()-sinkIOHandle alloc = sinkIO alloc IO.hClose-    (\handle bs -> liftIO (S.hPut handle bs) >> return IOProcessing)-    (const $ return ())+             -> GInfSink S.ByteString m+sinkIOHandle alloc = bracketP alloc IO.hClose sinkHandle  -- | Stream the contents of a file as binary data, starting from a certain -- offset and only consuming up to a certain number of bytes.@@ -137,51 +123,44 @@                 => FilePath                 -> Maybe Integer -- ^ Offset                 -> Maybe Integer -- ^ Maximum count-                -> Source m S.ByteString-sourceFileRange fp offset count = PipeM-    (do-        (key, handle) <- allocate (IO.openBinaryFile fp IO.ReadMode) IO.hClose+                -> GSource m S.ByteString+sourceFileRange fp offset count = bracketP+    (IO.openBinaryFile fp IO.ReadMode)+    IO.hClose+    start+  where+    start handle = do         case offset of             Nothing -> return ()             Just off -> liftIO $ IO.hSeek handle IO.AbsoluteSeek off         case count of-            Nothing -> pullUnlimited handle key-            Just c -> pullLimited c handle key)-    (return ())-  where-    pullUnlimited handle key = do+            Nothing -> pullUnlimited handle+            Just c -> pullLimited (fromInteger c) handle++    pullUnlimited handle = do         bs <- liftIO $ S.hGetSome handle 4096         if S.null bs-            then do-                release key-                return $ Done Nothing ()+            then return ()             else do-                let src = PipeM-                        (pullUnlimited handle key)-                        (release key)-                return $ HaveOutput src (release key) bs+                yield bs+                pullUnlimited handle -    pullLimited c0 handle key = do-        let c = fromInteger c0+    pullLimited c handle = do         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 $ Done Nothing ()+                then return ()                 else do-                    let src = PipeM-                            (pullLimited (toInteger c') handle key)-                            (release key)-                    return $ HaveOutput src (release key) bs+                    yield bs+                    pullLimited c' handle  -- | Stream all incoming data to the given file. -- -- Since 0.3.0 sinkFile :: MonadResource m          => FilePath-         -> Sink S.ByteString m ()+         -> GInfSink 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@@ -190,14 +169,13 @@ -- Since 0.3.0 conduitFile :: MonadResource m             => FilePath-            -> Conduit S.ByteString m S.ByteString-conduitFile fp = conduitIO+            -> GInfConduit S.ByteString m S.ByteString+conduitFile fp = bracketP     (IO.openBinaryFile fp IO.WriteMode)     IO.hClose-    (\handle bs -> do-        liftIO $ S.hPut handle bs-        return $ IOProducing [bs])-    (const $ return [])+    go+  where+    go h = awaitForever $ \bs -> liftIO (S.hPut h bs) >> yield bs  -- | 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@@ -206,112 +184,128 @@ -- Since 0.3.0 isolate :: Monad m         => Int-        -> Conduit S.ByteString m S.ByteString-isolate count0 = conduitState-    count0-    push-    close+        -> GLConduit S.ByteString m S.ByteString+isolate =+    loop   where-    push 0 bs = return $ StateFinished (Just bs) []-    push count bs = do-        let (a, b) = S.splitAt count bs-        let count' = count - S.length a-        return $-            if count' == 0-                then StateFinished (if S.null b then Nothing else Just b) (if S.null a then [] else [a])-                else assert (S.null b) $ StateProducing count' [a]-    close _ = return []+    loop 0 = return ()+    loop count = do+        mbs <- await+        case mbs of+            Nothing -> return ()+            Just bs -> do+                let (a, b) = S.splitAt count bs+                case count - S.length a of+                    0 -> do+                        unless (S.null b) $ leftover b+                        yield a+                    count' -> assert (S.null b) $ yield a >> loop count'  -- | Return the next byte from the stream, if available. -- -- Since 0.3.0-head :: Monad m => Sink S.ByteString m (Maybe Word8)-head =-    NeedInput push close-  where-    push bs =-        case S.uncons bs of-            Nothing -> NeedInput push close-            Just (w, bs') ->-                let lo = if S.null bs' then Nothing else Just bs'-                 in Done lo (Just w)-    close = return Nothing+head :: Monad m => GLSink S.ByteString m (Maybe Word8)+head = do+    mbs <- await+    case mbs of+        Nothing -> return Nothing+        Just bs ->+            case S.uncons bs of+                Nothing -> head+                Just (w, bs') -> leftover bs' >> return (Just w)  -- | Return all bytes while the predicate returns @True@. -- -- Since 0.3.0-takeWhile :: Monad m => (Word8 -> Bool) -> Conduit S.ByteString m S.ByteString+takeWhile :: Monad m => (Word8 -> Bool) -> GLConduit S.ByteString m S.ByteString takeWhile p =-    NeedInput push close+    loop   where-    push bs-        | S.null y =-            let r = NeedInput push close-             in if S.null x-                    then r-                    else HaveOutput r (return ()) x-        | otherwise =-            let f = Done (Just y) ()-             in if S.null x-                    then f-                    else HaveOutput f (return ()) x+    loop = await >>= maybe (return ()) go++    go bs+        | S.null x = next+        | otherwise = yield x >> next       where+        next = if S.null y then loop else leftover y         (x, y) = S.span p bs-    close = mempty  -- | Ignore all bytes while the predicate returns @True@. -- -- Since 0.3.0-dropWhile :: Monad m => (Word8 -> Bool) -> Sink S.ByteString m ()+dropWhile :: Monad m => (Word8 -> Bool) -> GLSink S.ByteString m () dropWhile p =-    NeedInput push close+    loop   where-    push bs-        | S.null bs' = NeedInput push close-        | otherwise  = Done (Just bs') ()-      where-        bs' = S.dropWhile p bs-    close = return ()+    loop = do+        mbs <- await+        case S.dropWhile p <$> mbs of+            Nothing -> return ()+            Just bs+                | S.null bs -> loop+                | otherwise -> leftover bs  -- | Take the given number of bytes, if available. -- -- Since 0.3.0-take :: Monad m => Int -> Sink S.ByteString m L.ByteString+take :: Monad m => Int -> GLSink S.ByteString m L.ByteString take n0 =     go n0 id   where     go n front =-        NeedInput push close+        await >>= maybe (return $ L.fromChunks $ front []) go'       where-        push bs =+        go' bs =             case S.length bs `compare` n of                 LT -> go (n - S.length bs) (front . (bs:))-                EQ -> Done Nothing $ L.fromChunks $ front [bs]+                EQ -> return $ L.fromChunks $ front [bs]                 GT ->                     let (x, y) = S.splitAt n bs-                     in Done (Just y) $ L.fromChunks $ front [x]-        close = return $ L.fromChunks $ front []+                     in assert (not $ S.null y) $ leftover y >> return (L.fromChunks $ front [x]) +-- | Drop up to the given number of bytes.+--+-- Since 0.5.0+drop :: Monad m => Int -> GLSink S.ByteString m ()+drop =+    go+  where+    go n =+        await >>= maybe (return ()) go'+      where+        go' bs =+            case S.length bs `compare` n of+                LT -> go (n - S.length bs)+                EQ -> return ()+                GT ->+                    let y = S.drop n bs+                     in assert (not $ S.null y) $ leftover y >> return ()+ -- | Split the input bytes into lines. In other words, split on the LF byte -- (10), and strip it from the output. -- -- Since 0.3.0-lines :: Monad m => Conduit S.ByteString m S.ByteString-lines = NeedInput (push id) (close S.empty)+lines :: Monad m => GInfConduit S.ByteString m S.ByteString+lines =+    loop id   where-    push :: Monad m => (S.ByteString -> S.ByteString)-                    -> S.ByteString-                    -> Conduit S.ByteString m S.ByteString-    push sofar more =+    loop front = awaitE >>= either (finish front) (go front)++    finish front r =+        let final = front S.empty+         in unless (S.null final) (yield final) >> return r++    go sofar more =         case S.uncons second of-            Just (_, second') -> HaveOutput (push id second') (return ()) (sofar first)+            Just (_, second') -> yield (sofar first) >> go id second'             Nothing ->                 let rest = sofar more-                 in NeedInput (push $ S.append rest) (close rest)+                 in loop $ S.append rest       where         (first, second) = S.breakByte 10 more -    close rest-        | S.null rest = Done Nothing ()-        | otherwise   = HaveOutput (Done Nothing ()) (return ()) rest-+-- | Stream the chunks from a lazy bytestring.+--+-- Since 0.5.0+sourceLbs :: Monad m => L.ByteString -> GSource m S.ByteString+sourceLbs = sourceList . L.toChunks

@@ -1,33 +1,56 @@ {-# OPTIONS_HADDOCK not-home #-}-{-# OPTIONS_GHC -O2 #-} -- necessary to avoid some space leaks-{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE UndecidableInstances #-} {-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TupleSections #-} module Data.Conduit.Internal     ( -- * Types       Pipe (..)     , Source+    , GSource     , Sink+    , GSink+    , GLSink+    , GInfSink+    , GLInfSink     , Conduit-    , Finalize (..)-      -- * Functions-    , pipeClose+    , GConduit+    , GLConduit+    , GInfConduit+    , GLInfConduit+    , ResumableSource (..)+      -- * Primitives+    , await+    , awaitE+    , awaitForever+    , yield+    , yieldOr+    , leftover+      -- * Finalization+    , bracketP+    , addCleanup+      -- * Composition+    , idP     , pipe-    , pipeResume+    , pipeL+    , connectResume     , runPipe+    , injectLeftovers+      -- * Generalizing+    , sourceToPipe     , sinkToPipe-    , await-    , yield-    , hasInput+    , conduitToPipe+      -- * Utilities     , transPipe     , mapOutput-    , runFinalize-    , addCleanup+    , mapOutputMaybe+    , mapInput+    , sourceList+    , withUpstream     ) where -import Control.Applicative (Applicative (..), (<$>))+import Control.Applicative (Applicative (..)) import Control.Monad ((>=>), liftM, ap) import Control.Monad.Trans.Class (MonadTrans (lift)) import Control.Monad.IO.Class (MonadIO (liftIO))@@ -35,366 +58,485 @@ import Data.Void (Void, absurd) import Data.Monoid (Monoid (mappend, mempty)) import Control.Monad.Trans.Resource---- | A cleanup action to be performed.------ Previously, we just had a plain action. However, most @Pipe@s simply have--- empty cleanup actions, and storing a large set of them wastes memory. But--- having strict fields and distinguishing between pure and impure actions, we--- can keep memory usage constant, and only allocate memory for the actual--- actions we have to track.------ Since 0.4.1-data Finalize m r = FinalizePure r-                  | FinalizeM (m r)--instance Monad m => Functor (Finalize m) where-    fmap f (FinalizePure r) = FinalizePure (f r)-    fmap f (FinalizeM mr) = FinalizeM (liftM f mr)--instance Monad m => Applicative (Finalize m) where-    pure = FinalizePure-    (<*>) = ap--instance Monad m => Monad (Finalize m) where-    return = FinalizePure-    FinalizePure x >>= f = f x-    FinalizeM mx >>= f = FinalizeM $ mx >>= \x ->-        case f x of-            FinalizePure y -> return y-            FinalizeM my -> my--instance MonadTrans Finalize where-    lift = FinalizeM--instance MonadThrow m => MonadThrow (Finalize m) where-    monadThrow = lift . monadThrow--instance MonadIO m => MonadIO (Finalize m) where-    liftIO = lift . liftIO--instance MonadResource m => MonadResource (Finalize m) where-    allocate a = lift . allocate a-    register = lift . register-    release = lift . release-    resourceMask = lift . resourceMask+import qualified GHC.Exts --- | The underlying datatype for all the types in this package.  In has four+-- | The underlying datatype for all the types in this package.  In has six -- type parameters: --+-- * /l/ is the type of values that may be left over from this @Pipe@. A @Pipe@+-- with no leftovers would use @Void@ here, and one with leftovers would use+-- the same type as the /i/ parameter. Leftovers are automatically provided to+-- the next @Pipe@ in the monadic chain.+-- -- * /i/ is the type of values for this @Pipe@'s input stream. -- -- * /o/ is the type of values for this @Pipe@'s output stream. --+-- * /u/ is the result type from the upstream @Pipe@.+-- -- * /m/ is the underlying monad. -- -- * /r/ is the result type. ----- Note that /o/ and /r/ are inherently different. /o/ is the type of the--- stream of values this @Pipe@ will produce and send downstream. /r/ is the--- final output of this @Pipe@.------ @Pipe@s can be composed via the 'pipe' function. To do so, the output type--- of the left pipe much match the input type of the left pipe, and the result--- type of the left pipe must be unit @()@. This is due to the fact that any--- result produced by the left pipe must be discarded in favor of the result of--- the right pipe.+-- A basic intuition is that every @Pipe@ produces a stream of output values+-- (/o/), and eventually indicates that this stream is terminated by sending a+-- result (/r/). On the receiving end of a @Pipe@, these become the /i/ and /u/+-- parameters. ----- Since 0.4.0-data Pipe i o m r =+-- Since 0.5.0+data Pipe l i o u m r =     -- | Provide new output to be sent downstream. This constructor has three-    -- fields: the next @Pipe@ to be used, an early-closed function, and the+    -- fields: the next @Pipe@ to be used, a finalization function, and the     -- output value.-    HaveOutput (Pipe i o m r) (Finalize m r) o+    HaveOutput (Pipe l i o u m r) (m ()) o     -- | Request more input from upstream. The first field takes a new input-    -- value and provides a new @Pipe@. The second is for early termination. It-    -- gives a new @Pipe@ which takes no input from upstream. This allows a-    -- @Pipe@ to provide a final stream of output values after no more input is-    -- available from upstream.-  | NeedInput (i -> Pipe i o m r) (Pipe i o m r)-    -- | Processing with this @Pipe@ is complete. Provides an optional leftover-    -- input value and and result.-  | Done (Maybe i) r-    -- | Require running of a monadic action to get the next @Pipe@. Second-    -- field is an early cleanup function. Technically, this second field-    -- could be skipped, but doing so would require extra operations to be-    -- performed in some cases. For example, for a @Pipe@ pulling data from a-    -- file, it may be forced to pull an extra, unneeded chunk before closing-    -- the @Handle@.-  | PipeM (m (Pipe i o m r)) (Finalize m r)---- | A @Pipe@ which provides a stream of output values, without consuming any--- input. The input parameter is set to @Void@ to indicate that this @Pipe@--- takes no input.  A @Source@ is not used to produce a final result, and thus--- the result parameter is set to @()@.------ Since 0.4.0-type Source m a = Pipe Void a m ()+    -- value and provides a new @Pipe@. The second takes an upstream result+    -- value, which indicates that upstream is producing no more results.+  | NeedInput (i -> Pipe l i o u m r) (u -> Pipe l i o u m r)+    -- | Processing with this @Pipe@ is complete, providing the final result.+  | Done r+    -- | Require running of a monadic action to get the next @Pipe@.+  | PipeM (m (Pipe l i o u m r))+    -- | Return leftover input, which should be provided to future operations.+  | Leftover (Pipe l i o u m r) l --- | A @Pipe@ which consumes a stream of input values and produces a final--- result. It cannot produce any output values, and thus the output parameter--- is set to @Void@. In other words, it is impossible to create a @HaveOutput@--- constructor for a @Sink@.------ Since 0.4.0-type Sink i m r = Pipe i Void m r+instance Monad m => Functor (Pipe l i o u m) where+    fmap = liftM --- | A @Pipe@ which consumes a stream of input values and produces a stream of--- output values. It does not produce a result value, and thus the result--- parameter is set to @()@.------ Since 0.4.0-type Conduit i m o = Pipe i o m ()+instance Monad m => Applicative (Pipe l i o u m) where+    pure = return+    (<*>) = ap --- | Perform any close actions available for the given @Pipe@.------ Since 0.4.0-pipeClose :: Monad m => Pipe i o m r -> Finalize m r-pipeClose (HaveOutput _ c _) = c-pipeClose (NeedInput _ p) = pipeClose p-pipeClose (Done _ r) = FinalizePure r-pipeClose (PipeM _ c) = c+instance Monad m => Monad (Pipe l i o u m) where+    return = Done -pipePush :: Monad m => i -> Pipe i o m r -> Pipe i o m r-pipePush i (HaveOutput p c o) = HaveOutput (pipePush i p) c o-pipePush i (NeedInput p _) = p i-pipePush i (Done _ r) = Done (Just i) r-pipePush i (PipeM mp c) = PipeM (pipePush i `liftM` mp) c+    Done x           >>= fp = fp x+    HaveOutput p c o >>= fp = HaveOutput (p >>= fp)            c          o+    NeedInput p c    >>= fp = NeedInput  (p >=> fp)            (c >=> fp)+    PipeM mp         >>= fp = PipeM      ((>>= fp) `liftM` mp)+    Leftover p i     >>= fp = Leftover   (p >>= fp)            i -instance Monad m => Functor (Pipe i o m) where-    fmap f (HaveOutput p c o) = HaveOutput (f <$> p) (f `liftM` c) o-    fmap f (NeedInput p c) = NeedInput (fmap f . p) (f <$> c)-    fmap f (Done l r) = Done l (f r)-    fmap f (PipeM mp mr) = PipeM ((fmap f) `liftM` mp) (f `liftM` mr)+instance MonadBase base m => MonadBase base (Pipe l i o u m) where+    liftBase = lift . liftBase -instance Monad m => Applicative (Pipe i o m) where-    pure = Done Nothing+instance MonadTrans (Pipe l i o u) where+    lift mr = PipeM (Done `liftM` mr) -    Done Nothing f <*> px = f <$> px-    Done (Just i) f <*> px = pipePush i $ f <$> px-    HaveOutput p c o <*> px = HaveOutput (p <*> px) (c `ap` pipeClose px) o-    NeedInput p c <*> px = NeedInput (\i -> p i <*> px) (c <*> px)-    PipeM mp c <*> px = PipeM ((<*> px) `liftM` mp) (c `ap` pipeClose px)+instance MonadIO m => MonadIO (Pipe l i o u m) where+    liftIO = lift . liftIO -instance Monad m => Monad (Pipe i o m) where-    return = Done Nothing+instance Monad m => Monoid (Pipe l i o u m ()) where+    mempty = return ()+    mappend = (>>) -    Done Nothing x >>= fp = fp x-    Done (Just i) x >>= fp = pipePush i $ fp x-    HaveOutput p c o >>= fp = HaveOutput (p >>= fp) (c >>= pipeClose . fp) o-    NeedInput p c >>= fp = NeedInput (p >=> fp) (c >>= fp)-    PipeM mp c >>= fp = PipeM ((>>= fp) `liftM` mp) (c >>= pipeClose . fp)+-- | 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 () -instance MonadBase base m => MonadBase base (Pipe i o m) where-    liftBase = lift . liftBase+-- | Generalized 'Source'.+--+-- Since 0.5.0+type GSource m o = forall l i u. Pipe l i o u m () -instance MonadTrans (Pipe i o) where-    lift mr = PipeM (Done Nothing `liftM` mr) (FinalizeM mr)+-- | 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 -instance MonadIO m => MonadIO (Pipe i o m) where-    liftIO = lift . liftIO+-- | Generalized 'Sink' without leftovers.+--+-- Since 0.5.0+type GSink i m r = forall l o u. Pipe l i o u m r -instance Monad m => Monoid (Pipe i o m ()) where-    mempty = return ()-    mappend = (>>)+-- | Generalized 'Sink' with leftovers.+--+-- Since 0.5.0+type GLSink i m r = forall o u. Pipe i i o u m r --- | Compose a left and right pipe together into a complete pipe. The left pipe--- will be automatically closed when the right pipe finishes, and any leftovers--- from the right pipe will be discarded.+-- | Generalized 'Sink' without leftovers returning upstream result. ----- This is in fact a wrapper around 'pipeResume'. This function closes the left--- @Pipe@ returns by @pipeResume@ and returns only the 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.4.0-pipe :: Monad m => Pipe a b m () -> Pipe b c m r -> Pipe a c m r-pipe l r = pipeResume l r >>= \(l', res) -> lift (runFinalize $ pipeClose l') >> return res+-- Since 0.5.0+type GLInfSink i m = forall o r. Pipe i i o r m r --- | Same as 'pipe', but retain both the new left pipe and the leftovers from--- the right pipe. The two components are combined together into a single pipe--- and returned, together with the result of the right pipe.+-- | Consumes a stream of input values and produces a stream of output values,+-- without producing a final result. ----- Note: we're biased towards checking the right side first to avoid pulling--- extra data which is not needed. Doing so could cause data loss.+-- Since 0.5.0+type Conduit i m o = Pipe i i o () m ()++-- | Generalized conduit without leftovers. ----- Since 0.4.0-pipeResume :: Monad m => Pipe a b m () -> Pipe b c m r -> Pipe a c m (Pipe a b m (), r)-pipeResume left right =-    -- We're using a case statement instead of pattern matching in the function-    -- itself to make the logic explicit. We first check the right pipe, and-    -- only if the right pipe is asking for more input do we process the left-    -- pipe.-    case right of-        -- Right pipe is done, grab leftovers and the left pipe-        Done leftoverr r ->-            -- Get any leftovers from the left pipe, the current state of the-            -- left pipe (sans leftovers), and a close action for the left-            -- pipe.-            let (leftover, left', leftClose) =-                    case left of-                        Done leftoverl () -> (leftoverl, Done Nothing (), FinalizePure ())-                        _ -> (Nothing, left, pipeClose left)-            -- Combine the current state of the left pipe with any leftovers-            -- from the right pipe.-                left'' =-                    case leftoverr of-                        Just a -> HaveOutput left' leftClose a-                        Nothing -> left'-            -- Return the leftovers, the final left pipe state, and the result.-             in Done leftover (left'', r)+-- Since 0.5.0+type GConduit i m o = forall l u. Pipe l i o u m () -        -- Right pipe needs to run a monadic action.-        PipeM mp c -> PipeM-            (pipeResume left `liftM` mp)-            (((,) left) `fmap` c)+-- | Generalized conduit with leftovers.+--+-- Since 0.5.0+type GLConduit i m o = forall u. Pipe i i o u m () -        -- Right pipe has some output, provide it downstream and continue.-        HaveOutput p c o -> HaveOutput-            (pipeResume left p)-            (((,) left) `fmap` c)-            o+-- | 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 -        -- Right pipe needs input, so let's get it-        NeedInput rp rc ->-            case left of-                -- Left pipe has output, right pipe wants it.-                HaveOutput lp _ a -> pipeResume lp $ rp a+-- | 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 -                -- Left pipe needs more input, ask for it.-                NeedInput p c -> NeedInput-                    (\a -> pipeResume (p a) right)-                    (do-                        -- There is no more input available, so connect the-                        -- no-more-input record with the right.-                        (left', res) <- pipeResume c right+-- | A @Source@ which has been started, but has not yet completed.+--+-- This type contains both the current state of the @Source@, and the finalizer+-- to be run to close it.+--+-- Since 0.5.0+data ResumableSource m o = ResumableSource (Source m o) (m ()) -                        -- Theoretically, we could return the left' value as-                        -- the first element in the tuple. However, it is not-                        -- recommended to give input to a pipe after it has-                        -- been told there is no more input. Instead, we close-                        -- the pipe and return mempty in its place.-                        lift $ runFinalize $ pipeClose left'-                        return (mempty, res)-                        )+-- | Wait for a single input value from upstream, terminating immediately if no+-- data is available.+--+-- Since 0.5.0+await :: Pipe l i o u m (Maybe i)+await = NeedInput (Done . Just) (\_ -> Done Nothing)+{-# RULES "await >>= maybe" forall x y. await >>= maybe x y = NeedInput y (const x) #-}+{-# INLINE [1] await #-} -                -- Left pipe is done, right pipe needs input. In such a case,-                -- tell the right pipe there is no more input, and eventually-                -- replace its leftovers with the left pipe's leftover.-                Done l () -> ((,) mempty) `liftM` replaceLeftover l rc+-- | This is similar to @await@, but will return the upstream result value as+-- @Left@ if available.+--+-- Since 0.5.0+awaitE :: Pipe l i o u m (Either u i)+awaitE = NeedInput (Done . Right) (Done . Left)+{-# RULES "awaitE >>= either" forall x y. awaitE >>= either x y = NeedInput y x #-}+{-# INLINE [1] awaitE #-} -                -- Left pipe needs to run a monadic action.-                PipeM mp c -> PipeM-                    ((`pipeResume` right) `liftM` mp)-                    (fmap ((,) mempty) $ combineFinalize c $ pipeClose right)+-- | Wait for input forever, calling the given inner @Pipe@ for each piece of+-- new input. Returns the upstream result type.+--+-- Since 0.5.0+awaitForever :: Monad m => (i -> Pipe l i o r m r') -> Pipe l i o r m r+awaitForever inner =+    self+  where+    self = awaitE >>= either return (\i -> inner i >> self)+{-# INLINE [1] awaitForever #-} --- | A minor optimization on @>>@ which does not cause any allocations for the--- common case of missing left actions.+-- | Send a single output value downstream. If the downstream @Pipe@+-- terminates, this @Pipe@ will terminate as well. ----- Since 0.4.1-combineFinalize :: Monad m => Finalize m () -> Finalize m r -> Finalize m r-combineFinalize (FinalizePure ()) f = f-combineFinalize (FinalizeM x) (FinalizeM y) = FinalizeM $ x >> y-combineFinalize (FinalizeM x) (FinalizePure y) = FinalizeM $ x >> return y+-- Since 0.5.0+yield :: Monad m+      => o -- ^ output value+      -> Pipe l i o u m ()+yield = HaveOutput (Done ()) (return ())+{-# INLINE [1] yield #-} -replaceLeftover :: Monad m => Maybe i -> Pipe i' o m r -> Pipe i o m r-replaceLeftover l (Done _ r) = Done l r-replaceLeftover l (HaveOutput p c o) = HaveOutput (replaceLeftover l p) c o+-- | Similar to @yield@, but additionally takes a finalizer to be run if the+-- downstream @Pipe@ terminates.+--+-- Since 0.5.0+yieldOr :: Monad m+        => o+        -> m () -- ^ finalizer+        -> Pipe l i o u m ()+yieldOr o f = HaveOutput (Done ()) f o+{-# INLINE [1] yieldOr #-} --- This function is only called on pipes when there is no more input available.--- Therefore, we can ignore the push record.-replaceLeftover l (NeedInput _ c) = replaceLeftover l c+{-# RULES+    "yield o >> p" forall o (p :: Pipe l i o u m r). yield o >> p = HaveOutput p (return ()) o+  ; "mapM_ yield" mapM_ yield = sourceList+  ; "yieldOr o c >> p" forall o c (p :: Pipe l i o u m r). yieldOr o c >> p = HaveOutput p c o+  #-} -replaceLeftover l (PipeM mp c) = PipeM (replaceLeftover l `liftM` mp) c+-- | Provide a single piece of leftover input to be consumed by the next pipe+-- in the current monadic binding.+--+-- /Note/: it is highly encouraged to only return leftover values from input+-- already consumed from upstream.+--+-- Since 0.5.0+leftover :: l -> Pipe l i o u m ()+leftover = Leftover (Done ())+{-# INLINE [1] leftover #-}+{-# RULES "leftover l >> p" forall l (p :: Pipe l i o u m r). leftover l >> p = Leftover p l #-} --- | Run a complete pipeline until processing completes.+-- | Perform some allocation and run an inner @Pipe@. Two guarantees are given+-- about resource finalization: ----- Since 0.4.0-runPipe :: Monad m => Pipe Void Void m r -> m r-runPipe (HaveOutput _ c _) = runFinalize c-runPipe (NeedInput _ c) = runPipe c-runPipe (Done _ r) = return r-runPipe (PipeM mp _) = mp >>= runPipe+-- 1. It will be /prompt/. The finalization will be run as early as possible.+--+-- 2. It is exception safe. Due to usage of @resourcet@, the finalization will+--    be run in the event of any exceptions.+--+-- Since 0.5.0+bracketP :: MonadResource m+         => IO a+         -> (a -> IO ())+         -> (a -> Pipe l i o u m r)+         -> Pipe l i o u m r+bracketP alloc free inside =+    PipeM start+  where+    start = do+        (key, seed) <- allocate alloc free+        return $ addCleanup (const $ release key) (inside seed) --- | Perform any necessary finalization actions.+-- | Add some code to be run when the given @Pipe@ cleans up. -- -- Since 0.4.1-runFinalize :: Monad m => Finalize m r -> m r-runFinalize (FinalizePure r) = return r-runFinalize (FinalizeM mr) = mr+addCleanup :: Monad m+           => (Bool -> m ()) -- ^ @True@ if @Pipe@ ran to completion, @False@ for early termination.+           -> Pipe l i o u m r+           -> Pipe l i o u m r+addCleanup cleanup (Done r) = PipeM (cleanup True >> return (Done r))+addCleanup cleanup (HaveOutput src close x) = HaveOutput+    (addCleanup cleanup src)+    (cleanup False >> close)+    x+addCleanup cleanup (PipeM msrc) = PipeM (liftM (addCleanup cleanup) msrc)+addCleanup cleanup (NeedInput p c) = NeedInput+    (addCleanup cleanup . p)+    (addCleanup cleanup . c)+addCleanup cleanup (Leftover p i) = Leftover (addCleanup cleanup p) i --- | Send a single output value downstream.+-- | The identity @Pipe@. ----- Since 0.4.0-yield :: Monad m => o -> Pipe i o m ()-yield = HaveOutput (Done Nothing ()) (FinalizePure ())+-- Since 0.5.0+idP :: Monad m => Pipe l a a r m r+idP = NeedInput (HaveOutput idP (return ())) Done --- | Wait for a single input value from upstream, and remove it from the--- stream. Returns @Nothing@ if no more data is available.+-- | Compose a left and right pipe together into a complete pipe. The left pipe+-- will be automatically closed when the right pipe finishes. ----- Since 0.4.0-await :: Pipe i o m (Maybe i)-await = NeedInput (Done Nothing . Just) (Done Nothing Nothing)+-- Since 0.5.0+pipe :: Monad m => Pipe l a b r0 m r1 -> Pipe Void b c r1 m r2 -> Pipe l a c r0 m r2+pipe =+    pipe' (return ())+  where+    pipe' final left right =+        case right of+            Done r2 -> PipeM (final >> return (Done r2))+            HaveOutput p c o -> HaveOutput (pipe' final left p) c o+            PipeM mp -> PipeM (liftM (pipe' final left) mp)+            Leftover _ i -> absurd i+            NeedInput rp rc -> upstream rp rc+      where+        upstream rp rc =+            case left of+                Done r1 -> pipe (Done r1) (rc r1)+                HaveOutput left' final' o -> pipe' final' left' (rp o)+                PipeM mp -> PipeM (liftM (\left' -> pipe' final left' right) mp)+                Leftover left' i -> Leftover (pipe' final left' right) i+                NeedInput left' lc -> NeedInput+                    (\a -> pipe' final (left' a) right)+                    (\r0 -> pipe' final (lc r0) right) --- | Check if input is available from upstream. Will not remove the data from--- the stream.+-- | Same as 'pipe', but automatically applies 'injectLeftovers' to the right @Pipe@. ----- Since 0.4.0-hasInput :: Pipe i o m Bool-hasInput = NeedInput (\i -> Done (Just i) True) (Done Nothing False)+-- Since 0.5.0+pipeL :: Monad m => Pipe l a b r0 m r1 -> Pipe b b c r1 m r2 -> Pipe l a c r0 m r2+-- Note: The following should be equivalent to the simpler:+--+--     pipeL l r = l `pipe` injectLeftovers r+--+-- However, this version tested as being significantly more efficient.+pipeL =+    pipe' (return ())+  where+    pipe' :: Monad m => m () -> Pipe l a b r0 m r1 -> Pipe b b c r1 m r2 -> Pipe l a c r0 m r2+    pipe' final left right =+        case right of+            Done r2 -> PipeM (final >> return (Done r2))+            HaveOutput p c o -> HaveOutput (pipe' final left p) c o+            PipeM mp -> PipeM (liftM (pipe' final left) mp)+            Leftover right' i -> pipe' final (HaveOutput left final i) right'+            NeedInput rp rc ->+                case left of+                    Done r1 -> pipe' (return ()) (Done r1) (rc r1)+                    HaveOutput left' final' o -> pipe' final' left' (rp o)+                    PipeM mp -> PipeM (liftM (\left' -> pipe' final left' right) mp)+                    NeedInput left' lc -> NeedInput+                        (\a -> pipe' final (left' a) right)+                        (\r0 -> pipe' final (lc r0) right)+                    Leftover left' i -> Leftover (pipe' final left' right) i --- | A @Sink@ has a @Void@ type parameter for the output, which makes it--- difficult to compose with @Source@s and @Conduit@s. This function replaces--- that parameter with a free variable. This function is essentially @id@; it--- only modifies the types, not the actions performed.+-- | Connect a @Source@ to a @Sink@ until the latter closes. Returns both the+-- most recent state of the @Source@ and the result of the @Sink@. ----- Since 0.4.0-sinkToPipe :: Monad m => Sink i m r -> Pipe i o m r-sinkToPipe (HaveOutput _ _ o) = absurd o-sinkToPipe (NeedInput p c) = NeedInput (sinkToPipe . p) (sinkToPipe c)-sinkToPipe (Done i r) = Done i r-sinkToPipe (PipeM mp c) = PipeM (liftM sinkToPipe mp) c+-- We use a @ResumableSource@ to keep track of the most recent finalizer+-- provided by the @Source@.+--+-- Since 0.5.0+connectResume :: Monad m+              => ResumableSource m o+              -> Sink o m r+              -> m (ResumableSource m o, r)+connectResume (ResumableSource left0 leftFinal0) =+    go leftFinal0 left0+  where+    go leftFinal left right =+        case right of+            Done r2 -> return (ResumableSource left leftFinal, r2)+            PipeM mp -> mp >>= go leftFinal left+            HaveOutput _ _ o -> absurd o+            Leftover p i -> go leftFinal (HaveOutput left leftFinal i) p+            NeedInput rp rc ->+                case left of+                    Leftover p () -> go leftFinal p right+                    HaveOutput left' leftFinal' o -> go leftFinal' left' (rp o)+                    NeedInput _ lc -> go leftFinal (lc ()) right+                    Done () -> go (return ()) (Done ()) (rc ())+                    PipeM mp -> mp >>= \left' -> go leftFinal left' right +-- | Run a pipeline until processing completes.+--+-- Since 0.5.0+runPipe :: Monad m => Pipe Void () Void () m r -> m r+runPipe (HaveOutput _ _ o) = absurd o+runPipe (NeedInput _ c) = runPipe (c ())+runPipe (Done r) = return r+runPipe (PipeM mp) = mp >>= runPipe+runPipe (Leftover _ i) = absurd i++-- | Transforms a @Pipe@ that provides leftovers to one which does not,+-- allowing it to be composed.+--+-- This function will provide any leftover values within this @Pipe@ to any+-- calls to @await@. If there are more leftover values than are demanded, the+-- remainder are discarded.+--+-- Since 0.5.0+injectLeftovers :: Monad m => Pipe i i o u m r -> Pipe l i o u m r+injectLeftovers (Done r) = Done r+injectLeftovers (PipeM mp) = PipeM (liftM injectLeftovers mp)+injectLeftovers (NeedInput p c) = NeedInput (injectLeftovers . p) (injectLeftovers . c)+injectLeftovers (HaveOutput p c o) = HaveOutput (injectLeftovers p) c o+injectLeftovers (Leftover p0 i0) =+    injectLeftovers $ inject i0 p0+  where+    inject _ (Done r) = Done r+    inject i (NeedInput p _) = p i+    inject i (PipeM mp) = PipeM $ liftM (inject i) mp+    inject i (HaveOutput p c o) = HaveOutput (inject i p) c o+    inject i (Leftover p i') =+        case inject i' p of+            Leftover p' _ -> p'+            p' -> Leftover p' i+ -- | Transform the monad that a @Pipe@ lives in. -- -- Since 0.4.0-transPipe :: Monad m => (forall a. m a -> n a) -> Pipe i o m r -> Pipe i o n r-transPipe f (HaveOutput p c o) = HaveOutput (transPipe f p) (transFinalize f c) o-transPipe f (NeedInput p c) = NeedInput (transPipe f . p) (transPipe f c)-transPipe _ (Done i r) = Done i r-transPipe f (PipeM mp c) = PipeM (f $ liftM (transPipe f) mp) (transFinalize f c)--transFinalize :: (forall a. m a -> n a) -> Finalize m r -> Finalize n r-transFinalize _ (FinalizePure r) = FinalizePure r-transFinalize f (FinalizeM mr) = FinalizeM $ f mr+transPipe :: Monad m => (forall a. m a -> n a) -> Pipe l i o u m r -> Pipe l i o u n r+transPipe f (HaveOutput p c o) = HaveOutput (transPipe f p) (f c) o+transPipe f (NeedInput p c) = NeedInput (transPipe f . p) (transPipe f . c)+transPipe _ (Done r) = Done r+transPipe f (PipeM mp) = PipeM (f $ liftM (transPipe f) mp)+transPipe f (Leftover p i) = Leftover (transPipe f p) i --- | Apply a function to all the output values of a `Pipe`.+-- | Apply a function to all the output values of a @Pipe@. -- -- This mimics the behavior of `fmap` for a `Source` and `Conduit` in pre-0.4 -- days. -- -- Since 0.4.1-mapOutput :: Monad m => (o1 -> o2) -> Pipe i o1 m r -> Pipe i o2 m r+mapOutput :: Monad m => (o1 -> o2) -> Pipe l i o1 u m r -> Pipe l i o2 u m r mapOutput f (HaveOutput p c o) = HaveOutput (mapOutput f p) c (f o)-mapOutput f (NeedInput p c) = NeedInput (mapOutput f . p) (mapOutput f c)-mapOutput _ (Done i r) = Done i r-mapOutput f (PipeM mp c) = PipeM (liftM (mapOutput f) mp) c+mapOutput f (NeedInput p c) = NeedInput (mapOutput f . p) (mapOutput f . c)+mapOutput _ (Done r) = Done r+mapOutput f (PipeM mp) = PipeM (liftM (mapOutput f) mp)+mapOutput f (Leftover p i) = Leftover (mapOutput f p) i --- | Add some code to be run when the given @Pipe@ cleans up.+-- | Same as 'mapOutput', but use a function that returns @Maybe@ values. ----- Since 0.4.1-addCleanup :: Monad m-           => (Bool -> m ()) -- ^ @True@ if @Pipe@ ran to completion, @False@ for early termination.-           -> Pipe i o m r-           -> Pipe i o m r-addCleanup cleanup (Done leftover r) = PipeM-    (cleanup True >> return (Done leftover r))-    (lift (cleanup True) >> return r)-addCleanup cleanup (HaveOutput src close x) = HaveOutput-    (addCleanup cleanup src)-    (lift (cleanup False) >> close)-    x-addCleanup cleanup (PipeM msrc close) = PipeM-    (liftM (addCleanup cleanup) msrc)-    (lift (cleanup False) >> close)-addCleanup cleanup (NeedInput p c) = NeedInput-    (addCleanup cleanup . p)-    (addCleanup cleanup c)+-- Since 0.5.0+mapOutputMaybe :: Monad m => (o1 -> Maybe o2) -> Pipe l i o1 u m r -> Pipe l i o2 u m r+mapOutputMaybe f (HaveOutput p c o) = maybe id (\o' p' -> HaveOutput p' c o') (f o) (mapOutputMaybe f p)+mapOutputMaybe f (NeedInput p c) = NeedInput (mapOutputMaybe f . p) (mapOutputMaybe f . c)+mapOutputMaybe _ (Done r) = Done r+mapOutputMaybe f (PipeM mp) = PipeM (liftM (mapOutputMaybe f) mp)+mapOutputMaybe f (Leftover p i) = Leftover (mapOutputMaybe f p) i++-- | Apply a function to all the input values of a @Pipe@.+--+-- Since 0.5.0+mapInput :: Monad m+         => (i1 -> i2) -- ^ map initial input to new input+         -> (l2 -> Maybe l1) -- ^ map new leftovers to initial leftovers+         -> Pipe l2 i2 o u m r+         -> Pipe l1 i1 o u m r+mapInput f f' (HaveOutput p c o) = HaveOutput (mapInput f f' p) c o+mapInput f f' (NeedInput p c)    = NeedInput (mapInput f f' . p . f) (mapInput f f' . c)+mapInput _ _  (Done r)           = Done r+mapInput f f' (PipeM mp)         = PipeM (liftM (mapInput f f') mp)+mapInput f f' (Leftover p i)     = maybe id (flip Leftover) (f' i) $ mapInput f f' p++-- | Convert a list into a source.+--+-- Since 0.3.0+sourceList :: Monad m => [a] -> Pipe l i a u m ()+sourceList =+    go+  where+    go [] = Done ()+    go (o:os) = HaveOutput (go os) (return ()) o+{-# INLINE [1] sourceList #-}++-- | The equivalent of @GHC.Exts.build@ for @Pipe@.+--+-- Since 0.4.2+build :: Monad m => (forall b. (o -> b -> b) -> b -> b) -> Pipe l i o u m ()+build g = g (\o p -> HaveOutput p (return ()) o) (return ())++{-# RULES+    "sourceList/build" forall (f :: (forall b. (a -> b -> b) -> b -> b)). sourceList (GHC.Exts.build f) = build f+  #-}++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++sinkToPipe :: Monad m => Sink i m r -> Pipe l i o u m r+sinkToPipe =+    go . injectLeftovers+  where+    go (Done r) = Done r+    go (PipeM mp) = PipeM (liftM go mp)+    go (NeedInput p c) = NeedInput (go . p) (const $ go $ c ())+    go (HaveOutput _ _ o) = absurd o+    go (Leftover _ l) = absurd l++conduitToPipe :: Monad m => Conduit i m o -> Pipe l i o u m ()+conduitToPipe =+    go . injectLeftovers+  where+    go (Done ()) = Done ()+    go (PipeM mp) = PipeM (liftM go mp)+    go (NeedInput p c) = NeedInput (go . p) (const $ go $ c ())+    go (HaveOutput p c o) = HaveOutput (go p) c o+    go (Leftover _ l) = absurd l++-- | Returns a tuple of the upstream and downstream results. Note that this+-- will force consumption of the entire input stream.+--+-- Since 0.5.0+withUpstream :: Monad m+             => Pipe l i o u m r+             -> Pipe l i o u m (u, r)+withUpstream down =+    down >>= go+  where+    go r =+        loop+      where+        loop = awaitE >>= either (\u -> return (u, r)) (\_ -> loop)

@@ -1,12 +1,14 @@ {-# 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.+-- warnings of lazy I\/O apply. In particular, if you are using this with a+-- @ResourceT@ transformer, you must force the list to be evaluated before+-- exiting the @ResourceT@. module Data.Conduit.Lazy     ( lazyConsume     ) where  import Data.Conduit+import Data.Conduit.Internal (Pipe (..)) import System.IO.Unsafe (unsafeInterleaveIO) import Control.Monad.Trans.Control (MonadBaseControl, liftBaseOp_) import Control.Monad.Trans.Resource (MonadActive (monadActive))@@ -17,14 +19,15 @@ -- state has been closed. -- -- Since 0.3.0-lazyConsume :: (MonadBaseControl IO m, MonadActive m) => Source m a -> m [a]-lazyConsume (Done _ ()) = return []+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+lazyConsume (PipeM msrc) = liftBaseOp_ unsafeInterleaveIO $ do     a <- monadActive     if a         then msrc >>= lazyConsume         else return []-lazyConsume (NeedInput _ c) = lazyConsume c+lazyConsume (NeedInput _ c) = lazyConsume (c ())+lazyConsume (Leftover p _) = lazyConsume p

@@ -1,4 +1,4 @@-{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE RankNTypes #-} -- | Higher-level functions to interact with the elements of a stream. Most of -- these are based on list functions. --@@ -14,21 +14,20 @@     , sourceNull     , unfold     , enumFromTo+    , iterate       -- * Sinks       -- ** Pure     , fold     , take     , drop     , head-    , zip-    , zipSinks     , peek     , consume     , sinkNull       -- ** Monadic     , foldM     , mapM_-      -- Conduits+      -- * Conduits       -- ** Pure     , map     , concatMap@@ -40,24 +39,27 @@     , mapM     , concatMapM     , concatMapAccumM+      -- * Misc+    , sequence     ) where +import qualified Prelude import Prelude     ( ($), return, (==), (-), Int     , (.), id, Maybe (..), Monad     , Bool (..)-    , Ordering (..)     , (>>)-    , flip+    , (>>=)     , seq     , otherwise     , Enum (succ), Eq+    , maybe+    , either     )-import Data.Conduit-import Data.Conduit.Internal (pipeClose, runFinalize)-import Data.Monoid (mempty)-import Data.Void (absurd)-import Control.Monad (liftM, liftM2)+import Data.Conduit hiding (Source, Sink, Conduit, Pipe)+import Data.Conduit.Internal (sourceList)+import Control.Monad (when, (<=<))+import Control.Monad.Trans.Class (lift)  -- | Generate a source from a seed value. --@@ -65,14 +67,14 @@ unfold :: Monad m        => (b -> Maybe (a, b))        -> b-       -> Source m a+       -> GSource m a unfold f =     go   where     go seed =         case f seed of-            Just (a, seed') -> HaveOutput (go seed') (return ()) a-            Nothing -> Done Nothing ()+            Just (a, seed') -> yield a >> go seed'+            Nothing -> return ()  -- | Enumerate from a value to a final value, inclusive, via 'succ'. --@@ -84,29 +86,37 @@ enumFromTo :: (Enum a, Eq a, Monad m)            => a            -> a-           -> Source m a+           -> GSource m a enumFromTo start stop =     go start   where     go i-        | i == stop = HaveOutput (Done Nothing ()) (return ()) i-        | otherwise = HaveOutput (go (succ i)) (return ()) i+        | i == stop = yield i+        | 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 f =+    go+  where+    go a = yield a >> go (f a)+ -- | A strict left fold. -- -- Since 0.3.0 fold :: Monad m      => (b -> a -> b)      -> b-     -> Sink a m b-fold f accum0 =-    go accum0+     -> GSink a m b+fold f =+    loop   where-    go accum = NeedInput (push accum) (return accum)--    push accum input =-        let accum' = f accum input-         in accum' `seq` go accum'+    loop accum =+        await >>= maybe (return accum) go+      where+        go a =+            let accum' = f accum a+             in accum' `seq` loop accum'  -- | A monadic strict left fold. --@@ -114,33 +124,24 @@ foldM :: Monad m       => (b -> a -> m b)       -> b-      -> Sink a m b-foldM f accum0 = sinkState-    accum0-    (\accum input -> do-        accum' <- f accum input-        return $ StateProcessing accum'-    )-    return+      -> GSink a m b+foldM f =+    loop+  where+    loop accum = do+        await >>= maybe (return accum) go+      where+        go a = do+            accum' <- lift $ f accum a+            accum' `seq` loop accum'  -- | Apply the action to all values in the stream. -- -- Since 0.3.0 mapM_ :: Monad m       => (a -> m ())-      -> Sink a m ()-mapM_ f =-    NeedInput push close-  where-    push input = PipeM (f input >> return (NeedInput push close)) (return ())-    close = return ()---- | Convert a list into a source.------ Since 0.3.0-sourceList :: Monad m => [a] -> Source m a-sourceList [] = Done Nothing ()-sourceList (x:xs) = HaveOutput (sourceList xs) (return ()) x+      -> GInfSink a m+mapM_ f = awaitForever $ lift . f  -- | Ignore a certain number of values in the stream. This function is -- semantically equivalent to:@@ -153,15 +154,12 @@ -- Since 0.3.0 drop :: Monad m      => Int-     -> Sink a m ()-drop 0 = NeedInput (flip Done () . Just) (return ())-drop count =-    NeedInput push (return ())+     -> GSink a m ()+drop =+    loop   where-    count' = count - 1-    push _-        | count' == 0 = Done Nothing ()-        | otherwise   = drop count'+    loop 0 = return ()+    loop count = await >>= maybe (return ()) (\_ -> loop (count - 1))  -- | 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'.@@ -172,50 +170,33 @@ -- Since 0.3.0 take :: Monad m      => Int-     -> Sink a m [a]-take count0 =-    go count0 id+     -> GSink a m [a]+take =+    loop id   where-    go count front = NeedInput (push count front) (return $ front [])--    push 0 front x = Done (Just x) (front [])-    push count front x-        | count' == 0 = Done Nothing (front [x])-        | otherwise   = NeedInput (push count' front') (return $ front' [])-      where-        count' = count - 1-        front' = front . (x:)+    loop front 0 = return $ front []+    loop front count = await >>= maybe+        (return $ front [])+        (\x -> loop (front .(x:)) (count - 1))  -- | Take a single value from the stream, if available. -- -- Since 0.3.0-head :: Monad m => Sink a m (Maybe a)-head =-    NeedInput push close-  where-    push x = Done Nothing (Just x)-    close = return Nothing+head :: Monad m => GSink 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 => Sink a m (Maybe a)-peek =-    NeedInput push close-  where-    push x = Done (Just x) (Just x)-    close = return Nothing+peek :: Monad m => GLSink 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) -> Conduit a m b-map f =-    NeedInput push close-  where-    push i = HaveOutput (NeedInput push close) (return ()) (f i)-    close = mempty+map :: Monad m => (a -> b) -> GInfConduit a m b+map f = awaitForever $ yield . f  -- | Apply a monadic transformation to all values in a stream. --@@ -223,84 +204,80 @@ -- side-effects of running the action, see 'mapM_'. -- -- Since 0.3.0-mapM :: Monad m => (a -> m b) -> Conduit a m b-mapM f =-    NeedInput push close-  where-    push = flip PipeM (return ()) . liftM (HaveOutput (NeedInput push close) (return ())) . f-    close = mempty+mapM :: Monad m => (a -> m b) -> GInfConduit a m b+mapM f = awaitForever $ yield <=< lift . f  -- | Apply a transformation to all values in a stream, concatenating the output -- values. -- -- Since 0.3.0-concatMap :: Monad m => (a -> [b]) -> Conduit a m b-concatMap f =-    NeedInput push close-  where-    push = haveMore (NeedInput push close) (return ()) . f-    close = mempty+concatMap :: Monad m => (a -> [b]) -> GInfConduit 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]) -> Conduit a m b-concatMapM f =-    NeedInput push close-  where-    push = flip PipeM (return ()) . liftM (haveMore (NeedInput push close) (return ())) . f-    close = mempty+concatMapM :: Monad m => (a -> m [b]) -> GInfConduit 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 -> Conduit a m b-concatMapAccum f accum = conduitState accum push close+concatMapAccum :: Monad m => (a -> accum -> (accum, [b])) -> accum -> GInfConduit a m b+concatMapAccum f =+    loop   where-    push state input = let (state', result) = f input state-                       in return $ StateProducing state' result-    close _ = return []+    loop accum =+        awaitE >>= either return go+      where+        go a = do+            let (accum', bs) = f a accum+            Prelude.mapM_ yield bs+            loop accum'  -- | 'concatMapM' with an accumulator. -- -- Since 0.3.0-concatMapAccumM :: Monad m => (a -> accum -> m (accum, [b])) -> accum -> Conduit a m b-concatMapAccumM f accum = conduitState accum push close+concatMapAccumM :: Monad m => (a -> accum -> m (accum, [b])) -> accum -> GInfConduit a m b+concatMapAccumM f =+    loop   where-    push state input = do (state', result) <- f input state-                          return $ StateProducing state' result-    close _ = return []+    loop accum = do+        awaitE >>= either return go+      where+        go a = do+            (accum', bs) <- lift $ f a accum+            Prelude.mapM_ yield bs+            loop accum'  -- | Consume all values from the stream and return as a list. Note that this -- will pull all values into memory. For a lazy variant, see -- "Data.Conduit.Lazy". -- -- Since 0.3.0-consume :: Monad m => Sink a m [a]+consume :: Monad m => GSink a m [a] consume =-    go id+    loop id   where-    go front = NeedInput (push front) (return $ front [])-    push front x = go (front . (x:))+    loop front = await >>= maybe (return $ front []) (\x -> loop $ front . (x:))  -- | Grouping input according to an equality function. -- -- Since 0.3.0-groupBy :: Monad m => (a -> a -> Bool) -> Conduit a m [a]-groupBy f = conduitState-    []-    push-    close+groupBy :: Monad m => (a -> a -> Bool) -> GInfConduit a m [a]+groupBy f =+    start   where-    push []      v = return $ StateProducing [v] []-    push s@(x:_) v =-      if f x v then-        return $ StateProducing (v:s) []-      else-        return $ StateProducing [v] [s]-    close s = return [s]+    start = awaitE >>= either return (loop id) +    loop rest x =+        awaitE >>= either (\r -> yield (x : rest []) >> return r) go+      where+        go y+            | f x y     = loop (rest . (y:)) x+            | otherwise = yield (x : rest []) >> loop id y+ -- | 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.:@@ -314,95 +291,41 @@ -- >     ... -- -- Since 0.3.0-isolate :: Monad m => Int -> Conduit a m a-isolate count0 = conduitState-    count0-    push-    close+isolate :: Monad m => Int -> GConduit a m a+isolate =+    loop   where-    close _ = return []-    push count x = do-        if count == 0-            then return $ StateFinished (Just x) []-            else do-                let count' = count - 1-                return $ if count' == 0-                    then StateFinished Nothing [x]-                    else StateProducing count' [x]+    loop 0 = return ()+    loop count = await >>= maybe (return ()) (\x -> yield x >> loop (count - 1))  -- | Keep only values in the stream passing a given predicate. -- -- Since 0.3.0-filter :: Monad m => (a -> Bool) -> Conduit a m a-filter f =-    NeedInput push close-  where-    push i | f i = HaveOutput (NeedInput push close) (return ()) i-    push _       = NeedInput push close-    close = mempty+filter :: Monad m => (a -> Bool) -> GInfConduit 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 => Sink a m ()-sinkNull =-    NeedInput push close-  where-    push _ = sinkNull-    close = return ()+sinkNull :: Monad m => GInfSink a m+sinkNull = awaitForever $ \_ -> return () --- | A source that returns nothing. Note that this is just a type-restricted+-- | A source that outputs no values. Note that this is just a type-restricted -- synonym for 'mempty'. -- -- Since 0.3.0-sourceNull :: Monad m => Source m a-sourceNull = mempty---- | Combines two sources. The new source will stop producing once either---   source has been exhausted.------ Since 0.3.0-zip :: Monad m => Source m a -> Source m b -> Source m (a, b)-zip (Done _ ()) (Done _ ()) = Done Nothing ()-zip (Done _ ()) (HaveOutput _ close _) = PipeM (runFinalize close >> return (Done Nothing ())) close-zip (HaveOutput _ close _) (Done _ ()) = PipeM (runFinalize close >> return (Done Nothing ())) close-zip (Done _ ()) (PipeM _ close) = PipeM (runFinalize close >> return (Done Nothing ())) close-zip (PipeM _ close) (Done _ ()) = PipeM (runFinalize close >> return (Done Nothing ())) close-zip (PipeM mx closex) (PipeM my closey) = PipeM (liftM2 zip mx my) (closex >> closey)-zip (PipeM mx closex) y@(HaveOutput _ closey _) = PipeM (liftM (\x -> zip x y) mx) (closex >> closey)-zip x@(HaveOutput _ closex _) (PipeM my closey) = PipeM (liftM (\y -> zip x y) my) (closex >> closey)-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-+sourceNull :: Monad m => GSource m a+sourceNull = return () --- | Combines two sinks. The new sink will complete when both input sinks have---   completed.------ If both sinks finish on the same chunk, and both report leftover input,--- arbitrarily yield the left sink's leftover input.+-- | Run a @Pipe@ repeatedly, and output its result value downstream. Stops+-- when no more input is available from upstream. ----- Since 0.4.1-zipSinks :: Monad m => Sink i m r -> Sink i m r' -> Sink i m (r, r')-zipSinks = zipSinks' EQ--zipSinks' :: Monad m => Ordering -> Sink i m r -> Sink i m r' -> Sink i m (r, r')-zipSinks' byInputUsed = (><)+-- Since 0.5.0+sequence :: Monad m+         => GLSink i m o -- ^ @Pipe@ to run repeatedly+         -> GLInfConduit i m o+sequence sink =+    self   where-    PipeM mpx mx     >< py               = PipeM (liftM (>< py) mpx) (liftM2 (,) mx (pipeClose py))-    px               >< PipeM mpy my     = PipeM (liftM (px ><) mpy) (liftM2 (,) (pipeClose px) my)--    Done ix x        >< Done iy y        = Done i (x, y)-      where-        i = case byInputUsed of-                 EQ -> iy >> ix-                 GT -> ix-                 LT -> iy--    NeedInput fpx px >< NeedInput fpy py = NeedInput (\i -> zipSinks' EQ (fpx i) (fpy i)) (px >< py)-    NeedInput fpx px >< py               = NeedInput (\i -> zipSinks' GT (fpx i) py)      (px >< py)-    px               >< NeedInput fpy py = NeedInput (\i -> zipSinks' LT px (fpy i))      (px >< py)--    HaveOutput _ _ o >< _                = absurd o-    _                >< HaveOutput _ _ o = absurd o+    self = awaitForever $ \i -> leftover i >> sink >>= yield

@@ -1,5 +1,4 @@ {-# LANGUAGE DeriveDataTypeable #-}-{-# LANGUAGE FlexibleContexts #-} -- | -- Copyright: 2011 Michael Snoyman, 2010-2011 John Millikin -- License: MIT@@ -41,10 +40,10 @@ import           System.IO.Unsafe (unsafePerformIO) import           Data.Typeable (Typeable) -import qualified Data.Conduit as C+import Data.Conduit hiding (Source, Conduit, Sink, Pipe) import qualified Data.Conduit.List as CL-import Control.Monad.Trans.Resource (MonadThrow (..)) import Control.Monad.Trans.Class (lift)+import Control.Monad (unless)  -- | A specific character encoding. --@@ -68,28 +67,30 @@ -- | Emit each line separately -- -- Since 0.4.1-lines :: Monad m => C.Conduit T.Text m T.Text+lines :: Monad m => GInfConduit T.Text m T.Text lines =-    C.NeedInput (push id) (close T.empty)+    loop id   where-    push sofar more =+    loop front = awaitE >>= either (finish front) (go front)++    finish front r =+        let final = front T.empty+         in unless (T.null final) (yield final) >> return r++    go sofar more =         case T.uncons second of-            Just (_, second') -> C.HaveOutput (push id second') (return ()) (sofar first')+            Just (_, second') -> yield (sofar first') >> go id second'             Nothing ->                 let rest = sofar more-                 in C.NeedInput (push $ T.append rest) (close rest)+                 in loop $ T.append rest       where         (first', second) = T.break (== '\n') more -    close rest-        | T.null rest = C.Done Nothing ()-        | otherwise   = C.HaveOutput (C.Done Nothing ()) (return ()) rest- -- | Convert text into bytes, using the provided codec. If the codec is -- not capable of representing an input character, an exception will be thrown. -- -- Since 0.3.0-encode :: MonadThrow m => Codec -> C.Conduit T.Text m B.ByteString+encode :: MonadThrow m => Codec -> GInfConduit T.Text m B.ByteString encode codec = CL.mapM $ \t -> do     let (bs, mexc) = codecEncode codec t     maybe (return bs) (monadThrow . fst) mexc@@ -99,31 +100,24 @@ -- not capable of decoding an input byte sequence, an exception will be thrown. -- -- Since 0.3.0-decode :: MonadThrow m => Codec -> C.Conduit B.ByteString m T.Text+decode :: MonadThrow m => Codec -> GInfConduit B.ByteString m T.Text decode codec =-    C.NeedInput push (close B.empty)+    loop id   where-    push bs =+    loop front = awaitE >>= either (finish front) (go front)++    finish front r =+        case B.uncons $ front B.empty of+            Nothing -> return r+            Just (w, _) -> lift $ monadThrow $ DecodeException codec w++    go front bs' =         case extra of-            Left (exc, _) -> C.PipeM (monadThrow exc) (monadThrow exc)-            Right bs' ->-                let push' = if B.null bs' then push else push . B.append bs'-                    close' = close bs'-                 in C.HaveOutput (C.NeedInput push' close') (close2 bs') text+            Left (exc, _) -> lift $ monadThrow exc+            Right bs'' -> yield text >> loop (B.append bs'')       where         (text, extra) = codecDecode codec bs--    close bs =-        case B.uncons bs of-            Nothing -> C.Done Nothing ()-            Just (w, _) ->-                let exc = monadThrow $ DecodeException codec w-                 in C.PipeM exc (lift exc)--    close2 bs =-        case B.uncons bs of-            Nothing -> return ()-            Just (w, _) -> monadThrow $ DecodeException codec w+        bs = front bs'  -- | -- Since 0.3.0@@ -322,7 +316,7 @@ tryEvaluate :: a -> Either TextException a tryEvaluate = unsafePerformIO . Exc.try . Exc.evaluate -maybeDecode:: (a, b) -> Maybe (a, b)+maybeDecode :: (a, b) -> Maybe (a, b) maybeDecode (a, b) = case tryEvaluate a of     Left _ -> Nothing     Right _ -> Just (a, b)

@@ -0,0 +1,65 @@+-- | Utility functions from older versions of @conduit@. These should be+-- 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+    , zipSinks+    ) where++import Prelude hiding (zip)+import Control.Monad (liftM, liftM2)+import Data.Conduit.Internal (Pipe (..), Source, Sink, injectLeftovers)+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 ())++-- | Combines two sinks. The new sink will complete when both input sinks have+--   completed.+--+-- Any leftovers are discarded.+--+-- 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+  where+    (><) :: Monad m => Pipe Void i Void () m r1 -> Pipe Void i Void () m r2 -> Sink i m (r1, r2)++    Leftover _  i    >< _                = absurd i+    _                >< Leftover _  i    = absurd i+    HaveOutput _ _ o >< _                = absurd o+    _                >< HaveOutput _ _ o = absurd o++    PipeM mx         >< y                = PipeM (liftM (>< y) mx)+    x                >< PipeM my         = PipeM (liftM (x ><) my)+    Done x           >< Done y           = Done (x, y)+    NeedInput px cx  >< NeedInput py cy  = NeedInput (\i -> px i >< py i) (\() -> cx () >< cy ())+    NeedInput px cx  >< y@Done{}         = NeedInput (\i -> px i >< y)    (\u -> cx u >< y)+    x@Done{}         >< NeedInput py cy  = NeedInput (\i -> x >< py i)    (\u -> x >< cy u)

@@ -1,7 +1,3 @@-{-# 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@@ -13,14 +9,13 @@       -- *** Sequencing     , SequencedSink     , sequenceSink-    , sequence     , SequencedSinkResponse (..)     ) where  import Prelude hiding (sequence) import Control.Monad.Trans.Resource-import Data.Conduit.Internal-import Control.Monad (liftM, when)+import Data.Conduit.Internal hiding (leftover)+import Control.Monad (liftM)  -- | A helper function for returning a list of values from a @Conduit@. --@@ -30,7 +25,7 @@          -> [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) (FinalizeM close) x+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@@ -56,20 +51,20 @@     -> (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)+    NeedInput (push state0) (\() -> close state0)   where-    push state input = PipeM (liftM goRes' $ state `seq` push0 state input) (return ())+    push state input = PipeM (liftM goRes' $ state `seq` push0 state input)      close state = PipeM (do         os <- close0 state-        return $ fromList os) (return ())+        return $ sourceList os) -    goRes' (StateFinished leftover output) = haveMore-        (Done leftover ())+    goRes' (StateFinished leftover output) = maybe id pipePush leftover $ haveMore+        (Done ())         (return ())         output     goRes' (StateProducing state output) = haveMore-        (NeedInput (push state) (close state))+        (NeedInput (push state) (\() -> close state))         (return ())         output @@ -95,37 +90,33 @@            -> (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 -> flip PipeM (return ()) $ do+    (\input -> PipeM $ do         (key, state) <- allocate alloc cleanup         push key state input)-    (PipeM (do+    (\() -> PipeM $ do         (key, state) <- allocate alloc cleanup         os <- close0 state         release key-        return $ fromList os) (return ()))+        return $ sourceList os)   where     push key state input = do         res <- push0 state input         case res of             IOProducing output -> return $ haveMore-                (NeedInput (flip PipeM (FinalizeM $ release key) . push key state) (close key state))-                (release key >> return ())+                (NeedInput (PipeM . push key state) (\() -> close key state))+                (release key)                 output             IOFinished leftover output -> do                 release key-                return $ haveMore-                    (Done leftover ())+                return $ maybe id pipePush leftover $ haveMore+                    (Done ())                     (return ())                     output -    close key state = PipeM (do+    close key state = PipeM $ do         output <- close0 state         release key-        return $ fromList output) (FinalizeM $ release key)--fromList :: Monad m => [a] -> Pipe i a m ()-fromList [] = Done Nothing ()-fromList (x:xs) = HaveOutput (fromList xs) (return ()) x+        return $ sourceList output  -- | Return value from a 'SequencedSink'. --@@ -158,24 +149,28 @@             res <- sinkToPipe $ fsink state0             case res of                 Emit state os -> do-                    fromList os+                    sourceList os                     sequenceSink state fsink                 Stop -> return ()                 StartConduit c -> c         else return () --- | Specialised version of 'sequenceSink'------ Note that this function will return an infinite stream if provided a--- @Sink@ which does not consume data. In other words, you probably don\'t want to do--- @sequence . 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.3.0-sequence :: Monad m => Sink input m output -> Conduit input m output-sequence sink = self-  where-    self = do-        x <- hasInput-        when x $ do-          sinkToPipe sink >>= yield-          self+-- Since 0.4.0+hasInput :: Pipe i i o u m Bool -- FIXME consider removing+hasInput = NeedInput (Leftover (Done True)) (const $ Done False)

@@ -1,7 +1,3 @@-{-# 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@@ -35,15 +31,14 @@     -> (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)+    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 $ Done mleftover output)-        (FinalizeM $ close0 state)+                StateProcessing state' -> return $ NeedInput (push state') (\() -> close state')+                StateDone mleftover output -> return $ maybe id (flip Leftover) mleftover $ Done output)      close = lift . close0 @@ -64,12 +59,10 @@        -> (state -> m output) -- ^ close        -> Sink input m output sinkIO alloc cleanup push0 close0 = NeedInput-    (\input -> PipeM (do+    (\input -> PipeM $ do         (key, state) <- allocate alloc cleanup-        push key state input) (FinalizeM $ do-            (key, state) <- allocate alloc cleanup-            close key state))-    (do+        push key state input)+    (\() -> do         (key, state) <- lift $ allocate alloc cleanup         lift $ close key state)   where@@ -78,12 +71,10 @@         case res of             IODone a b -> do                 release key-                return $ Done a b+                return $ maybe id (flip Leftover) a $ Done b             IOProcessing -> return $ NeedInput-                (\i ->-                    let mpipe = push key state i-                     in PipeM mpipe (lift mpipe >>= pipeClose))-                (lift $ close key state)+                (PipeM . push key state)+                (\() -> lift $ close key state)     close key state = do         res <- close0 state         release key

@@ -1,7 +1,3 @@-{-# 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.@@ -34,13 +30,13 @@ sourceState state0 pull0 =     src state0   where-    src state = PipeM (pull state) (return ())+    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 Nothing ()+            StateClosed -> Done ()  -- | The return value when pulling in the @sourceIO@ function. Either indicates -- no more data, or the next value.@@ -59,16 +55,16 @@ sourceIO alloc cleanup pull0 =     PipeM (do         (key, state) <- allocate alloc cleanup-        pull key state) (return ())+        pull key state)   where-    src key state = PipeM (pull key state) (release key)+    src key state = PipeM (pull key state)      pull key state = do         res <- pull0 state         case res of             IOClosed -> do                 release key-                return $ Done Nothing ()+                return $ Done ()             IOOpen val -> return $ HaveOutput (src key state) (release key) val  -- | A combination of 'sourceIO' and 'sourceState'.@@ -82,16 +78,16 @@ sourceStateIO alloc cleanup pull0 =     PipeM (do         (key, state) <- allocate alloc cleanup-        pull key state) (return ())+        pull key state)   where-    src key state = PipeM (pull key state) (release key)+    src key state = PipeM (pull key state)      pull key state = do         res <- pull0 state         case res of             StateClosed -> do                 release key-                return $ Done Nothing ()+                return $ Done ()             StateOpen state' val -> return $ HaveOutput (src key state') (release key) val  -- FIXME transPipe

@@ -20,7 +20,6 @@ import qualified Data.ByteString as S import qualified Data.ByteString.Unsafe as BU import Prelude hiding (read)-import Data.Conduit.Util.Source (SourceIOResult (..))  #include <fcntl.h> @@ -49,13 +48,13 @@         then throwErrno $ "Could not open file: " ++ fp         else return $ FD h -read :: FD -> IO (SourceIOResult S.ByteString)+read :: FD -> IO (Maybe S.ByteString) read fd = do     cstr <- mallocBytes 4096     len <- c_read fd cstr 4096     if len == 0-        then free cstr >> return IOClosed-        else fmap IOOpen $ BU.unsafePackCStringFinalizer+        then free cstr >> return Nothing+        else fmap Just $ BU.unsafePackCStringFinalizer                 cstr                 (fromIntegral len)                 (free cstr)

@@ -8,7 +8,11 @@  import Foreign.C.String (CString) import Foreign.Marshal.Alloc (mallocBytes, free)+#if __GLASGOW_HASKELL__ >= 704+import Foreign.C.Types (CInt (..))+#else import Foreign.C.Types (CInt)+#endif import Foreign.C.Error (throwErrno) import Foreign.Ptr (Ptr) import Data.Bits (Bits, (.|.))@@ -18,7 +22,6 @@ import Data.Text.Encoding (encodeUtf16LE) import Data.Word (Word8) import Prelude hiding (read)-import Data.Conduit (SourceIOResult (..))  #include <fcntl.h> #include <Share.h>@@ -74,16 +77,16 @@         then throwErrno $ "Could not open file: " ++ fp         else return $ FD h -read :: FD -> IO (SourceIOResult S.ByteString)+read :: FD -> IO (Maybe S.ByteString) read fd = do     cstr <- mallocBytes 4096     len <- c_read fd cstr 4096     if len == 0         then do             free cstr-            return IOClosed+            return Nothing         else do-            fmap IOOpen $ BU.unsafePackCStringFinalizer+            fmap Just $ BU.unsafePackCStringFinalizer                 cstr                 (fromIntegral len)                 (free cstr)

@@ -1,11 +1,13 @@ Name:                conduit-Version:             0.4.2+Version:             0.5.0 Synopsis:            Streaming data processing library. Description:-	Conduits are an approach to the streaming data problem. It is meant as an alternative to enumerators\/iterators, hoping to address the same issues with different trade-offs based on real-world experience with enumerators. For more information, see <http://www.yesodweb.com/book/conduits>.+    @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:     .+    [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.     .     [0.3] ResourceT has been greatly simplified, specialized for IO, and moved into a separate package. Instead of hard-coding ResourceT into the conduit datatypes, they can now live around any monad. The Conduit datatype has been enhanced to better allow generation of streaming output. The SourceResult, SinkResult, and ConduitResult datatypes have been removed entirely.@@ -45,6 +47,7 @@                        Data.Conduit.List                        Data.Conduit.Lazy                        Data.Conduit.Internal+                       Data.Conduit.Util   Other-modules:       Data.Conduit.Util.Source                        Data.Conduit.Util.Sink                        Data.Conduit.Util.Conduit@@ -76,6 +79,7 @@                    , transformers                    , text                    , resourcet+                   , void     ghc-options:     -Wall  source-repository head

@@ -6,6 +6,8 @@ import Test.HUnit  import qualified Data.Conduit as C+import qualified Data.Conduit.Util as C+import qualified Data.Conduit.Internal as CI import qualified Data.Conduit.List as CL import qualified Data.Conduit.Lazy as CLazy import qualified Data.Conduit.Binary as CB@@ -28,9 +30,11 @@ import Control.Monad.IO.Class (liftIO) import Control.Applicative (pure, (<$>), (<*>)) import Data.Functor.Identity (runIdentity)+import Control.Monad (forever)+import Data.Void (Void)  main :: IO ()-main = hspecX $ do+main = hspec $ do     describe "data loss rules" $ do         it "consumes the source to quickly" $ do             x <- runResourceT $ CL.sourceList [1..10 :: Int] C.$$ do@@ -133,18 +137,18 @@      describe "zipping" $ do         it "zipping two small lists" $ do-            res <- runResourceT $ CL.zip (CL.sourceList [1..10]) (CL.sourceList [11..12]) C.$$ CL.consume+            res <- runResourceT $ C.zip (CL.sourceList [1..10]) (CL.sourceList [11..12]) C.$$ CL.consume             res @=? zip [1..10 :: Int] [11..12 :: Int]      describe "zipping sinks" $ do         it "take all" $ do-            res <- runResourceT $ CL.sourceList [1..10] C.$$ CL.zipSinks CL.consume CL.consume+            res <- runResourceT $ CL.sourceList [1..10] C.$$ C.zipSinks CL.consume CL.consume             res @=? ([1..10 :: Int], [1..10 :: Int])         it "take fewer on left" $ do-            res <- runResourceT $ CL.sourceList [1..10] C.$$ CL.zipSinks (CL.take 4) CL.consume+            res <- runResourceT $ CL.sourceList [1..10] C.$$ C.zipSinks (CL.take 4) CL.consume             res @=? ([1..4 :: Int], [1..10 :: Int])         it "take fewer on right" $ do-            res <- runResourceT $ CL.sourceList [1..10] C.$$ CL.zipSinks CL.consume (CL.take 4)+            res <- runResourceT $ CL.sourceList [1..10] C.$$ C.zipSinks CL.consume (CL.take 4)             res @=? ([1..10 :: Int], [1..4 :: Int])      describe "Monad instance for Sink" $ do@@ -165,8 +169,8 @@             (x, y, z) <- runResourceT $ do                 let src1 = CL.sourceList [1..10 :: Int]                 (src2, x) <- src1 C.$$+ CL.take 5-                (src3, y) <- src2 C.$$+ CL.fold (+) 0-                z <- src3 C.$$ CL.consume+                (src3, y) <- src2 C.$$++ CL.fold (+) 0+                z <- src3 C.$$+- CL.consume                 return (x, y, z)             x @?= [1..5] :: IO ()             y @?= sum [6..10]@@ -219,7 +223,7 @@             (x, y) <- runResourceT $ do                 let src1 = CL.sourceList [1..10 :: Int]                 (src2, x) <- src1 C.$= CL.isolate 5 C.$$+ CL.consume-                y <- src2 C.$$ CL.consume+                y <- src2 C.$$+- CL.consume                 return (x, y)             x @?= [1..5]             y @?= []@@ -237,7 +241,7 @@             (x, y) <- runResourceT $ do                 let src1 = CL.sourceList [1..10 :: Int]                 (src2, x) <- src1 C.$$+ CL.isolate 5 C.=$ CL.consume-                y <- src2 C.$$ CL.consume+                y <- src2 C.$$+- CL.consume                 return (x, y)             x @?= [1..5]             y @?= [6..10]@@ -274,28 +278,26 @@      describe "sequence" $ do         it "simple sink" $ do-            let sumSink :: C.MonadResource m => C.Sink Int m Int-                sumSink = do+            let sumSink = do                     ma <- CL.head                     case ma of                         Nothing -> return 0                         Just a  -> (+a) . fromMaybe 0 <$> CL.head -            res <- runResourceT $ CL.sourceList [1..11]-                             C.$= C.sequence sumSink+            res <- runResourceT $ CL.sourceList [1..11 :: Int]+                             C.$= CL.sequence sumSink                              C.$$ CL.consume             res @?= [3, 7, 11, 15, 19, 11]          it "sink with unpull behaviour" $ do-            let sumSink :: C.MonadResource m => C.Sink Int m Int-                sumSink = do+            let sumSink = do                     ma <- CL.head                     case ma of                         Nothing -> return 0                         Just a  -> (+a) . fromMaybe 0 <$> CL.peek -            res <- runResourceT $ CL.sourceList [1..11]-                             C.$= C.sequence sumSink+            res <- runResourceT $ CL.sourceList [1..11 :: Int]+                             C.$= CL.sequence sumSink                              C.$$ CL.consume             res @?= [3, 5, 7, 9, 11, 13, 15, 17, 19, 21, 11] @@ -392,7 +394,7 @@             x <- runResourceT $ do                 let src1 = CL.sourceList [1..10 :: Int]                 (src2, ()) <- src1 C.$$+ CL.drop 5-                src2 C.$$ CL.fold (+) 0+                src2 C.$$+- CL.fold (+) 0             x @?= sum [6..10]      describe "operators" $ do@@ -400,10 +402,10 @@             runIdentity             (    CL.sourceList [1..10 :: Int]               C.$$ CL.map (+ 1)-             C.=$= CL.map (subtract 1)-             C.=$= CL.map (* 2)-             C.=$= CL.map (`div` 2)-             C.=$= CL.fold (+) 0+             C.=$  CL.map (subtract 1)+             C.=$  CL.mapM (return . (* 2))+             C.=$  CL.map (`div` 2)+             C.=$  CL.fold (+) 0             ) @?= sum [1..10]         it "only use =$" $             runIdentity@@ -417,16 +419,16 @@         it "chain" $ do             x <-      CL.sourceList [1..10 :: Int]                 C.$=  CL.map (+ 1)-                C.=$= CL.map (+ 1)+                C.$= CL.map (+ 1)                 C.$=  CL.map (+ 1)-                C.=$= CL.map (subtract 3)-                C.=$= CL.map (* 2)+                C.$= CL.map (subtract 3)+                C.$= CL.map (* 2)                 C.$$  CL.map (`div` 2)-                C.=$= CL.map (+ 1)                 C.=$  CL.map (+ 1)-                C.=$= CL.map (+ 1)+                C.=$  CL.map (+ 1)+                C.=$  CL.map (+ 1)                 C.=$  CL.map (subtract 3)-                C.=$= CL.fold (+) 0+                C.=$  CL.fold (+) 0             x @?= sum [1..10]  @@ -455,6 +457,17 @@              in runIdentity $ do                 bss2 <- CL.sourceList bss C.$$ CB.takeWhile (>= 5) C.=$ CL.consume                 return $ L.fromChunks bss2 == L.takeWhile (>= 5) (L.fromChunks bss)+        prop "leftovers present" $ \bss' ->+            let bss = map S.pack bss'+             in runIdentity $ do+                result <- CL.sourceList bss C.$$ do+                    x <- CB.takeWhile (>= 5) C.=$ CL.consume+                    y <- CL.consume+                    return (S.concat x, S.concat y)+                let expected = S.span (>= 5) $ S.concat bss+                if result == expected+                    then return True+                    else error $ show (S.concat bss, result, expected)      describe "binary dropWhile" $ do         prop "works" $ \bss' ->@@ -510,5 +523,150 @@             res <- src C.$$ CB.lines C.=$ CL.consume             return $ S8.lines bs == res -it' :: String -> IO () -> Specs+    describe "termination" $ do+        it "terminates early" $ do+            let src = forever $ CI.yield ()+            x <- src C.$$ CL.head+            x @?= Just ()+        it "bracket" $ do+            ref <- I.newIORef (0 :: Int)+            let src = CI.bracketP+                    (I.modifyIORef ref (+ 1))+                    (\() -> I.modifyIORef ref (+ 2))+                    (\() -> forever $ CI.yield (1 :: Int))+            val <- C.runResourceT $ src C.$$ CL.isolate 10 C.=$ CL.fold (+) 0+            val @?= 10+            i <- I.readIORef ref+            i @?= 3+        it "bracket skipped if not needed" $ do+            ref <- I.newIORef (0 :: Int)+            let src = CI.bracketP+                    (I.modifyIORef ref (+ 1))+                    (\() -> I.modifyIORef ref (+ 2))+                    (\() -> forever $ CI.yield (1 :: Int))+                src' = CL.sourceList $ repeat 1+            val <- C.runResourceT $ (src' >> src) C.$$ CL.isolate 10 C.=$ CL.fold (+) 0+            val @?= 10+            i <- I.readIORef ref+            i @?= 0+        it "bracket + toPipe" $ do+            ref <- I.newIORef (0 :: Int)+            let src = CI.bracketP+                    (I.modifyIORef ref (+ 1))+                    (\() -> I.modifyIORef ref (+ 2))+                    (\() -> forever $ CI.yield (1 :: Int))+            val <- C.runResourceT $ src C.$$ CL.isolate 10 C.=$ CL.fold (+) 0+            val @?= 10+            i <- I.readIORef ref+            i @?= 3+        it "bracket skipped if not needed" $ do+            ref <- I.newIORef (0 :: Int)+            let src = CI.bracketP+                    (I.modifyIORef ref (+ 1))+                    (\() -> I.modifyIORef ref (+ 2))+                    (\() -> forever $ CI.yield (1 :: Int))+                src' = CL.sourceList $ repeat 1+            val <- C.runResourceT $ (src' >> src) C.$$ CL.isolate 10 C.=$ CL.fold (+) 0+            val @?= 10+            i <- I.readIORef ref+            i @?= 0++    describe "invariant violations" $ do+        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++            _ <- CI.yield 1 C.$$ adder+            x <- I.readIORef ref+            x @?= [1 :: Int]+            I.writeIORef ref []++            _ <- CI.yield 1 C.$$ (residue 2 >> residue 3) >> adder+            y <- I.readIORef ref+            y @?= [1, 2, 3]+            I.writeIORef ref []++            _ <- CI.yield 1 C.$$ residue 2 >> (residue 3 >> adder)+            z <- I.readIORef ref+            z @?= [1, 2, 3]+            I.writeIORef ref []++    describe "sane yield/await'" $ do+        it' "yield terminates" $ do+            let is = [1..10] ++ undefined+                src [] = return ()+                src (x:xs) = CI.yield x >> src xs+            x <- src is C.$$ CL.take 10+            x @?= [1..10 :: Int]+        it' "yield terminates (2)" $ do+            let is = [1..10] ++ undefined+            x <- mapM_ CI.yield is C.$$ CL.take 10+            x @?= [1..10 :: Int]+        it' "yieldOr finalizer called" $ do+            iref <- I.newIORef (0 :: Int)+            let src = mapM_ (\i -> C.yieldOr i $ I.writeIORef iref i) [1..]+            src C.$$ CL.isolate 10 C.=$ CL.sinkNull+            x <- I.readIORef iref+            x @?= 10++    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+            (x, y) @?= (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 @?= 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 @?= 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 <- CL.isolate 5 C.=$ CL.fold (+) 0+                    y <- CL.peek+                    return (x :: Int, y :: Maybe Int)) :: C.Sink Int IO (Int, Maybe Int))+                z <- CL.consume+                return (x, y, concat z)++            xyz @?= (sum [1..5], Just 6, "678910")++    describe "left/right identity" $ do+        it' "left identity" $ do+            x <- CL.sourceList [1..10 :: Int] C.$$ CI.idP C.=$ CL.fold (+) 0+            y <- CL.sourceList [1..10 :: Int] C.$$ CL.fold (+) 0+            x @?= 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 @?= y++    describe "generalizing" $ do+        it' "works" $ do+            x <-     C.runPipe+                   $ C.sourceToPipe  (CL.sourceList [1..10 :: Int])+               C.>+> C.conduitToPipe (CL.map (+ 1))+               C.>+> C.sinkToPipe    (CL.fold (+) 0)+            x @?= 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+            res @?= (True, sum [1..10])++    describe "iterate" $ do+        it' "works" $ do+            res <- CL.iterate (+ 1) (1 :: Int) C.$$ CL.isolate 10 C.=$ CL.fold (+) 0+            res @?= sum [1..10]++it' :: String -> IO () -> Spec it' = it