diff --git a/Data/Conduit.hs b/Data/Conduit.hs --- a/Data/Conduit.hs +++ b/Data/Conduit.hs @@ -53,6 +53,15 @@ -- * Flushing , Flush (..) + -- * Newtype wrappers + -- ** ZipSource + , ZipSource (..) + , sequenceSources + + -- ** ZipSink + , ZipSink (..) + , sequenceSinks + -- * Convenience re-exports , ResourceT , MonadResource @@ -70,6 +79,9 @@ 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) +import Control.Monad (liftM, forever) +import Control.Applicative (Applicative (..)) +import Data.Traversable (Traversable (..)) -- Define fixity of all our operators infixr 0 $$ @@ -289,3 +301,59 @@ instance Functor Flush where fmap _ Flush = Flush fmap f (Chunk a) = Chunk (f a) + +-- | A wrapper for defining an 'Applicative' instance for 'Sink's which allows +-- to combine sinks together, generalizing 'zipSources'. A combined sources +-- will take input yielded from each of its @Source@s until any of them stop +-- producing output. +-- +-- Since 1.0.13 +newtype ZipSource m o = ZipSource { getZipSource :: Source m o } + +instance Monad m => Functor (ZipSource m) where + fmap f = ZipSource . mapOutput f . getZipSource +instance Monad m => Applicative (ZipSource m) where + pure = ZipSource . forever . yield + (ZipSource f) <*> (ZipSource x) = ZipSource $ zipSourcesApp f x + +-- | Coalesce all values yielding by all of the @Source@s. +-- +-- Implemented on top of @ZipSource@, see that data type for more details. +-- +-- Since 1.0.13 +sequenceSources :: (Traversable f, Monad m) => f (Source m o) -> Source m (f o) +sequenceSources = getZipSource . sequenceA . fmap ZipSource + +-- | A wrapper for defining an 'Applicative' instance for 'Sink's which allows +-- to combine sinks together, generalizing 'zipSinks'. A combined sink +-- distributes the input to all its participants and when all finish, produces +-- the result. This allows to define functions like +-- +-- @ +-- sequenceSinks :: (Monad m) +-- => [Sink i m r] -> Sink i m [r] +-- sequenceSinks = getZipSink . sequenceA . fmap ZipSink +-- @ +-- +-- Note that the standard 'Applicative' instance for conduits works +-- differently. It feeds one sink with input until it finishes, then switches +-- to another, etc., and at the end combines their results. +-- +-- Since 1.0.13 +newtype ZipSink i m r = ZipSink { getZipSink :: Sink i m r } + +instance Monad m => Functor (ZipSink i m) where + fmap f (ZipSink x) = ZipSink (liftM f x) +instance Monad m => Applicative (ZipSink i m) where + pure = ZipSink . return + (ZipSink f) <*> (ZipSink x) = + ZipSink $ liftM (uncurry ($)) $ zipSinks f x + +-- | Send incoming values to all of the @Sink@ providing, and ultimately +-- coalesce together all return values. +-- +-- Implemented on top of @ZipSink@, see that data type for more details. +-- +-- Since 1.0.13 +sequenceSinks :: (Traversable f, Monad m) => f (Sink i m r) -> Sink i m (f r) +sequenceSinks = getZipSink . sequenceA . fmap ZipSink diff --git a/Data/Conduit/Internal.hs b/Data/Conduit/Internal.hs --- a/Data/Conduit/Internal.hs +++ b/Data/Conduit/Internal.hs @@ -59,11 +59,14 @@ , withUpstream , unwrapResumable , Data.Conduit.Internal.enumFromTo + , zipSinks + , zipSources + , zipSourcesApp ) where import Control.Applicative (Applicative (..)) import Control.Exception.Lifted as E (Exception, catch) -import Control.Monad ((>=>), liftM, ap, when) +import Control.Monad ((>=>), liftM, ap, when, liftM2) import Control.Monad.Error.Class(MonadError(..)) import Control.Monad.Reader.Class(MonadReader(..)) import Control.Monad.RWS.Class(MonadRWS()) @@ -295,6 +298,10 @@ -- Since 0.5.0 data ResumableSource m o = ResumableSource (Source m o) (m ()) +-- | Since 1.0.13 +instance MFunctor ResumableSource where + hoist nat (ResumableSource src m) = ResumableSource (hoist nat src) (nat m) + -- | Wait for a single input value from upstream. -- -- Since 0.5.0 @@ -838,3 +845,71 @@ -> ConduitM i o m (Either e r) tryC = ConduitM . tryP . unConduitM {-# INLINE tryC #-} + +-- | 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 (ConduitM x0) (ConduitM y0) = + ConduitM $ injectLeftovers x0 >< injectLeftovers y0 + where + (><) :: Monad m => Pipe Void i Void () m r1 -> Pipe Void i Void () m r2 -> Pipe l i o () m (r1, r2) + + Leftover _ i >< _ = absurd i + _ >< Leftover _ i = absurd i + 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) + +-- | Combines two sources. The new source will stop producing once either +-- source has been exhausted. +-- +-- Since 1.0.13 +zipSources :: Monad m => Source m a -> Source m b -> Source m (a, b) +zipSources (ConduitM left0) (ConduitM right0) = + ConduitM $ go left0 right0 + where + go (Leftover left ()) right = go left right + go left (Leftover right ()) = go left right + go (Done ()) (Done ()) = Done () + go (Done ()) (HaveOutput _ close _) = PipeM (close >> return (Done ())) + go (HaveOutput _ close _) (Done ()) = PipeM (close >> return (Done ())) + go (Done ()) (PipeM _) = Done () + go (PipeM _) (Done ()) = Done () + go (PipeM mx) (PipeM my) = PipeM (liftM2 go mx my) + go (PipeM mx) y@HaveOutput{} = PipeM (liftM (\x -> go x y) mx) + go x@HaveOutput{} (PipeM my) = PipeM (liftM (go x) my) + go (HaveOutput srcx closex x) (HaveOutput srcy closey y) = HaveOutput (go srcx srcy) (closex >> closey) (x, y) + go (NeedInput _ c) right = go (c ()) right + go left (NeedInput _ c) = go left (c ()) + +-- | Combines two sources. The new source will stop producing once either +-- source has been exhausted. +-- +-- Since 1.0.13 +zipSourcesApp :: Monad m => Source m (a -> b) -> Source m a -> Source m b +zipSourcesApp (ConduitM left0) (ConduitM right0) = + ConduitM $ go left0 right0 + where + go (Leftover left ()) right = go left right + go left (Leftover right ()) = go left right + go (Done ()) (Done ()) = Done () + go (Done ()) (HaveOutput _ close _) = PipeM (close >> return (Done ())) + go (HaveOutput _ close _) (Done ()) = PipeM (close >> return (Done ())) + go (Done ()) (PipeM _) = Done () + go (PipeM _) (Done ()) = Done () + go (PipeM mx) (PipeM my) = PipeM (liftM2 go mx my) + go (PipeM mx) y@HaveOutput{} = PipeM (liftM (\x -> go x y) mx) + go x@HaveOutput{} (PipeM my) = PipeM (liftM (go x) my) + go (HaveOutput srcx closex x) (HaveOutput srcy closey y) = HaveOutput (go srcx srcy) (closex >> closey) (x y) + go (NeedInput _ c) right = go (c ()) right + go left (NeedInput _ c) = go left (c ()) diff --git a/Data/Conduit/List.hs b/Data/Conduit/List.hs --- a/Data/Conduit/List.hs +++ b/Data/Conduit/List.hs @@ -66,6 +66,7 @@ , Enum (succ), Eq , maybe , either + , (<=) ) import Data.Monoid (Monoid, mempty, mappend) import qualified Data.Foldable as F @@ -205,7 +206,7 @@ drop = loop where - loop 0 = return () + loop i | i <= 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 @@ -350,7 +351,7 @@ scanl f = loop where - loop s = await >>= F.mapM_ go + loop s = await >>= maybe (return ()) go where go a = case f a s of (s',b) -> yield b >> loop s' @@ -362,7 +363,7 @@ scanlM f = loop where - loop s = await >>= F.mapM_ go + loop s = await >>= maybe (return ()) go where go a = do (s',b) <- lift $ f a s yield b >> loop s' diff --git a/Data/Conduit/Util.hs b/Data/Conduit/Util.hs --- a/Data/Conduit/Util.hs +++ b/Data/Conduit/Util.hs @@ -2,61 +2,22 @@ module Data.Conduit.Util ( -- * Misc zip + , zipSources , zipSinks , passthroughSink ) where import Prelude hiding (zip) -import Control.Monad (liftM, liftM2) -import Data.Conduit.Internal (Pipe (..), Source, Sink, injectLeftovers, ConduitM (..), Conduit, awaitForever, yield, await) -import Data.Void (Void, absurd) +import Data.Conduit.Internal (Pipe (..), Source, Sink, ConduitM (..), Conduit, awaitForever, yield, await, zipSinks, zipSources) +import Data.Void (absurd) import Control.Monad.Trans.Class (lift) --- | Combines two sources. The new source will stop producing once either --- source has been exhausted. +-- | Deprecated synonym for 'zipSources'. -- -- Since 0.3.0 zip :: Monad m => Source m a -> Source m b -> Source m (a, b) -zip (ConduitM left0) (ConduitM right0) = - ConduitM $ go left0 right0 - where - go (Leftover left ()) right = go left right - go left (Leftover right ()) = go left right - go (Done ()) (Done ()) = Done () - go (Done ()) (HaveOutput _ close _) = PipeM (close >> return (Done ())) - go (HaveOutput _ close _) (Done ()) = PipeM (close >> return (Done ())) - go (Done ()) (PipeM _) = Done () - go (PipeM _) (Done ()) = Done () - go (PipeM mx) (PipeM my) = PipeM (liftM2 go mx my) - go (PipeM mx) y@HaveOutput{} = PipeM (liftM (\x -> go x y) mx) - go x@HaveOutput{} (PipeM my) = PipeM (liftM (go x) my) - go (HaveOutput srcx closex x) (HaveOutput srcy closey y) = HaveOutput (go srcx srcy) (closex >> closey) (x, y) - go (NeedInput _ c) right = go (c ()) right - go left (NeedInput _ c) = go left (c ()) - --- | Combines two sinks. The new sink will complete when both input sinks have --- completed. --- --- 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 (ConduitM x0) (ConduitM y0) = - ConduitM $ injectLeftovers x0 >< injectLeftovers y0 - where - (><) :: Monad m => Pipe Void i Void () m r1 -> Pipe Void i Void () m r2 -> Pipe l i o () m (r1, r2) - - Leftover _ i >< _ = absurd i - _ >< Leftover _ i = absurd i - 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) +zip = zipSources +{-# DEPRECATED zip "Use zipSources instead" #-} -- | Turn a @Sink@ into a @Conduit@ in the following way: -- diff --git a/conduit.cabal b/conduit.cabal --- a/conduit.cabal +++ b/conduit.cabal @@ -1,5 +1,5 @@ Name: conduit -Version: 1.0.12 +Version: 1.0.13 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 @@ -194,7 +194,7 @@ describe "zipping" $ do it "zipping two small lists" $ do - res <- runResourceT $ C.zip (CL.sourceList [1..10]) (CL.sourceList [11..12]) C.$$ CL.consume + res <- runResourceT $ C.zipSources (CL.sourceList [1..10]) (CL.sourceList [11..12]) C.$$ CL.consume res @=? zip [1..10 :: Int] [11..12 :: Int] describe "zipping sinks" $ do