diff --git a/Data/Conduit.hs b/Data/Conduit.hs --- a/Data/Conduit.hs +++ b/Data/Conduit.hs @@ -1,4 +1,5 @@ {-# LANGUAGE RankNTypes #-} +{-# LANGUAGE DeriveFunctor #-} -- | If this is your first time with conduit, you should probably start with -- the tutorial: -- . @@ -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 diff --git a/Data/Conduit/Internal.hs b/Data/Conduit/Internal.hs --- a/Data/Conduit/Internal.hs +++ b/Data/Conduit/Internal.hs @@ -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') diff --git a/conduit.cabal b/conduit.cabal --- a/conduit.cabal +++ b/conduit.cabal @@ -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 . diff --git a/test/main.hs b/test/main.hs --- a/test/main.hs +++ b/test/main.hs @@ -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