Conduit.Simple

Please see the project README for more details:

https://github.com/jwiegley/simple-conduit/blob/master/README.md

Also see this blog article:

https://www.newartisans.com/2014/06/simpler-conduit-library

Documentation

newtype Source m a

The type of Source should recall foldM:

Monad m => (a -> b -> m a) -> a -> [b] -> m a

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]

Constructors

Source

Fields

runSource :: forall r. r -> (r -> a -> EitherT r m r) -> EitherT r m r

Instances

MFunctor Source

MMonad Source

MonadTrans Source

MonadError e m => MonadError e (Source m)

MonadReader r m => MonadReader r (Source m)

MonadState s m => MonadState s (Source m)

MonadWriter w m => MonadWriter w (Source m)

Monad m => Alternative (Source m)

Monad (Source m)

Functor (Source m)

Monad m => MonadPlus (Source m)

Applicative (Source m)

Foldable (Source Identity)

MonadThrow m => MonadThrow (Source m)

MonadCatch m => MonadCatch (Source m)

MonadMask m => MonadMask (Source m)

MonadIO m => MonadIO (Source m)

Monad m => Monoid (Source m a)

Monad m => Semigroup (Source m a)

type Conduit a m b = Source m a -> Source m b

type Sink a m r = Source m a -> m r

sequenceSources :: (Traversable f, Monad m) => f (Source m a) -> Source m (f a)

Sequence a collection of sources.

>>> sinkList $ sequenceSources [yieldOne 1, yieldOne 2, yieldOne 3]
[[1,2,3]]

newtype ZipSink a m r

Constructors

ZipSink

Fields

getZipSink :: Source m a -> m r

Instances

Monad m => Functor (ZipSink a m)

Monad m => Applicative (ZipSink a m)

sequenceSinks :: (Traversable f, Monad m) => f (Sink a m r) -> Sink a m (f r)

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.

source :: Monad m => (forall r. r -> (r -> a -> EitherT r m r) -> EitherT r m r) -> Source m a

conduit :: (forall r. r -> (r -> b -> EitherT r m r) -> a -> EitherT r m r) -> Conduit a m b

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

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.

sink :: forall m a r. Monad m => r -> (r -> a -> EitherT r m r) -> Sink a m r

($=) :: a -> (a -> b) -> b infixl 1

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 same thing.

(=$) :: (a -> b) -> (b -> c) -> a -> c infixr 2

Compose a Conduit and a Sink into a new Sink. Note that this is just function composition, so (.) can be used to achieve the same thing.

($$) :: a -> (a -> b) -> b infixr 0

Compose a Source and a Sink and compute the result. Note that this is just flipped function application, so ($) can be used to achieve the same thing.

returnC :: Monad m => m a -> Source m a

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 lift.

abort :: Monad m => Source m a

skip :: Monad m => Source m a

awaitForever :: (a -> Source m b) -> Conduit a m b

yieldMany :: (Monad m, MonoFoldable mono) => mono -> Source m (Element mono)

sourceList :: Monad m => [a] -> Source m a

unfoldC :: forall m a b. Monad m => (b -> Maybe (a, b)) -> b -> Source m a

enumFromToC :: forall m a. (Monad m, Enum a, Eq a) => a -> a -> Source m a

iterateC :: forall m a. Monad m => (a -> a) -> a -> Source m a

repeatC :: forall m a. Monad m => a -> Source m a

replicateC :: forall m a. Monad m => Int -> a -> Source m a

sourceLazy :: (Monad m, LazySequence lazy strict) => lazy -> Source m strict

repeatMC :: forall m a. Monad m => m a -> Source m a

repeatWhileMC :: forall m a. Monad m => m a -> (a -> Bool) -> Source m a

replicateMC :: forall m a. Monad m => Int -> m a -> Source m a

sourceHandle :: forall m a. (MonadIO m, IOData a) => Handle -> Source m a

sourceFile :: (MonadBaseControl IO m, MonadIO m, IOData a) => FilePath -> Source m a

sourceIOHandle :: (MonadBaseControl IO m, MonadIO m, IOData a) => IO Handle -> Source m a

