simple-conduit 0.3.0 → 0.4.0

raw patch · 3 files changed

+600/−266 lines, 3 filesdep +semigroupsPVP ok

version bump matches the API change (PVP)

Dependencies added: semigroups

API changes (from Hackage documentation)

Files

• Conduit/Simple.hs +597/−263
• simple-conduit.cabal +2/−1
• test/bench.hs +1/−2

@@ -2,7 +2,10 @@ {-# LANGUAGE TypeFamilies #-} {-# LANGUAGE TupleSections #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE UndecidableInstances #-}  -- | Please see the project README for more details: --@@ -12,42 +15,198 @@ -- --   https://www.newartisans.com/2014/06/simpler-conduit-library -module Conduit.Simple where+module Conduit.Simple+    ( Source(..), Conduit, Sink+    , sequenceSources+    , ZipSink(..), sequenceSinks+    , source, conduit, conduitWith, sink+    , ($=), (=$), ($$)+    , returnC, abort, skip, awaitForever+    , yieldMany, sourceList+    , unfoldC+    , enumFromToC+    , iterateC+    , repeatC+    , replicateC+    , sourceLazy+    , repeatMC+    , repeatWhileMC+    , replicateMC+    , sourceHandle+    , sourceFile+    , sourceIOHandle+    , stdinC+    , initRepeat+    , initReplicate+    , sourceRandom+    , sourceRandomN+    , sourceRandomGen+    , sourceRandomNGen+    , sourceDirectory+    , sourceDirectoryDeep+    , dropC+    , dropCE+    , dropWhileC+    , dropWhileCE+    , foldC+    , foldCE+    , foldlC+    , foldlCE+    , foldMapC+    , foldMapCE+    , allC+    , allCE+    , anyC+    , anyCE+    , andC+    , andCE+    , orC+    , orCE+    , elemC+    , elemCE+    , notElemC+    , notElemCE+    , sinkLazy+    , sinkList+    , sinkVector+    , sinkVectorN+    , sinkBuilder+    , sinkLazyBuilder+    , sinkNull+    , awaitNonNull+    , headCE+    , lastC+    , lastCE+    , lengthC+    , lengthCE+    , lengthIfC+    , lengthIfCE+    , maximumC+    , maximumCE+    , minimumC+    , minimumCE+    , sumC+    , sumCE+    , productC+    , productCE+    , findC+    , mapM_C+    , mapM_CE+    , foldMC+    , foldMCE+    , foldMapMC+    , foldMapMCE+    , sinkFile+    , sinkHandle+    , sinkIOHandle+    , printC+    , stdoutC+    , stderrC+    , mapC+    , mapCE+    , omapCE+    , concatMapC+    , concatMapCE+    , takeC+    , takeCE+    , takeWhileC+    , takeWhileCE+    , takeExactlyC+    , takeExactlyCE+    , concatC+    , filterC+    , filterCE+    , mapWhileC+    , conduitVector+    , scanlC+    , concatMapAccumC+    , intersperseC+    , encodeBase64C+    , decodeBase64C+    , encodeBase64URLC+    , decodeBase64URLC+    , encodeBase16C+    , decodeBase16C+    , mapMC+    , mapMCE+    , omapMCE+    , concatMapMC+    , filterMC+    , filterMCE+    , iterMC+    , scanlMC+    , concatMapAccumMC+    , encodeUtf8C+    , decodeUtf8C+    , lineC+    , lineAsciiC+    , unlinesC+    , unlinesAsciiC+    , linesUnboundedC_+    , linesUnboundedC+    , linesUnboundedAsciiC+    , zipSinks+    , sourceMaybeMVar+    , sourceMaybeTMVar+    , asyncC+    , fromFoldM+    , toFoldM+    , sourceTChan+    , sourceTQueue+    , sourceTBQueue+    , untilMC+    , whileMC+    ) where -import           Control.Applicative-import           Control.Concurrent.Async.Lifted+import           Control.Applicative (Alternative((<|>), empty),+                                      Applicative((<*>), pure), (<$>))+import           Control.Concurrent (MVar, takeMVar, putMVar, newEmptyMVar)+import           Control.Concurrent.Async.Lifted (Async, withAsync, waitBoth,+                                                  async) import           Control.Concurrent.STM-import           Control.Exception.Lifted-import           Control.Foldl-import           Control.Monad hiding (mapM)-import           Control.Monad.Base-import           Control.Monad.Catch hiding (bracket, catch)-import           Control.Monad.IO.Class-import           Control.Monad.Morph-import           Control.Monad.Primitive-import           Control.Monad.Trans.Control-import           Control.Monad.Trans.Either-import           Data.Bifunctor-import           Data.Builder-import           Data.ByteString hiding (hPut, putStrLn)-import           Data.IOData-{--import           Data.IORef--}-import           Data.MonoTraversable-import           Data.Monoid-import           Data.NonNull as NonNull-import           Data.Sequences as Seq-import           Data.Sequences.Lazy+import           Control.Exception.Lifted (bracket)+import           Control.Foldl (PrimMonad, Vector, FoldM(..))+import           Control.Monad (liftM, MonadPlus(..), ap, (<=<))+import           Control.Monad.Base (MonadBase(..))+import           Control.Monad.Catch (MonadThrow(..), MonadMask, MonadCatch)+import qualified Control.Monad.Catch as Catch+import           Control.Monad.Error.Class (MonadError(..))+import           Control.Monad.IO.Class (MonadIO(..))+import           Control.Monad.Morph (MonadTrans(..), MMonad(..), MFunctor(..))+import           Control.Monad.Primitive (PrimMonad(PrimState))+import           Control.Monad.Reader.Class (MonadReader(..))+import           Control.Monad.State.Class (MonadState(..))+import           Control.Monad.Trans.Control (MonadBaseControl(StM))+import           Control.Monad.Trans.Either (EitherT(..), left)+import           Control.Monad.Writer.Class (MonadWriter(..))+import           Data.Bifunctor (Bifunctor(bimap))+import           Data.Builder (Builder(builderToLazy), ToBuilder(..))+import           Data.ByteString (ByteString)+import           Data.Foldable (Foldable(foldMap))+import           Data.Functor.Identity (Identity(runIdentity))+import           Data.IOData (IOData(hGetChunk, hPut))+import           Data.List (unfoldr)+import           Data.MonoTraversable (MonoTraversable, MonoFunctor, Element,+                                       MonoFoldable(oall, oany, ofoldMap,+                                                    ofoldl', ofoldlM, olength,+                                                    onull))+import           Data.NonNull as NonNull (NonNull, fromNullable)+import           Data.Semigroup (Any(..), All(..), Monoid(..), Semigroup((<>)))+import           Data.Sequences as Seq (OrdSequence, EqSequence(elem, notElem),+                                        SemiSequence(Index), singleton,+                                        IsSequence(break, drop, dropWhile,+                                                   fromList, splitAt))+import           Data.Sequences.Lazy (LazySequence(fromChunks, toChunks)) import qualified Data.Streaming.Filesystem as F-import           Data.Text-import           Data.Textual.Encoding-import           Data.Traversable-import           Data.Word-import           Prelude hiding (mapM)+import           Data.Text (Text)+import           Data.Textual.Encoding (Utf8(encodeUtf8))+import           Data.Traversable (Traversable(sequenceA))+import           Data.Word (Word8) import           System.FilePath ((</>))-import           System.IO-import           System.Random.MWC as MWC+import           System.IO (stdout, stdin, stderr, openFile, hClose,+                            Handle, IOMode(ReadMode, WriteMode))+import           System.Random.MWC as MWC (Gen, Variate(uniform),+                                           createSystemRandom)  -- | The type of Source should recall 'foldM': --@@ -55,53 +214,181 @@ -- Monad m => (a -> b -> m a) -> a -> [b] -> m a -- @ ----- 'EitherT' is used to signal short-circuiting of the pipeline.  If it--- weren't for conduits like 'dropC' and 'takeC', we would not need it most of--- the time.+-- 'EitherT' is used to signal short-circuiting of the pipeline.  And if it+-- weren't for conduits like 'takeC', we wouldn't even need that most of the+-- time.+--+-- Sources form a Monad that behaves a lot like 'ListT'; for example:+--+-- @+-- do line <- sourceFile "foo.txt"+--    liftIO $ putStrLn $ "line: " ++ show line+--    x <- yieldMany [1..10]+--    return (x, line)+-- @+--+-- The above Source yields a series of pairs, proving ten copies of each line+-- from the file plus an index number.+--+-- To skip to the next value in a Source, use the function 'skip' or 'mempty';+-- to abort the whole pipeline, use 'abort' or 'mzero'.  For example:+--+-- @+-- do x <- yieldMany [1..10]+--    if x == 2 || x == 9+--    then return x+--    else if x < 5+--         then skip+--         else abort+-- @+--+-- This outputs the list @[2]@.+--+-- One difference from conduit is that monadic chaining of sources with '>>'+-- results in the values from the first source being used to determine how+-- many values are generated by the next source, just like 'ListT':+--+-- >>> sinkList $ yieldMany [1..3] >> yieldMany [4..6]+-- [4,5,6,4,5,6,4,5,6]+--+-- To achieve the same behavior as conduit, use the Monoid instance for+-- Sources:+--+-- >>> sinkList $ yieldMany [1..3] <> yieldMany [4..6]+-- [1,2,3,4,5,6] newtype Source m a = Source-    { getSource :: forall r. r -> (r -> a -> EitherT r m r) -> EitherT r m r }+    { runSource :: forall r. r -> (r -> a -> EitherT r m r) -> EitherT r m r }  type Conduit a m b = Source m a -> Source m b type Sink a m r    = Source m a -> m r +instance Monad m => Semigroup (Source m a) where+    x <> y = Source $ \r f -> lift $ do+        r' <- sink r f x+        sink r' f y+    {-# INLINE (<>) #-}+ instance Monad m => Monoid (Source m a) where-    mempty = Source $ const . return-    mappend x y = Source $ \r f -> flip (getSource y) f =<< getSource x r f+    mempty  = skip+    {-# INLINE mempty #-}+    mappend = (<>)+    {-# INLINE mappend #-} +instance Monad m => Alternative (Source m) where+    empty = skip+    {-# INLINE empty #-}+    (<|>) = (<>)+    {-# INLINE (<|>) #-}+ instance Functor (Source m) where-    fmap f (Source await) = Source $ \z yield -> await z $ \r x -> yield r (f x)+    fmap f = conduit $ \r yield -> yield r . f+    {-# INLINE fmap #-}  instance Applicative (Source m) where     pure  = return+    {-# INLINE pure #-}     (<*>) = ap+    {-# INLINE (<*>) #-} +instance Monad m => MonadPlus (Source m) where+    mzero = abort+    {-# INLINE mzero #-}+    mplus = (<|>)+    {-# INLINE mplus #-}+ instance Monad (Source m) where     return x = Source $ \z yield -> yield z x+    {-# INLINE return #-}     Source await >>= f = Source $ \z yield ->-        await z $ \r x -> getSource (f x) r $ \r' y -> yield r' y+        await z $ \r x -> runSource (f x) r yield+    {-# INLINE (>>=) #-} +instance MFunctor Source where+    hoist nat m = Source $ \z yield -> runSource (hoist nat m) z yield+    {-# INLINE hoist #-}++instance MMonad Source where+    embed f m = Source $ \z yield -> runSource (embed f m) z yield+    {-# INLINE embed #-}+ instance MonadIO m => MonadIO (Source m) where     liftIO m = Source $ \z yield -> yield z =<< liftIO m+    {-# INLINE liftIO #-}  instance MonadTrans Source where     lift m = Source $ \z yield -> yield z =<< lift m+    {-# INLINE lift #-} +instance MonadReader r m => MonadReader r (Source m) where+    ask     = lift ask+    {-# INLINE ask #-}+    local f = conduit $ \r yield -> local f . yield r+    {-# INLINE local #-}+    reader  = lift . reader+    {-# INLINE reader #-}++instance MonadState s m => MonadState s (Source m) where+    get   = lift get+    {-# INLINE get #-}+    put   = lift . put+    {-# INLINE put #-}+    state = lift . state+    {-# INLINE state #-}++instance MonadWriter w m => MonadWriter w (Source m) where+    writer = lift . writer+    {-# INLINE writer #-}+    tell = lift . tell+    {-# INLINE tell #-}+    listen = conduit $ \r yield x -> do+        ((), w) <- listen $ return ()+        yield r (x, w)+    {-# INLINE listen #-}+    pass = conduit $ \r yield (x, f) -> do+        pass $ return ((), f)+        yield r x+    {-# INLINE pass #-}++instance MonadError e m => MonadError e (Source m) where+    throwError = lift . throwError+    {-# INLINE throwError #-}+    catchError src f = Source $ \z yield -> EitherT $+        runEitherT (runSource src z yield)+            `catchError` \e -> runEitherT (runSource (f e) z yield)+    {-# INLINE catchError #-}++instance MonadThrow m => MonadThrow (Source m) where+    throwM e = lift $ throwM e+    {-# INLINE throwM #-}++instance MonadCatch m => MonadCatch (Source m) where+    catch src f = Source $ \z yield -> EitherT $+        runEitherT (runSource src z yield)+            `Catch.catch` \e -> runEitherT (runSource (f e) z yield)+    {-# INLINE catch #-}++instance MonadMask m => MonadMask (Source m) where+    mask a = Source $ \z yield -> EitherT $ Catch.mask $ \u ->+        runEitherT $ runSource (a $ \b -> Source $ \r yield' ->+            EitherT $ liftM Right $ u $ sink r yield' b) z yield+    {-# INLINE mask #-}+    uninterruptibleMask a =+        Source $ \z yield -> EitherT $ Catch.uninterruptibleMask $ \u ->+            runEitherT $ runSource (a $ \b -> Source $ \r yield' ->+                EitherT $ liftM Right $ u $ sink r yield' b) z yield+    {-# INLINE uninterruptibleMask #-}++instance Foldable (Source Identity) where+    foldMap f = runIdentity . sink mempty (\r x -> return $ r `mappend` f x)+    {-# INLINE foldMap #-}+ -- | Sequence a collection of sources. -- -- >>> sinkList $ sequenceSources [yieldOne 1, yieldOne 2, yieldOne 3] -- [[1,2,3]] sequenceSources :: (Traversable f, Monad m) => f (Source m a) -> Source m (f a) sequenceSources = sequenceA---- | Promote any sink to a source.  This can be used as if it were a source---   transformer (aka, a conduit):------ >>> sinkList $ returnC $ sumC $ mapC (+1) $ yieldMany [1..10]--- [65]------ Note that 'returnC' is a synonym for 'Control.Monad.Trans.Class.lift'.-returnC :: Monad m => m a -> Source m a-returnC = lift+{-# INLINE sequenceSources #-}  -- | Compose a 'Source' and a 'Conduit' into a new 'Source'.  Note that this --   is just flipped function application, so ($) can be used to achieve the@@ -126,81 +413,89 @@ ($$) = flip ($) {-# INLINE ($$) #-} --- | This is just like 'Control.Monad.Trans.Either.bimapEitherT', but it only---   requires a 'Monad' constraint rather than 'Functor'.+awaitForever :: (a -> Source m b) -> Conduit a m b+awaitForever = flip (>>=)+{-# INLINE awaitForever #-}++-- | Promote any sink to a source.  This can be used as if it were a source+--   transformer (aka, a conduit):+--+-- >>> sinkList $ returnC $ sumC $ mapC (+1) $ yieldMany [1..10]+-- [65]+--+-- Note that 'returnC' is a synonym for 'Control.Monad.Trans.Class.lift'.+returnC :: Monad m => m a -> Source m a+returnC = lift++abort :: Monad m => Source m a+abort = Source $ const . left+{-# INLINE abort #-}++skip :: Monad m => Source m a+skip = Source $ const . return+{-# INLINE skip #-}++conduit :: (forall r. r -> (r -> b -> EitherT r m r) -> a -> EitherT r m r)+        -> Conduit a m b+conduit f (Source await) = Source $ \z -> await z . flip f+{-# INLINE conduit #-}++sink :: forall m a r. Monad m => r -> (r -> a -> EitherT r m r) -> Sink a m r+sink z f (Source await) = either id id `liftM` runEitherT (await z f)+{-# INLINE sink #-}++-- | Most of the time conduit will pass through the fold variable unmolested,+--   but sometimes you need to ignore that variable and use your own within+--   that stage of the pipeline.  This is done by wrapping the fold variable+--   in a tuple and then unwrapping it when the conduit is done.+--   'conduitWith' makes this transparent.+conduitWith :: Monad m+            => s+            -> (forall r. (r, s) -> (r -> b -> EitherT (r, s) m (r, s)) -> a+                -> EitherT (r, s) m (r, s))+            -> Conduit a m b+conduitWith s f (Source await) = Source $ \z yield ->+    rewrap fst $ await (z, s) $ \(r, t) ->+        f (r, t) (\r' -> rewrap (, t) . yield r')+{-# INLINE conduitWith #-}+ rewrap :: Monad m => (a -> b) -> EitherT a m a -> EitherT b m b rewrap f k = EitherT $ bimap f f `liftM` runEitherT k {-# INLINE rewrap #-} -rewrapM :: Monad m => (a -> EitherT b m b) -> EitherT a m a -> EitherT b m b-rewrapM f k = EitherT $ do-    eres <- runEitherT k-    runEitherT $ either f f eres-{-# INLINE rewrapM #-}--resolve :: Monad m => (r -> a -> EitherT r m r) -> r -> a -> m r-resolve await z f = either id id `liftM` runEitherT (await z f)-{-# INLINE resolve #-}+source :: Monad m+       => (forall r. r -> (r -> a -> EitherT r m r) -> EitherT r m r)+       -> Source m a+source = Source+{-# INLINE source #-}  yieldMany :: (Monad m, MonoFoldable mono) => mono -> Source m (Element mono) yieldMany xs = Source $ \z yield -> ofoldlM yield z xs {-# INLINE yieldMany #-} -sourceList :: (Monad m, MonoFoldable mono) => mono -> Source m (Element mono)+sourceList :: Monad m => [a] -> Source m a sourceList = yieldMany {-# INLINE sourceList #-} -yieldOne :: Monad m => a -> Source m a-yieldOne x = Source $ \z yield -> yield z x-{-# INLINE yieldOne #-}- unfoldC :: forall m a b. Monad m => (b -> Maybe (a, b)) -> b -> Source m a-unfoldC f i = Source $ go i-  where-    go :: forall r. b -> r -> (r -> a -> EitherT r m r) -> EitherT r m r-    go y z yield = loop y z-      where-        loop x r = case f x of-            Nothing      -> return r-            Just (a, x') -> loop x' =<< yield r a+unfoldC = (yieldMany .) . Data.List.unfoldr+{-# INLINE unfoldC #-}  enumFromToC :: forall m a. (Monad m, Enum a, Eq a) => a -> a -> Source m a-enumFromToC start stop = Source $ go start-  where-    go :: forall r. a -> r -> (r -> a -> EitherT r m r) -> EitherT r m r-    go y z yield = loop y z-      where-        loop a r-            | a == stop = return r-            | otherwise = loop (succ a) =<< yield r a+enumFromToC = (yieldMany .) . enumFromTo+{-# INLINE enumFromToC #-}  iterateC :: forall m a. Monad m => (a -> a) -> a -> Source m a-iterateC f i = Source $ go i-  where-    go :: forall r. a -> r -> (r -> a -> EitherT r m r) -> EitherT r m r-    go y z yield = loop y z-      where-        loop x r = let x' = f x-                   in loop x' =<< yield r x'+iterateC = (yieldMany .) . iterate+{-# INLINE iterateC #-}  repeatC :: forall m a. Monad m => a -> Source m a-repeatC x = Source go-  where-    go :: forall r. r -> (r -> a -> EitherT r m r) -> EitherT r m r-    go z yield = loop z-      where-        loop y = loop =<< yield y x+repeatC = yieldMany . Prelude.repeat {-# INLINE repeatC #-}  replicateC :: forall m a. Monad m => Int -> a -> Source m a-replicateC n x = Source $ go n-  where-    go :: Int -> r -> (r -> a -> EitherT r m r) -> EitherT r m r-    go i z yield = loop i z-      where-        loop n' r-            | n' >= 0   = loop (n' - 1) =<< yield r x-            | otherwise = return r+replicateC = (yieldMany .) . Prelude.replicate+{-# INLINE replicateC #-}  sourceLazy :: (Monad m, LazySequence lazy strict) => lazy -> Source m strict sourceLazy = yieldMany . toChunks@@ -250,45 +545,50 @@ sourceFile :: (MonadBaseControl IO m, MonadIO m, IOData a)            => FilePath -> Source m a sourceFile path = Source $ \z yield ->-    bracket-        (liftIO $ openFile path ReadMode)-        (liftIO . hClose)-        (\h -> getSource (sourceHandle h) z yield)+    bracket (liftIO $ openFile path ReadMode) (liftIO . hClose)+        (\h -> runSource (sourceHandle h) z yield)+{-# INLINE sourceFile #-}  sourceIOHandle :: (MonadBaseControl IO m, MonadIO m, IOData a)                => IO Handle -> Source m a sourceIOHandle f = Source $ \z yield ->-    bracket-        (liftIO f)-        (liftIO . hClose)-        (\h -> getSource (sourceHandle h) z yield)+    bracket (liftIO f) (liftIO . hClose)+        (\h -> runSource (sourceHandle h) z yield)+{-# INLINE sourceIOHandle #-}  stdinC :: (MonadBaseControl IO m, MonadIO m, IOData a) => Source m a stdinC = sourceHandle stdin+{-# INLINE stdinC #-}  initRepeat :: Monad m => m seed -> (seed -> m a) -> Source m a initRepeat mseed f = Source $ \z yield ->-    lift mseed >>= \seed -> getSource (repeatMC (f seed)) z yield+    lift mseed >>= \seed -> runSource (repeatMC (f seed)) z yield+{-# INLINE initRepeat #-}  initReplicate :: Monad m => m seed -> (seed -> m a) -> Int -> Source m a initReplicate mseed f n = Source $ \z yield ->-    lift mseed >>= \seed -> getSource (replicateMC n (f seed)) z yield+    lift mseed >>= \seed -> runSource (replicateMC n (f seed)) z yield+{-# INLINE initReplicate #-}  sourceRandom :: (Variate a, MonadIO m) => Source m a sourceRandom =     initRepeat (liftIO MWC.createSystemRandom) (liftIO . MWC.uniform)+{-# INLINE sourceRandom #-}  sourceRandomN :: (Variate a, MonadIO m) => Int -> Source m a sourceRandomN =     initReplicate (liftIO MWC.createSystemRandom) (liftIO . MWC.uniform)+{-# INLINE sourceRandomN #-}  sourceRandomGen :: (Variate a, MonadBase base m, PrimMonad base)                 => Gen (PrimState base) -> Source m a sourceRandomGen gen = initRepeat (return gen) (liftBase . MWC.uniform)+{-# INLINE sourceRandomGen #-}  sourceRandomNGen :: (Variate a, MonadBase base m, PrimMonad base)                  => Gen (PrimState base) -> Int -> Source m a sourceRandomNGen gen = initReplicate (return gen) (liftBase . MWC.uniform)+{-# INLINE sourceRandomNGen #-}  sourceDirectory :: forall m. (MonadBaseControl IO m, MonadIO m)                 => FilePath -> Source m FilePath@@ -314,7 +614,7 @@     go :: r -> (r -> FilePath -> EitherT r m r) -> EitherT r m r     go z yield = start startDir z       where-        start dir r = getSource (sourceDirectory dir) r entry+        start dir r = runSource (sourceDirectory dir) r entry         entry r fp = do             ft <- liftIO $ F.getFileType fp             case ft of@@ -327,112 +627,125 @@                 F.FTOther -> return r  dropC :: Monad m => Int -> Conduit a m a-dropC n (Source await) = Source $ \z yield ->-    rewrap snd $ await (n, z) (go yield)+dropC n = conduitWith n go   where-    go _ (n', r) _ | n' > 0 = return (n' - 1, r)-    go yield (_, r) x = rewrap (0,) $ yield r x+    go (r, n') _ _ | n' > 0 = return (r, n' - 1)+    go (r, _) yield x       = yield r x+{-# INLINE dropC #-} -dropCE :: (Monad m, IsSequence seq)-       => Index seq -> Conduit seq m seq-dropCE n (Source await) = Source $ \z yield ->-    rewrap snd $ await (n, z) (go yield)+dropCE :: (Monad m, IsSequence seq) => Index seq -> Conduit seq m seq+dropCE n = conduitWith n go   where-    go yield (n', r) s-        | onull y   = return (n' - xn, r)-        | otherwise = rewrap (0,) $ yield r y+    go (r, n') yield s+        | onull y   = return (r, n' - xn)+        | otherwise = yield r y       where         (x, y) = Seq.splitAt n' s         xn = n' - fromIntegral (olength x)  dropWhileC :: Monad m => (a -> Bool) -> Conduit a m a-dropWhileC f (Source await) = Source $ \z yield ->-    rewrap snd $ await (f, z) (go yield)+dropWhileC f = conduitWith f go   where-    go _ (k, r) x | k x = return (k, r)-    go yield (_, r) x = rewrap (const False,) $ yield r x+    go (r, k) _ x | k x = return (r, k)+    -- Change out the predicate for one that always fails+    go (r, _) yield x = fmap (const (const False)) <$> yield r x  dropWhileCE :: (Monad m, IsSequence seq)             => (Element seq -> Bool)             -> Conduit seq m seq-dropWhileCE f (Source await) =-    Source $ \z yield -> rewrap snd $ await (f, z) (go yield)+dropWhileCE f = conduitWith f go   where-    go yield (k, r) s-        | onull x   = return (k, r)-        | otherwise = rewrap (const False,) $ yield r s+    go (r, k) yield s+        | onull x   = return (r, k)+        | otherwise = fmap (const (const False)) <$> yield r s       where         x = Seq.dropWhile k s  foldC :: (Monad m, Monoid a) => Sink a m a foldC = foldMapC id+{-# INLINE foldC #-}  foldCE :: (Monad m, MonoFoldable mono, Monoid (Element mono))        => Sink mono m (Element mono)-foldCE = foldlC (\acc mono -> acc <> ofoldMap id mono) mempty+foldCE = foldlC (\acc mono -> acc `mappend` ofoldMap id mono) mempty+{-# INLINE foldCE #-}  foldlC :: Monad m => (a -> b -> a) -> a -> Sink b m a-foldlC f z (Source await) = resolve await z ((return .) . f)+foldlC f z = sink z ((return .) . f) {-# INLINE foldlC #-}  foldlCE :: (Monad m, MonoFoldable mono)         => (a -> Element mono -> a) -> a -> Sink mono m a foldlCE f = foldlC (ofoldl' f)+{-# INLINE foldlCE #-}  foldMapC :: (Monad m, Monoid b) => (a -> b) -> Sink a m b-foldMapC f = foldlC (\acc x -> acc <> f x) mempty+foldMapC f = foldlC (\acc x -> acc `mappend` f x) mempty+{-# INLINE foldMapC #-}  foldMapCE :: (Monad m, MonoFoldable mono, Monoid w)           => (Element mono -> w) -> Sink mono m w foldMapCE = foldMapC . ofoldMap+{-# INLINE foldMapCE #-}  allC :: Monad m => (a -> Bool) -> Sink a m Bool allC f = liftM getAll `liftM` foldMapC (All . f)+{-# INLINE allC #-}  allCE :: (Monad m, MonoFoldable mono)       => (Element mono -> Bool) -> Sink mono m Bool allCE = allC . oall+{-# INLINE allCE #-}  anyC :: Monad m => (a -> Bool) -> Sink a m Bool anyC f = liftM getAny `liftM` foldMapC (Any . f)+{-# INLINE anyC #-}  anyCE :: (Monad m, MonoFoldable mono)       => (Element mono -> Bool) -> Sink mono m Bool anyCE = anyC . oany+{-# INLINE anyCE #-}  andC :: Monad m => Sink Bool m Bool andC = allC id+{-# INLINE andC #-}  andCE :: (Monad m, MonoFoldable mono, Element mono ~ Bool)       => Sink mono m Bool andCE = allCE id+{-# INLINE andCE #-}  orC :: Monad m => Sink Bool m Bool orC = anyC id+{-# INLINE orC #-}  orCE :: (Monad m, MonoFoldable mono, Element mono ~ Bool)      => Sink mono m Bool orCE = anyCE id+{-# INLINE orCE #-}  elemC :: (Monad m, Eq a) => a -> Sink a m Bool elemC x = anyC (== x)+{-# INLINE elemC #-}  elemCE :: (Monad m, EqSequence seq) => Element seq -> Sink seq m Bool elemCE = anyC . Seq.elem+{-# INLINE elemCE #-}  notElemC :: (Monad m, Eq a) => a -> Sink a m Bool notElemC x = allC (/= x)+{-# INLINE notElemC #-}  notElemCE :: (Monad m, EqSequence seq) => Element seq -> Sink seq m Bool notElemCE = allC . Seq.notElem+{-# INLINE notElemCE #-} -produceList :: Monad m => ([a] -> b) -> Source m a -> m b-produceList f (Source await) =-    (f . ($ [])) `liftM` resolve await id (\front x -> return (front . (x:)))+produceList :: Monad m => ([a] -> b) -> Sink a m b+produceList f =+    liftM (liftM (f . ($ []))) $ sink id (\front x -> return (front . (x:))) {-# INLINE produceList #-} -sinkLazy :: (Monad m, LazySequence lazy strict)-         => Sink strict m lazy+sinkLazy :: (Monad m, LazySequence lazy strict) => Sink strict m lazy sinkLazy = produceList fromChunks -- {-# INLINE sinkLazy #-} @@ -451,21 +764,26 @@ sinkBuilder :: (Monad m, Monoid builder, ToBuilder a builder)             => Sink a m builder sinkBuilder = foldMapC toBuilder+{-# INLINE sinkBuilder #-}  sinkLazyBuilder :: (Monad m, Monoid builder, ToBuilder a builder,                     Builder builder lazy)                 => Sink a m lazy sinkLazyBuilder = liftM builderToLazy . foldMapC toBuilder+{-# INLINE sinkLazyBuilder #-}  sinkNull :: Monad m => Sink a m () sinkNull _ = return ()+{-# INLINE sinkNull #-}  awaitNonNull :: (Monad m, MonoFoldable a) => Conduit a m (Maybe (NonNull a))-awaitNonNull (Source await) = Source $ \z yield -> await z $ \r x ->+awaitNonNull = conduit $ \r yield x ->     maybe (return r) (yield r . Just) (NonNull.fromNullable x)+{-# INLINE awaitNonNull #-}  headCE :: (Monad m, IsSequence seq) => Sink seq m (Maybe (Element seq)) headCE = undefined+{-# INLINE headCE #-}  -- jww (2014-06-07): These two cannot be implemented without leftover support. -- peekC :: Monad m => Sink a m (Maybe a)@@ -475,41 +793,45 @@ -- peekCE = undefined  lastC :: Monad m => Sink a m (Maybe a)-lastC (Source await) = resolve await Nothing (const (return . Just))+lastC = sink Nothing (const (return . Just))+{-# INLINE lastC #-}  lastCE :: (Monad m, IsSequence seq) => Sink seq m (Maybe (Element seq)) lastCE = undefined+{-# INLINE lastCE #-}  lengthC :: (Monad m, Num len) => Sink a m len lengthC = foldlC (\x _ -> x + 1) 0+{-# INLINE lengthC #-}  lengthCE :: (Monad m, Num len, MonoFoldable mono) => Sink mono m len lengthCE = foldlC (\x y -> x + fromIntegral (olength y)) 0+{-# INLINE lengthCE #-}  lengthIfC :: (Monad m, Num len) => (a -> Bool) -> Sink a m len lengthIfC f = foldlC (\cnt a -> if f a then cnt + 1 else cnt) 0+{-# INLINE lengthIfC #-}  lengthIfCE :: (Monad m, Num len, MonoFoldable mono)            => (Element mono -> Bool) -> Sink mono m len lengthIfCE f = foldlCE (\cnt a -> if f a then cnt + 1 else cnt) 0+{-# INLINE lengthIfCE #-}  maximumC :: (Monad m, Ord a) => Sink a m (Maybe a)-maximumC (Source await) = resolve await Nothing $ \r y ->-    return $ Just $ case r of-        Just x -> max x y-        _      -> y+maximumC = sink Nothing $ \r y -> return $ Just $ maybe y (max y) r+{-# INLINE maximumC #-}  maximumCE :: (Monad m, OrdSequence seq) => Sink seq m (Maybe (Element seq)) maximumCE = undefined+{-# INLINE maximumCE #-}  minimumC :: (Monad m, Ord a) => Sink a m (Maybe a)-minimumC (Source await) = resolve await Nothing $ \r y ->-    return $ Just $ case r of-        Just x -> min x y-        _      -> y+minimumC = sink Nothing $ \r y -> return $ Just $ maybe y (min y) r+{-# INLINE minimumC #-}  minimumCE :: (Monad m, OrdSequence seq) => Sink seq m (Maybe (Element seq)) minimumCE = undefined+{-# INLINE minimumCE #-}  -- jww (2014-06-07): These two cannot be implemented without leftover support. -- nullC :: Monad m => Sink a m Bool@@ -520,101 +842,112 @@  sumC :: (Monad m, Num a) => Sink a m a sumC = foldlC (+) 0+{-# INLINE sumC #-}  sumCE :: (Monad m, MonoFoldable mono, Num (Element mono))       => Sink mono m (Element mono) sumCE = undefined+{-# INLINE sumCE #-}  productC :: (Monad m, Num a) => Sink a m a productC = foldlC (*) 1+{-# INLINE productC #-}  productCE :: (Monad m, MonoFoldable mono, Num (Element mono))           => Sink mono m (Element mono) productCE = undefined+{-# INLINE productCE #-}  findC :: Monad m => (a -> Bool) -> Sink a m (Maybe a)-findC f (Source await) = resolve await Nothing $ \r x ->-    if f x then left (Just x) else return r+findC f = sink Nothing $ \r x -> if f x then left (Just x) else return r+{-# INLINE findC #-}  mapM_C :: Monad m => (a -> m ()) -> Sink a m ()-mapM_C f (Source await) = resolve await () (const $ lift . f)+mapM_C f = sink () (const $ lift . f) {-# INLINE mapM_C #-}  mapM_CE :: (Monad m, MonoFoldable mono)         => (Element mono -> m ()) -> Sink mono m () mapM_CE = undefined+{-# INLINE mapM_CE #-}  foldMC :: Monad m => (a -> b -> m a) -> a -> Sink b m a-foldMC f z (Source await) = resolve await z (\r x -> lift (f r x))+foldMC f = flip sink ((lift .) . f)+{-# INLINE foldMC #-}  foldMCE :: (Monad m, MonoFoldable mono)         => (a -> Element mono -> m a) -> a -> Sink mono m a foldMCE = undefined+{-# INLINE foldMCE #-}  foldMapMC :: (Monad m, Monoid w) => (a -> m w) -> Sink a m w-foldMapMC f = foldMC (\acc x -> (acc <>) `liftM` f x) mempty+foldMapMC f = foldMC (\acc x -> (acc `mappend`) `liftM` f x) mempty+{-# INLINE foldMapMC #-}  foldMapMCE :: (Monad m, MonoFoldable mono, Monoid w)            => (Element mono -> m w) -> Sink mono m w foldMapMCE = undefined+{-# INLINE foldMapMCE #-}  sinkFile :: (MonadBaseControl IO m, MonadIO m, IOData a)          => FilePath -> Sink a m () sinkFile fp = sinkIOHandle (liftIO $ openFile fp WriteMode)+{-# INLINE sinkFile #-}  sinkHandle :: (MonadIO m, IOData a) => Handle -> Sink a m () sinkHandle = mapM_C . hPut+{-# INLINE sinkHandle #-}  sinkIOHandle :: (MonadBaseControl IO m, MonadIO m, IOData a)              => IO Handle -> Sink a m () sinkIOHandle alloc =-    bracket-        (liftIO alloc)-        (liftIO . hClose)-        . flip sinkHandle+    bracket (liftIO alloc) (liftIO . hClose) . flip sinkHandle+{-# INLINE sinkIOHandle #-}  printC :: (Show a, MonadIO m) => Sink a m () printC = mapM_C (liftIO . print)+{-# INLINE printC #-}  stdoutC :: (MonadIO m, IOData a) => Sink a m () stdoutC = sinkHandle stdout+{-# INLINE stdoutC #-}  stderrC :: (MonadIO m, IOData a) => Sink a m () stderrC = sinkHandle stderr+{-# INLINE stderrC #-}  mapC :: Monad m => (a -> b) -> Conduit a m b-mapC f (Source await) = Source $ \z yield -> await z $ \acc -> yield acc . f+mapC = fmap {-# INLINE mapC #-} -mapC' :: Monad m => (a -> b) -> Conduit a m b-mapC' f (Source await) = Source $ \z yield -> await z $ \acc x ->-    let y = f x in y `seq` acc `seq` yield acc y-{-# INLINE mapC' #-}- mapCE :: (Monad m, Functor f) => (a -> b) -> Conduit (f a) m (f b) mapCE = undefined+{-# INLINE mapCE #-}  omapCE :: (Monad m, MonoFunctor mono)        => (Element mono -> Element mono) -> Conduit mono m mono omapCE = undefined+{-# INLINE omapCE #-}  concatMapC :: (Monad m, MonoFoldable mono)            => (a -> mono) -> Conduit a m (Element mono)-concatMapC f (Source await) = Source $ \z yield -> await z $ \r x -> ofoldlM yield r (f x)+concatMapC f = conduit $ \r yield -> ofoldlM yield r . f+{-# INLINE concatMapC #-}  concatMapCE :: (Monad m, MonoFoldable mono, Monoid w)             => (Element mono -> w) -> Conduit mono m w concatMapCE = undefined+{-# INLINE concatMapCE #-} -takeC :: Monad m => Int -> Source m a -> Source m a-takeC n (Source await) = Source $ \z yield -> rewrap snd $ await (n, z) (go yield)+takeC :: Monad m => Int -> Conduit a m a+takeC n = conduitWith n go   where-    go yield (n', z') x+    go (z', n') yield x         | n' > 1    = next         | n' > 0    = left =<< next-        | otherwise = left (0, z')+        | otherwise = left (z', 0)       where-        next = rewrap (n' - 1,) $ yield z' x+        next = fmap pred <$> yield z' x  takeCE :: (Monad m, IsSequence seq) => Index seq -> Conduit seq m seq takeCE = undefined@@ -623,11 +956,11 @@ --   problem if Sinks were monadic, as they are in conduit or pipes.  There is --   no such concept as "resuming where the last conduit left off" in this --   library.-takeWhileC :: Monad m => (a -> Bool) -> Source m a -> Source m a-takeWhileC f (Source await) = Source $ \z yield -> rewrap snd $ await (f, z) (go yield)+takeWhileC :: Monad m => (a -> Bool) -> Conduit a m a+takeWhileC f = conduitWith f go   where-    go yield (k, z') x | k x = rewrap (k,) $ yield z' x-    go _ (_, z') _ = left (const False, z')+    go (z', k) yield x | k x = yield z' x+    go (z', _) _ _           = left (z', const False)  takeWhileCE :: (Monad m, IsSequence seq)             => (Element seq -> Bool) -> Conduit seq m seq@@ -644,16 +977,17 @@ concatC = undefined  filterC :: Monad m => (a -> Bool) -> Conduit a m a-filterC f (Source await) = Source $ \z yield ->-    await z $ \r x -> if f x then yield r x else return r+filterC f = awaitForever $ \x -> if f x then return x else skip+{-# INLINE filterC #-}  filterCE :: (IsSequence seq, Monad m)          => (Element seq -> Bool) -> Conduit seq m seq filterCE = undefined+{-# INLINE filterCE #-}  mapWhileC :: Monad m => (a -> Maybe b) -> Conduit a m b-mapWhileC f (Source await) = Source $ \z yield -> await z $ \z' x ->-    maybe (left z') (yield z') (f x)+mapWhileC f = awaitForever $ \x -> case f x of Just y -> return y; _ -> abort+{-# INLINE mapWhileC #-}  conduitVector :: (MonadBase base m, Vector v a, PrimMonad base)               => Int -> Conduit a m (v a)@@ -669,8 +1003,7 @@ intersperseC s (Source await) = Source $ \z yield -> EitherT $ do     eres <- runEitherT $ await (Nothing, z) $ \(my, r) x ->         case my of-            Nothing ->-                return (Just x, r)+            Nothing -> return (Just x, r)             Just y  -> do                 r' <- rewrap (Nothing,) $ yield r y                 rewrap (Just x,) $ yield (snd r') s@@ -698,11 +1031,12 @@ decodeBase16C = undefined  mapMC :: Monad m => (a -> m b) -> Conduit a m b-mapMC f (Source await) = Source $ \z yield -> await z (\r x -> yield r =<< lift (f x))+mapMC f src = src >>= lift . f {-# INLINE mapMC #-}  mapMCE :: (Monad m, Traversable f) => (a -> m b) -> Conduit (f a) m (f b) mapMCE = undefined+{-# INLINE mapMCE #-}  omapMCE :: (Monad m, MonoTraversable mono)         => (Element mono -> m (Element mono)) -> Conduit mono m mono@@ -710,14 +1044,15 @@  concatMapMC :: (Monad m, MonoFoldable mono)             => (a -> m mono) -> Conduit a m (Element mono)-concatMapMC = undefined+concatMapMC f = awaitForever $ yieldMany <=< lift . f  filterMC :: Monad m => (a -> m Bool) -> Conduit a m a-filterMC f (Source await) = Source $ \z yield -> await z $ \z' x -> do+filterMC f = awaitForever $ \x -> do     res <- lift $ f x     if res-        then yield z' x-        else return z'+        then return x+        else skip+{-# INLINE filterMC #-}  filterMCE :: (Monad m, IsSequence seq)           => (Element seq -> m Bool) -> Conduit seq m seq@@ -735,6 +1070,7 @@  encodeUtf8C :: (Monad m, Utf8 text binary) => Conduit text m binary encodeUtf8C = mapC encodeUtf8+{-# INLINE encodeUtf8C #-}  decodeUtf8C :: MonadThrow m => Conduit ByteString m Text decodeUtf8C = undefined@@ -749,11 +1085,13 @@  unlinesC :: (Monad m, IsSequence seq, Element seq ~ Char)          => Conduit seq m seq-unlinesC = concatMapC (:[Seq.singleton '\n'])+unlinesC = concatMapC (: [Seq.singleton '\n'])+{-# INLINE unlinesC #-}  unlinesAsciiC :: (Monad m, IsSequence seq, Element seq ~ Word8)               => Conduit seq m seq-unlinesAsciiC = concatMapC (:[Seq.singleton 10])+unlinesAsciiC = concatMapC (: [Seq.singleton 10])+{-# INLINE unlinesAsciiC #-}  linesUnboundedC_ :: forall m seq. (Monad m, IsSequence seq, Eq (Element seq))                  => Element seq -> Conduit seq m seq@@ -782,26 +1120,16 @@ linesUnboundedC :: (Monad m, IsSequence seq, Element seq ~ Char)                 => Conduit seq m seq linesUnboundedC = linesUnboundedC_ '\n'+{-# INLINE linesUnboundedC #-}  linesUnboundedAsciiC :: (Monad m, IsSequence seq, Element seq ~ Word8)                      => Conduit seq m seq linesUnboundedAsciiC = linesUnboundedC_ 10---- | The use of 'awaitForever' in this library is just a bit different from---   conduit:------ >>> awaitForever $ \x yield done -> if even x then yield x else done-awaitForever :: Monad m-             => (forall r. a -> (b -> EitherT r m r) -> EitherT r m r-                 -> EitherT r m r)-             -> Conduit a m b-awaitForever f (Source await) = Source $ \z yield ->-    await z $ \r x -> f x (yield r) (return r)+{-# INLINE linesUnboundedAsciiC #-} -{- -- | Zip sinks together.  This function may be used multiple times: ----- >>> let mySink s await => resolve await () $ \() x -> liftIO $ print $ s <> show x+-- >>> let mySink s = sink () $ \() x -> liftIO $ print $ s <> show x -- >>> zipSinks sinkList (zipSinks (mySink "foo") (mySink "bar")) $ yieldMany [1,2,3] -- "foo: 1" -- "bar: 1"@@ -811,32 +1139,48 @@ -- "bar: 3" -- ([1,2,3],((),())) ----- jww (2014-06-09): Can this be written sanely without resorting to IORefs?-zipSinks :: MonadIO m-         => Sink a (StateT s m) r -> Sink a (StateT s' m) r'-         -> Sink a m (r, r')-zipSinks x y (Source await) = do-    res_s  <- liftIO $ newIORef (error "zipSinks: s never assigned a value")-    res_r' <- liftIO $ newIORef (error "zipSinks: r' never assigned a value")-    r <- x $ Source $ \rx yieldx -> do-        r' <- lift $ y $ Source $ \ry yieldy -> EitherT $ do-            (rx', ry') <- resolve await (rx, ry) $ \(rx', ry') u -> EitherT $ do-                eres <- runEitherT $ yieldx rx' u-                case eres of-                    Left r -> return $ Left (r, ry')-                    Right r -> do-                        eres' <- runEitherT $ yieldy ry' u-                        case eres' of-                            Left r' -> return $ Left (r, r')-                            Right r' -> return $ Right (r, r')-            liftIO $ writeIORef res_s rx'-            return $ Right ry'-        liftIO $ do-            writeIORef res_r' r'-            readIORef res_s-    r' <- liftIO $ readIORef res_r'-    return (r, r')+-- Note that the two sinks are run concurrently, so watch out for possible+-- race conditions if they try to interact with the same resources.+zipSinks :: forall a m r r'. (MonadBaseControl IO m, MonadIO m)+         => Sink a m r -> Sink a m r' -> Sink a m (r, r')+zipSinks sink1 sink2 (Source await) = do+    x <- liftIO newEmptyMVar+    y <- liftIO newEmptyMVar+    withAsync (sink1 $ sourceMaybeMVar x) $ \a ->+        withAsync (sink2 $ sourceMaybeMVar y) $ \b -> do+            _ <- runEitherT $ await () $ \() val -> do+                liftIO $ putMVar x (Just val)+                liftIO $ putMVar y (Just val)+            liftIO $ putMVar x Nothing+            liftIO $ putMVar y Nothing+            waitBoth a b +-- | Keep taking from an @MVar (Maybe a)@ until it yields 'Nothing'.+sourceMaybeMVar :: forall m a. MonadIO m => MVar (Maybe a) -> Source m a+sourceMaybeMVar var = Source go+  where+    go :: r -> (r -> a -> EitherT r m r) -> EitherT r m r+    go z yield = loop z+      where+        loop r = do+            mx <- liftIO $ takeMVar var+            case mx of+                Nothing -> return r+                Just x  -> loop =<< yield r x++-- | Keep taking from an @TMVar (Maybe a)@ until it yields 'Nothing'.+sourceMaybeTMVar :: forall a. TMVar (Maybe a) -> Source STM a+sourceMaybeTMVar var = Source go+  where+    go :: r -> (r -> a -> EitherT r STM r) -> EitherT r STM r+    go z yield = loop z+      where+        loop r = do+            mx <- lift $ takeTMVar var+            case mx of+                Nothing -> return r+                Just x  -> loop =<< yield r x+ newtype ZipSink a m r = ZipSink { getZipSink :: Source m a -> m r }  instance Monad m => Functor (ZipSink a m) where@@ -852,12 +1196,12 @@ -- Implemented on top of @ZipSink@, see that data type for more details. sequenceSinks :: (Traversable f, Monad m) => f (Sink a m r) -> Sink a m (f r) sequenceSinks = getZipSink . sequenceA . fmap ZipSink--}+{-# INLINE sequenceSinks #-}  asyncC :: (MonadBaseControl IO m, Monad m)        => (a -> m b) -> Conduit a m (Async (StM m b))-asyncC f (Source await) = Source $ \k yield ->-    await k $ \r x -> yield r =<< lift (async (f x))+asyncC f = awaitForever $ lift . async . f+{-# INLINE asyncC #-}  -- | Convert a 'Control.Foldl.FoldM' fold abstraction into a Sink. --@@ -866,8 +1210,10 @@ -- >>> fromFoldM (FoldM ((return .) . (+)) (return 0) return) $ yieldMany [1..10] -- 55 fromFoldM :: Monad m => FoldM m a b -> Source m a -> m b-fromFoldM (FoldM step initial final) (Source await) =-    initial >>= flip (resolve await) ((lift .) . step) >>= final+fromFoldM (FoldM step initial final) src = do+    r <- initial+    final =<< sink r ((lift .) . step) src+{-# INLINE fromFoldM #-}  -- | Convert a Sink into a 'Control.Foldl.FoldM', passing it into a --   continuation.@@ -876,50 +1222,38 @@ -- 55 toFoldM :: Monad m         => Sink a m r -> (forall s. FoldM (EitherT s m) a s -> EitherT s m s) -> m r-toFoldM sink f = sink $ Source $ \k yield -> f $ FoldM yield (return k) return+toFoldM s f = s $ source $ \k yield -> f $ FoldM yield (return k) return+{-# INLINE toFoldM #-} --- | A Source for exhausting a TChan, but blocks if it is initially empty.-sourceTChan :: forall a. TChan a -> Source STM a-sourceTChan chan = Source go+sourceSTM :: forall container a. (container a -> STM a)+          -> (container a -> STM Bool)+          -> container a+          -> Source STM a+sourceSTM getter tester chan = Source go   where     go :: r -> (r -> a -> EitherT r STM r) -> EitherT r STM r     go z yield = loop z       where         loop r = do-            x  <- lift $ readTChan chan+            x  <- lift $ getter chan             r' <- yield r x-            mt <- lift $ isEmptyTChan chan+            mt <- lift $ tester chan             if mt                 then return r'                 else loop r' +-- | A Source for exhausting a TChan, but blocks if it is initially empty.+sourceTChan :: forall a. TChan a -> Source STM a+sourceTChan = sourceSTM readTChan isEmptyTChan+{-# INLINE sourceTChan #-}+ sourceTQueue :: forall a. TQueue a -> Source STM a-sourceTQueue chan = Source go-  where-    go :: r -> (r -> a -> EitherT r STM r) -> EitherT r STM r-    go z yield = loop z-      where-        loop r = do-            x  <- lift $ readTQueue chan-            r' <- yield r x-            mt <- lift $ isEmptyTQueue chan-            if mt-                then return r'-                else loop r'+sourceTQueue = sourceSTM readTQueue isEmptyTQueue+{-# INLINE sourceTQueue #-}  sourceTBQueue :: forall a. TBQueue a -> Source STM a-sourceTBQueue chan = Source go-  where-    go :: r -> (r -> a -> EitherT r STM r) -> EitherT r STM r-    go z yield = loop z-      where-        loop r = do-            x  <- lift $ readTBQueue chan-            r' <- yield r x-            mt <- lift $ isEmptyTBQueue chan-            if mt-                then return r'-                else loop r'+sourceTBQueue = sourceSTM readTBQueue isEmptyTBQueue+{-# INLINE sourceTBQueue #-}  untilMC :: forall m a. Monad m => m a -> m Bool -> Source m a untilMC m f = Source go

@@ -1,5 +1,5 @@ Name:                simple-conduit-Version:             0.3.0+Version:             0.4.0 Synopsis:            A simple streaming library based on composing monadic folds. Description:     @simple-conduit@ follows a similar UI to the more capable @conduit@ library, but reduces the scope of what it can solve donw to what can be expressed by chaining monadic folds that allow for early termination.  This allows for more predictable resource management behavior, at the cost of not allowing scenarios that @conduit@ is better designed.@@ -31,6 +31,7 @@                      , mtl                      , mwc-random                      , primitive+                     , semigroups                      , stm                      , streaming-commons                      , text

@@ -59,6 +59,5 @@ {-# INLINE yieldMany2 #-}  sinkList2 :: Monad m => Sink a m [a]-sinkList2 (Source await) =-    liftM ($ []) $ resolve await id $ \r x -> return (r . (x:))+sinkList2 = liftM (liftM ($ [])) $ sink id $ \r x -> return (r . (x:)) {-# INLINE sinkList2 #-}