conduit 1.0.16 → 1.0.17
raw patch · 4 files changed
+254/−2 lines, 4 files
Files
- Data/Conduit.hs +94/−0
- Data/Conduit/Internal.hs +139/−0
- conduit.cabal +1/−1
- test/main.hs +20/−1
Data/Conduit.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE RankNTypes #-}+{-# LANGUAGE DeriveFunctor #-} -- | If this is your first time with conduit, you should probably start with -- the tutorial: -- <https://haskell.fpcomplete.com/user/snoyberg/library-documentation/conduit-overview>.@@ -51,6 +52,17 @@ , ($=+) , unwrapResumable + -- ** For @Conduit@s+ , ResumableConduit+ , (=$$+)+ , (=$$++)+ , (=$$+-)+ , unwrapResumableConduit++ -- * Fusion with leftovers+ , fuseLeftovers+ , fuseReturnLeftovers+ -- * Flushing , Flush (..) @@ -63,6 +75,10 @@ , ZipSink (..) , sequenceSinks + -- ** ZipConduit+ , ZipConduit (..)+ , sequenceConduits+ -- * Convenience re-exports , ResourceT , MonadResource@@ -77,6 +93,7 @@ ) where import Control.Monad.Trans.Resource+import Control.Monad.Trans.Class (lift) import Data.Conduit.Internal hiding (await, awaitForever, yield, yieldOr, leftover, bracketP, addCleanup, transPipe, mapOutput, mapOutputMaybe, mapInput) import qualified Data.Conduit.Internal as CI import Control.Monad.Morph (hoist)@@ -383,3 +400,80 @@ -- Since 1.0.13 sequenceSinks :: (Traversable f, Monad m) => f (Sink i m r) -> Sink i m (f r) sequenceSinks = getZipSink . sequenceA . fmap ZipSink++-- | The connect-and-resume operator. This does not close the @Conduit@, but+-- instead returns it to be used again. This allows a @Conduit@ to be used+-- incrementally in a large program, without forcing the entire program to live+-- in the @Sink@ monad.+--+-- Leftover data returned from the @Sink@ will be discarded.+--+-- Mnemonic: connect + do more.+--+-- Since 1.0.17+(=$$+) :: Monad m => Conduit a m b -> Sink b m r -> Sink a m (ResumableConduit a m b, r)+(=$$+) conduit = connectResumeConduit (ResumableConduit conduit (return ()))+{-# INLINE (=$$+) #-}++-- | Continue processing after usage of '=$$+'. Connect a 'ResumableConduit' to+-- a sink and return the output of the sink together with a new+-- 'ResumableConduit'.+--+-- Since 1.0.17+(=$$++) :: Monad m => ResumableConduit i m o -> Sink o m r -> Sink i m (ResumableConduit i m o, r)+(=$$++) = connectResumeConduit+{-# INLINE (=$$++) #-}++-- | Complete processing of a 'ResumableConduit'. This will run the finalizer+-- associated with the @ResumableConduit@. In order to guarantee process+-- resource finalization, you /must/ use this operator after using '=$$+' and+-- '=$$++'.+--+-- Since 1.0.17+(=$$+-) :: Monad m => ResumableConduit i m o -> Sink o m r -> Sink i m r+rsrc =$$+- sink = do+ (ResumableConduit _ final, res) <- connectResumeConduit rsrc sink+ lift final+ return res+{-# INLINE (=$$+-) #-}+++infixr 0 =$$++infixr 0 =$$+++infixr 0 =$$+-++-- | Provides an alternative @Applicative@ instance for @ConduitM@. In this instance,+-- every incoming value is provided to all @ConduitM@s, and output is coalesced together.+-- Leftovers from individual @ConduitM@s will be used within that component, and then discarded+-- at the end of their computation. Output and finalizers will both be handled in a left-biased manner.+--+-- As an example, take the following program:+--+-- @+-- main :: IO ()+-- main = do+-- let src = mapM_ yield [1..3 :: Int]+-- conduit1 = CL.map (+1)+-- conduit2 = CL.concatMap (replicate 2)+-- conduit = getZipConduit $ ZipConduit conduit1 <* ZipConduit conduit2+-- sink = CL.mapM_ print+-- src $$ conduit =$ sink+-- @+--+-- It will produce the output: 2, 1, 1, 3, 2, 2, 4, 3, 3+--+-- Since 1.0.17+newtype ZipConduit i o m r = ZipConduit { getZipConduit :: ConduitM i o m r }+ deriving Functor+instance Monad m => Applicative (ZipConduit i o m) where+ pure = ZipConduit . pure+ ZipConduit left <*> ZipConduit right = ZipConduit (zipConduitApp left right)++-- | Provide identical input to all of the @Conduit@s and combine their outputs+-- into a single stream.+--+-- Implemented on top of @ZipConduit@, see that data type for more details.+--+-- Since 1.0.17+sequenceConduits :: (Traversable f, Monad m) => f (ConduitM i o m r) -> ConduitM i o m (f r)+sequenceConduits = getZipConduit . sequenceA . fmap ZipConduit
Data/Conduit/Internal.hs view
@@ -18,6 +18,7 @@ , Consumer , Conduit , ResumableSource (..)+ , ResumableConduit (..) -- * Primitives , await , awaitE@@ -33,10 +34,13 @@ , pipe , pipeL , connectResume+ , connectResumeConduit , runPipe , injectLeftovers , (>+>) , (<+<)+ , fuseLeftovers+ , fuseReturnLeftovers -- * Generalizing , sourceToPipe , sinkToPipe@@ -58,10 +62,12 @@ , sourceList , withUpstream , unwrapResumable+ , unwrapResumableConduit , Data.Conduit.Internal.enumFromTo , zipSinks , zipSources , zipSourcesApp+ , zipConduitApp ) where import Control.Applicative (Applicative (..))@@ -913,3 +919,136 @@ go (HaveOutput srcx closex x) (HaveOutput srcy closey y) = HaveOutput (go srcx srcy) (closex >> closey) (x y) go (NeedInput _ c) right = go (c ()) right go left (NeedInput _ c) = go left (c ())++-- |+--+-- Since 1.0.17+zipConduitApp+ :: Monad m+ => ConduitM i o m (x -> y)+ -> ConduitM i o m x+ -> ConduitM i o m y+zipConduitApp (ConduitM left0) (ConduitM right0) =+ ConduitM $ go (return ()) (return ()) (injectLeftovers left0) (injectLeftovers right0)+ where+ go _ _ (Done f) (Done x) = Done (f x)+ go _ finalY (HaveOutput x finalX o) y = HaveOutput+ (go finalX finalY x y)+ (finalX >> finalY)+ o+ go finalX _ x (HaveOutput y finalY o) = HaveOutput+ (go finalX finalY x y)+ (finalX >> finalY)+ o+ go _ _ (Leftover _ i) _ = absurd i+ go _ _ _ (Leftover _ i) = absurd i+ go finalX finalY (PipeM mx) y = PipeM (flip (go finalX finalY) y `liftM` mx)+ go finalX finalY x (PipeM my) = PipeM (go finalX finalY x `liftM` my)+ go finalX finalY (NeedInput px cx) (NeedInput py cy) = NeedInput+ (\i -> go finalX finalY (px i) (py i))+ (\u -> go finalX finalY (cx u) (cy u))+ go finalX finalY (NeedInput px cx) (Done y) = NeedInput+ (\i -> go finalX finalY (px i) (Done y))+ (\u -> go finalX finalY (cx u) (Done y))+ go finalX finalY (Done x) (NeedInput py cy) = NeedInput+ (\i -> go finalX finalY (Done x) (py i))+ (\u -> go finalX finalY (Done x) (cy u))++-- | Same as normal fusion (e.g. @=$=@), except instead of discarding leftovers+-- from the downstream component, return them.+--+-- Since 1.0.17+fuseReturnLeftovers :: Monad m+ => ConduitM a b m ()+ -> ConduitM b c m r+ -> ConduitM a c m (r, [b])+fuseReturnLeftovers (ConduitM left0) (ConduitM right0) =+ ConduitM $ goRight (return ()) [] left0 right0+ where+ goRight final bs left right =+ case right of+ HaveOutput p c o -> HaveOutput (recurse p) (c >> final) o+ NeedInput rp rc ->+ case bs of+ [] -> goLeft rp rc final left+ b:bs' -> goRight final bs' left (rp b)+ Done r2 -> PipeM (final >> return (Done (r2, bs)))+ PipeM mp -> PipeM (liftM recurse mp)+ Leftover p b -> goRight final (b:bs) left p+ where+ recurse = goRight final bs left++ goLeft rp rc final left =+ case left of+ HaveOutput left' final' o -> goRight final' [] left' (rp o)+ NeedInput left' lc -> NeedInput (recurse . left') (recurse . lc)+ Done r1 -> goRight (return ()) [] (Done r1) (rc r1)+ PipeM mp -> PipeM (liftM recurse mp)+ Leftover left' i -> Leftover (recurse left') i+ where+ recurse = goLeft rp rc final++-- | Similar to @fuseReturnLeftovers@, but use the provided function to convert+-- downstream leftovers to upstream leftovers.+--+-- Since 1.0.17+fuseLeftovers+ :: Monad m+ => ([b] -> [a])+ -> ConduitM a b m ()+ -> ConduitM b c m r+ -> ConduitM a c m r+fuseLeftovers f left right = do+ (r, bs) <- fuseReturnLeftovers left right+ ConduitM $ mapM_ leftover $ reverse $ f bs+ return r++-- | A generalization of 'ResumableSource'. Allows to resume an arbitrary+-- conduit, keeping its state and using it later (or finalizing it).+--+-- Since 1.0.17+data ResumableConduit i m o =+ ResumableConduit (Conduit i m o) (m ())++-- | Connect a 'ResumableConduit' to a sink and return the output of the sink+-- together with a new 'ResumableConduit'.+--+-- Since 1.0.17+connectResumeConduit+ :: Monad m+ => ResumableConduit i m o+ -> Sink o m r+ -> Sink i m (ResumableConduit i m o, r)+connectResumeConduit (ResumableConduit (ConduitM left0) leftFinal0) (ConduitM right0) =+ ConduitM $ goRight leftFinal0 left0 right0+ where+ goRight leftFinal left right =+ case right of+ HaveOutput _ _ o -> absurd o+ NeedInput rp rc -> goLeft rp rc leftFinal left+ Done r2 -> Done (ResumableConduit (ConduitM left) leftFinal, r2)+ PipeM mp -> PipeM (liftM (goRight leftFinal left) mp)+ Leftover p i -> goRight leftFinal (HaveOutput left leftFinal i) p++ goLeft rp rc leftFinal left =+ case left of+ HaveOutput left' leftFinal' o -> goRight leftFinal' left' (rp o)+ NeedInput left' lc -> NeedInput (recurse . left') (recurse . lc)+ Done () -> goRight (return ()) (Done ()) (rc ())+ PipeM mp -> PipeM (liftM recurse mp)+ Leftover left' i -> Leftover (recurse left') i -- recurse p+ where+ recurse = goLeft rp rc leftFinal++-- | Unwraps a @ResumableConduit@ into a @Conduit@ and a finalizer.+--+-- Since 'unwrapResumable' for more information.+--+-- Since 1.0.17+unwrapResumableConduit :: MonadIO m => ResumableConduit i m o -> m (Conduit i m o, m ())+unwrapResumableConduit (ResumableConduit src final) = do+ ref <- liftIO $ I.newIORef True+ let final' = do+ x <- liftIO $ I.readIORef ref+ when x final+ return (liftIO (I.writeIORef ref False) >> src, final')
conduit.cabal view
@@ -1,5 +1,5 @@ Name: conduit-Version: 1.0.16+Version: 1.0.17 Synopsis: Streaming data processing library. Description: @conduit@ is a solution to the streaming data problem, allowing for production, transformation, and consumption of streams of data in constant memory. It is an alternative to lazy I\/O which guarantees deterministic resource handling, and fits in the same general solution space as @enumerator@\/@iteratee@ and @pipes@. For a tutorial, please visit <https://haskell.fpcomplete.com/user/snoyberg/library-documentation/conduit-overview>.
test/main.hs view
@@ -14,7 +14,7 @@ import qualified Data.Conduit.Binary as CB import qualified Data.Conduit.Text as CT import Data.Conduit (runResourceT)-import Data.Maybe (fromMaybe,catMaybes)+import Data.Maybe (fromMaybe,catMaybes,fromJust) import qualified Data.List as DL import Control.Monad.ST (runST) import Data.Monoid@@ -41,6 +41,8 @@ import Control.Monad.Error (catchError, throwError, Error) import qualified Data.Map as Map import Control.Arrow (first)+import qualified Data.Conduit.ExtraSpec as ES+import qualified Data.Conduit.Extra.ZipConduitSpec as ZipConduit (@=?) :: (Eq a, Show a) => a -> a -> IO () (@=?) = flip shouldBe@@ -1243,6 +1245,23 @@ , Map.fromList [(1, 2), (2, 2), (3, 2)] , Map.fromList [(1, 3), (2, 1), (3, 2)] ]+ describe "zipSink" $ do+ it "zip equal-sized" $ do+ x <- runResourceT $+ CL.sourceList [1..100] C.$$+ C.sequenceSinks [ CL.fold (+) 0,+ (`mod` 101) <$> CL.fold (*) 1 ]+ x `shouldBe` [5050, 100 :: Integer]++ it "zip distinct sizes" $ do+ let sink = C.getZipSink $+ (*) <$> C.ZipSink (CL.fold (+) 0)+ <*> C.ZipSink (Data.Maybe.fromJust <$> C.await)+ x <- C.runResourceT $ CL.sourceList [100,99..1] C.$$ sink+ x `shouldBe` (505000 :: Integer)++ ES.spec+ ZipConduit.spec it' :: String -> IO () -> Spec it' = it