stdinC :: (MonadBaseControl IO m, MonadIO m, IOData a) => Source m a

initRepeat :: Monad m => m seed -> (seed -> m a) -> Source m a

initReplicate :: Monad m => m seed -> (seed -> m a) -> Int -> Source m a

sourceRandom :: (Variate a, MonadIO m) => Source m a

sourceRandomN :: (Variate a, MonadIO m) => Int -> Source m a

sourceRandomGen :: (Variate a, MonadBase base m, PrimMonad base) => Gen (PrimState base) -> Source m a

sourceRandomNGen :: (Variate a, MonadBase base m, PrimMonad base) => Gen (PrimState base) -> Int -> Source m a

sourceDirectory :: forall m. (MonadBaseControl IO m, MonadIO m) => FilePath -> Source m FilePath

sourceDirectoryDeep :: forall m. (MonadBaseControl IO m, MonadIO m) => Bool -> FilePath -> Source m FilePath

dropC :: Monad m => Int -> Conduit a m a

dropCE :: (Monad m, IsSequence seq) => Index seq -> Conduit seq m seq

dropWhileC :: Monad m => (a -> Bool) -> Conduit a m a

dropWhileCE :: (Monad m, IsSequence seq) => (Element seq -> Bool) -> Conduit seq m seq

foldC :: (Monad m, Monoid a) => Sink a m a

foldCE :: (Monad m, MonoFoldable mono, Monoid (Element mono)) => Sink mono m (Element mono)

foldlC :: Monad m => (a -> b -> a) -> a -> Sink b m a

foldlCE :: (Monad m, MonoFoldable mono) => (a -> Element mono -> a) -> a -> Sink mono m a

foldMapC :: (Monad m, Monoid b) => (a -> b) -> Sink a m b

foldMapCE :: (Monad m, MonoFoldable mono, Monoid w) => (Element mono -> w) -> Sink mono m w

allC :: Monad m => (a -> Bool) -> Sink a m Bool

allCE :: (Monad m, MonoFoldable mono) => (Element mono -> Bool) -> Sink mono m Bool

anyC :: Monad m => (a -> Bool) -> Sink a m Bool

anyCE :: (Monad m, MonoFoldable mono) => (Element mono -> Bool) -> Sink mono m Bool

andC :: Monad m => Sink Bool m Bool

andCE :: (Monad m, MonoFoldable mono, Element mono ~ Bool) => Sink mono m Bool

orC :: Monad m => Sink Bool m Bool

orCE :: (Monad m, MonoFoldable mono, Element mono ~ Bool) => Sink mono m Bool

elemC :: (Monad m, Eq a) => a -> Sink a m Bool

elemCE :: (Monad m, EqSequence seq) => Element seq -> Sink seq m Bool

notElemC :: (Monad m, Eq a) => a -> Sink a m Bool

notElemCE :: (Monad m, EqSequence seq) => Element seq -> Sink seq m Bool

sinkLazy :: (Monad m, LazySequence lazy strict) => Sink strict m lazy

sinkList :: Monad m => Sink a m [a]

sinkVector :: (MonadBase base m, Vector v a, PrimMonad base) => Sink a m (v a)

sinkVectorN :: (MonadBase base m, Vector v a, PrimMonad base) => Int -> Sink a m (v a)

sinkBuilder :: (Monad m, Monoid builder, ToBuilder a builder) => Sink a m builder

sinkLazyBuilder :: (Monad m, Monoid builder, ToBuilder a builder, Builder builder lazy) => Sink a m lazy

sinkNull :: Monad m => Sink a m ()

awaitNonNull :: (Monad m, MonoFoldable a) => Conduit a m (Maybe (NonNull a))

headCE :: (Monad m, IsSequence seq) => Sink seq m (Maybe (Element seq))

lastC :: Monad m => Sink a m (Maybe a)

lastCE :: (Monad m, IsSequence seq) => Sink seq m (Maybe (Element seq))

lengthC :: (Monad m, Num len) => Sink a m len

lengthCE :: (Monad m, Num len, MonoFoldable mono) => Sink mono m len

lengthIfC :: (Monad m, Num len) => (a -> Bool) -> Sink a m len

lengthIfCE :: (Monad m, Num len, MonoFoldable mono) => (Element mono -> Bool) -> Sink mono m len

maximumC :: (Monad m, Ord a) => Sink a m (Maybe a)

maximumCE :: (Monad m, OrdSequence seq) => Sink seq m (Maybe (Element seq))

minimumC :: (Monad m, Ord a) => Sink a m (Maybe a)

minimumCE :: (Monad m, OrdSequence seq) => Sink seq m (Maybe (Element seq))

sumC :: (Monad m, Num a) => Sink a m a

sumCE :: (Monad m, MonoFoldable mono, Num (Element mono)) => Sink mono m (Element mono)

productC :: (Monad m, Num a) => Sink a m a

productCE :: (Monad m, MonoFoldable mono, Num (Element mono)) => Sink mono m (Element mono)

findC :: Monad m => (a -> Bool) -> Sink a m (Maybe a)

mapM_C :: Monad m => (a -> m ()) -> Sink a m ()

mapM_CE :: (Monad m, MonoFoldable mono) => (Element mono -> m ()) -> Sink mono m ()

foldMC :: Monad m => (a -> b -> m a) -> a -> Sink b m a

foldMCE :: (Monad m, MonoFoldable mono) => (a -> Element mono -> m a) -> a -> Sink mono m a

foldMapMC :: (Monad m, Monoid w) => (a -> m w) -> Sink a m w

foldMapMCE :: (Monad m, MonoFoldable mono, Monoid w) => (Element mono -> m w) -> Sink mono m w

sinkFile :: (MonadBaseControl IO m, MonadIO m, IOData a) => FilePath -> Sink a m ()

sinkHandle :: (MonadIO m, IOData a) => Handle -> Sink a m ()

sinkIOHandle :: (MonadBaseControl IO m, MonadIO m, IOData a) => IO Handle -> Sink a m ()

printC :: (Show a, MonadIO m) => Sink a m ()

stdoutC :: (MonadIO m, IOData a) => Sink a m ()

stderrC :: (MonadIO m, IOData a) => Sink a m ()

mapC :: Monad m => (a -> b) -> Conduit a m b

mapCE :: (Monad m, Functor f) => (a -> b) -> Conduit (f a) m (f b)

omapCE :: (Monad m, MonoFunctor mono) => (Element mono -> Element mono) -> Conduit mono m mono

concatMapC :: (Monad m, MonoFoldable mono) => (a -> mono) -> Conduit a m (Element mono)

concatMapCE :: (Monad m, MonoFoldable mono, Monoid w) => (Element mono -> w) -> Conduit mono m w

takeC :: Monad m => Int -> Conduit a m a

takeCE :: (Monad m, IsSequence seq) => Index seq -> Conduit seq m seq

takeWhileC :: Monad m => (a -> Bool) -> Conduit a m a

This function reads one more element than it yields, which would be a 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.

takeWhileCE :: (Monad m, IsSequence seq) => (Element seq -> Bool) -> Conduit seq m seq

takeExactlyC :: Monad m => Int -> Conduit a m b -> Conduit a m b

takeExactlyCE :: (Monad m, IsSequence a) => Index a -> Conduit a m b -> Conduit a m b

concatC :: (Monad m, MonoFoldable mono) => Conduit mono m (Element mono)

filterC :: Monad m => (a -> Bool) -> Conduit a m a

filterCE :: (IsSequence seq, Monad m) => (Element seq -> Bool) -> Conduit seq m seq

mapWhileC :: Monad m => (a -> Maybe b) -> Conduit a m b

conduitVector :: (MonadBase base m, Vector v a, PrimMonad base) => Int -> Conduit a m (v a)

scanlC :: Monad m => (a -> b -> a) -> a -> Conduit b m a

concatMapAccumC :: Monad m => (a -> accum -> (accum, [b])) -> accum -> Conduit a m b

intersperseC :: Monad m => a -> Source m a -> Source m a

encodeBase64C :: Monad m => Conduit ByteString m ByteString

decodeBase64C :: Monad m => Conduit ByteString m ByteString

encodeBase64URLC :: Monad m => Conduit ByteString m ByteString

decodeBase64URLC :: Monad m => Conduit ByteString m ByteString

encodeBase16C :: Monad m => Conduit ByteString m ByteString

decodeBase16C :: Monad m => Conduit ByteString m ByteString

mapMC :: Monad m => (a -> m b) -> Conduit a m b

mapMCE :: (Monad m, Traversable f) => (a -> m b) -> Conduit (f a) m (f b)

omapMCE :: (Monad m, MonoTraversable mono) => (Element mono -> m (Element mono)) -> Conduit mono m mono

concatMapMC :: (Monad m, MonoFoldable mono) => (a -> m mono) -> Conduit a m (Element mono)

filterMC :: Monad m => (a -> m Bool) -> Conduit a m a

filterMCE :: (Monad m, IsSequence seq) => (Element seq -> m Bool) -> Conduit seq m seq

iterMC :: Monad m => (a -> m ()) -> Conduit a m a

scanlMC :: Monad m => (a -> b -> m a) -> a -> Conduit b m a

concatMapAccumMC :: Monad m => (a -> accum -> m (accum, [b])) -> accum -> Conduit a m b

encodeUtf8C :: (Monad m, Utf8 text binary) => Conduit text m binary

decodeUtf8C :: MonadThrow m => Conduit ByteString m Text

lineC :: (Monad m, IsSequence seq, Element seq ~ Char) => Conduit seq m o -> Conduit seq m o

lineAsciiC :: (Monad m, IsSequence seq, Element seq ~ Word8) => Conduit seq m o -> Conduit seq m o

unlinesC :: (Monad m, IsSequence seq, Element seq ~ Char) => Conduit seq m seq

unlinesAsciiC :: (Monad m, IsSequence seq, Element seq ~ Word8) => Conduit seq m seq

linesUnboundedC_ :: forall m seq. (Monad m, IsSequence seq, Eq (Element seq)) => Element seq -> Conduit seq m seq

linesUnboundedC :: (Monad m, IsSequence seq, Element seq ~ Char) => Conduit seq m seq

linesUnboundedAsciiC :: (Monad m, IsSequence seq, Element seq ~ Word8) => Conduit seq m seq

zipSinks :: forall a m r r'. (MonadBaseControl IO m, MonadIO m) => Sink a m r -> Sink a m r' -> Sink a m (r, r')

Zip sinks together. This function may be used multiple times:

>>> 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"
"foo: 2"
"bar: 2"
"foo: 3"
"bar: 3"
([1,2,3],((),()))

Note that the two sinks are run concurrently, so watch out for possible race conditions if they try to interact with the same resources.

sourceMaybeMVar :: forall m a. MonadIO m => MVar (Maybe a) -> Source m a

Keep taking from an MVar (Maybe a) until it yields Nothing.

sourceMaybeTMVar :: forall a. TMVar (Maybe a) -> Source STM a

Keep taking from an TMVar (Maybe a) until it yields Nothing.

asyncC :: (MonadBaseControl IO m, Monad m) => (a -> m b) -> Conduit a m (Async (StM m b))

fromFoldM :: Monad m => FoldM m a b -> Source m a -> m b

Convert a FoldM fold abstraction into a Sink.

NOTE: This requires ImpredicativeTypes in the code that uses it.

>>> fromFoldM (FoldM ((return .) . (+)) (return 0) return) $ yieldMany [1..10]
55

toFoldM :: Monad m => Sink a m r -> (forall s. FoldM (EitherT s m) a s -> EitherT s m s) -> m r

Convert a Sink into a FoldM, passing it into a continuation.

>>> toFoldM sumC (\f -> Control.Foldl.foldM f [1..10])
55

sourceTChan :: forall a. TChan a -> Source STM a

A Source for exhausting a TChan, but blocks if it is initially empty.

sourceTQueue :: forall a. TQueue a -> Source STM a

sourceTBQueue :: forall a. TBQueue a -> Source STM a

untilMC :: forall m a. Monad m => m a -> m Bool -> Source m a

whileMC :: forall m a. Monad m => m Bool -> m a -> Source m a