conduit-1.0.13: Streaming data processing library.

Safe Haskell | None

Data.Conduit

Contents

- Core interface
  - Types
  - Connect/fuse operators
  - Primitives
  - Finalization
  - Exception handling
- Generalized conduit types
- Utility functions
- Connect-and-resume
- Flushing
- Newtype wrappers
  - ZipSource
  - ZipSink
- Convenience re-exports

Description

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.

Synopsis

- type Source m o = ConduitM () o m ()
- type Conduit i m o = ConduitM i o m ()
- type Sink i = ConduitM i Void
- data ConduitM i o m r
- ($$) :: Monad m => Source m a -> Sink a m b -> m b
- ($=) :: Monad m => Source m a -> Conduit a m b -> Source m b
- (=$) :: Monad m => Conduit a m b -> Sink b m c -> Sink a m c
- (=$=) :: Monad m => Conduit a m b -> ConduitM b c m r -> ConduitM a c m r
- await :: Monad m => Consumer i m (Maybe i)
- yield :: Monad m => o -> ConduitM i o m ()
- leftover :: i -> ConduitM i o m ()
- bracketP :: MonadResource m => IO a -> (a -> IO ()) -> (a -> ConduitM i o m r) -> ConduitM i o m r
- addCleanup :: Monad m => (Bool -> m ()) -> ConduitM i o m r -> ConduitM i o m r
- yieldOr :: Monad m => o -> m () -> ConduitM i o m ()
- catchC :: (MonadBaseControl IO m, Exception e) => ConduitM i o m r -> (e -> ConduitM i o m r) -> ConduitM i o m r
- handleC :: (MonadBaseControl IO m, Exception e) => (e -> ConduitM i o m r) -> ConduitM i o m r -> ConduitM i o m r
- tryC :: (MonadBaseControl IO m, Exception e) => ConduitM i o m r -> ConduitM i o m (Either e r)
- type Producer m o = forall i. ConduitM i o m ()
- type Consumer i m r = forall o. ConduitM i o m r
- toProducer :: Monad m => Source m a -> Producer m a
- toConsumer :: Monad m => Sink a m b -> Consumer a m b
- awaitForever :: Monad m => (i -> ConduitM i o m r) -> ConduitM i o m ()
- transPipe :: Monad m => (forall a. m a -> n a) -> ConduitM i o m r -> ConduitM i o n r
- mapOutput :: Monad m => (o1 -> o2) -> ConduitM i o1 m r -> ConduitM i o2 m r
- mapOutputMaybe :: Monad m => (o1 -> Maybe o2) -> ConduitM i o1 m r -> ConduitM i o2 m r
- mapInput :: Monad m => (i1 -> i2) -> (i2 -> Maybe i1) -> ConduitM i2 o m r -> ConduitM i1 o m r
- data ResumableSource m o
- ($$+) :: Monad m => Source m a -> Sink a m b -> m (ResumableSource m a, b)
- ($$++) :: Monad m => ResumableSource m a -> Sink a m b -> m (ResumableSource m a, b)
- ($$+-) :: Monad m => ResumableSource m a -> Sink a m b -> m b
- unwrapResumable :: MonadIO m => ResumableSource m o -> m (Source m o, m ())
- data Flush a
  - = Chunk a
  - | Flush
- newtype ZipSource m o = ZipSource {
  - getZipSource :: Source m o
  }
- sequenceSources :: (Traversable f, Monad m) => f (Source m o) -> Source m (f o)
- newtype ZipSink i m r = ZipSink {
  - getZipSink :: Sink i m r
  }
- sequenceSinks :: (Traversable f, Monad m) => f (Sink i m r) -> Sink i m (f r)
- data ResourceT m a
- class (MonadThrow m, MonadUnsafeIO m, MonadIO m, Applicative m) => MonadResource m
- class Monad m => MonadThrow m where
  - monadThrow :: Exception e => e -> m a
- class Monad m => MonadUnsafeIO m where
  - unsafeLiftIO :: IO a -> m a
- runResourceT :: MonadBaseControl IO m => ResourceT m a -> m a
- newtype ExceptionT m a = ExceptionT {
  - runExceptionT :: m (Either SomeException a)
  }
- runExceptionT_ :: Monad m => ExceptionT m a -> m a
- runException :: ExceptionT Identity a -> Either SomeException a
- runException_ :: ExceptionT Identity a -> a
- class MonadBase b m => MonadBaseControl b m | m -> b

# Core interface

## Types

type Source m o = ConduitM () o m ()Source

Provides a stream of output values, without consuming any input or producing a final result.

Since 0.5.0

type Conduit i m o = ConduitM i o m ()Source

Consumes a stream of input values and produces a stream of output values, without producing a final result.

Since 0.5.0

type Sink i = ConduitM i Void Source

Consumes a stream of input values and produces a final result, without producing any output.

     type Sink i m r = ConduitM i Void m r

Since 0.5.0

data ConduitM i o m r Source

Core datatype of the conduit package. This type represents a general component which can consume a stream of input values `i`, produce a stream of output values `o`, perform actions in the `m` monad, and produce a final result `r`. The type synonyms provided here are simply wrappers around this type.

Since 1.0.0

Instances

MonadRWS r w s m => MonadRWS r w s (ConduitM i o m) |
MonadBase base m => MonadBase base (ConduitM i o m) |
MonadError e m => MonadError e (ConduitM i o m) |
MonadReader r m => MonadReader r (ConduitM i o m) |
MonadState s m => MonadState s (ConduitM i o m) |
MonadWriter w m => MonadWriter w (ConduitM i o m) |
MFunctor (ConduitM i o) |
MonadTrans (ConduitM i o) |
Monad m => Monad (ConduitM i o m) |
Monad m => Functor (ConduitM i o m) |
Monad m => Applicative (ConduitM i o m) |
MonadIO m => MonadIO (ConduitM i o m) |
MonadResource m => MonadResource (ConduitM i o m) |
MonadThrow m => MonadThrow (ConduitM i o m) |
MonadActive m => MonadActive (ConduitM i o m) |
Monad m => Monoid (ConduitM i o m ()) |

## Connect/fuse operators

($$) :: Monad m => Source m a -> Sink a m b -> m bSource

The connect operator, which pulls data from a source and pushes to a sink. If you would like to keep the `Source` open to be used for other operations, use the connect-and-resume operator `$$+`.

Since 0.4.0

($=) :: Monad m => Source m a -> Conduit a m b -> Source m bSource

Left fuse, combining a source and a conduit together into a new source.

Both the `Source` and `Conduit` will be closed when the newly-created `Source` is closed.

Leftover data from the `Conduit` will be discarded.

Since 0.4.0

(=$) :: Monad m => Conduit a m b -> Sink b m c -> Sink a m cSource

Right fuse, combining a conduit and a sink together into a new sink.

Both the `Conduit` and `Sink` will be closed when the newly-created `Sink` is closed.

Leftover data returned from the `Sink` will be discarded.

Since 0.4.0

(=$=) :: Monad m => Conduit a m b -> ConduitM b c m r -> ConduitM a c m rSource

Fusion operator, combining two `Conduit`s together into a new `Conduit`.

Both `Conduit`s will be closed when the newly-created `Conduit` is closed.

Leftover data returned from the right `Conduit` will be discarded.

Since 0.4.0

## Primitives

await :: Monad m => Consumer i m (Maybe i)Source

Wait for a single input value from upstream. If no data is available, returns `Nothing`.

Since 0.5.0

yield Source

Arguments

:: Monad m |
=> o | output value
-> ConduitM i o m () |

Send a value downstream to the next component to consume. If the downstream component terminates, this call will never return control. If you would like to register a cleanup function, please use `yieldOr` instead.

Since 0.5.0

leftover :: i -> ConduitM i o m ()Source

Provide a single piece of leftover input to be consumed by the next component in the current monadic binding.

Note: it is highly encouraged to only return leftover values from input already consumed from upstream.

Since 0.5.0

## Finalization

bracketP :: MonadResource m => IO a -> (a -> IO ()) -> (a -> ConduitM i o m r) -> ConduitM i o m rSource

Perform some allocation and run an inner component. Two guarantees are given about resource finalization:

1. It will be prompt. The finalization will be run as early as possible.
2. It is exception safe. Due to usage of `resourcet`, the finalization will be run in the event of any exceptions.

Since 0.5.0

addCleanup :: Monad m => (Bool -> m ()) -> ConduitM i o m r -> ConduitM i o m rSource

Add some code to be run when the given component cleans up.

The supplied cleanup function will be given a `True` if the component ran to completion, or `False` if it terminated early due to a downstream component terminating.

Note that this function is not exception safe. For that, please use `bracketP`.

Since 0.4.1

yieldOr Source

Arguments

:: Monad m |
=> o |
-> m () | finalizer
-> ConduitM i o m () |

Similar to `yield`, but additionally takes a finalizer to be run if the downstream component terminates.

Since 0.5.0

## Exception handling

catchC :: (MonadBaseControl IO m, Exception e) => ConduitM i o m r -> (e -> ConduitM i o m r) -> ConduitM i o m rSource

Catch all exceptions thrown by the current component of the pipeline.

Note: this will not catch exceptions thrown by other components! For example, if an exception is thrown in a `Source` feeding to a `Sink`, and the `Sink` uses `catchC`, the exception will not be caught.

Due to this behavior (as well as lack of async exception handling), you should not try to implement combinators such as `onException` in terms of this primitive function.

Note also that the exception handling will not be applied to any finalizers generated by this conduit.

Since 1.0.11

handleC :: (MonadBaseControl IO m, Exception e) => (e -> ConduitM i o m r) -> ConduitM i o m r -> ConduitM i o m rSource

The same as `flip catchC`.

Since 1.0.11

tryC :: (MonadBaseControl IO m, Exception e) => ConduitM i o m r -> ConduitM i o m (Either e r)Source

A version of `try` for use within a pipeline. See the comments in `catchC` for more details.

Since 1.0.11

# Generalized conduit types

type Producer m o = forall i. ConduitM i o m ()Source

A component which produces a stream of output values, regardless of the input stream. A `Producer` is a generalization of a `Source`, and can be used as either a `Source` or a `Conduit`.

Since 1.0.0

type Consumer i m r = forall o. ConduitM i o m rSource

A component which consumes a stream of input values and produces a final result, regardless of the output stream. A `Consumer` is a generalization of a `Sink`, and can be used as either a `Sink` or a `Conduit`.

Since 1.0.0

toProducer :: Monad m => Source m a -> Producer m aSource

Generalize a `Source` to a `Producer`.

Since 1.0.0

toConsumer :: Monad m => Sink a m b -> Consumer a m bSource

Generalize a `Sink` to a `Consumer`.

Since 1.0.0

# Utility functions

awaitForever :: Monad m => (i -> ConduitM i o m r) -> ConduitM i o m ()Source

Wait for input forever, calling the given inner component for each piece of new input. Returns the upstream result type.

This function is provided as a convenience for the common pattern of `await`ing input, checking if it's `Just` and then looping.

Since 0.5.0

transPipe :: Monad m => (forall a. m a -> n a) -> ConduitM i o m r -> ConduitM i o n rSource

Transform the monad that a `ConduitM` lives in.

Note that the monad transforming function will be run multiple times, resulting in unintuitive behavior in some cases. For a fuller treatment, please see:

https://github.com/snoyberg/conduit/wiki/Dealing-with-monad-transformers

This function is just a synonym for `hoist`.

Since 0.4.0

mapOutput :: Monad m => (o1 -> o2) -> ConduitM i o1 m r -> ConduitM i o2 m rSource

Apply a function to all the output values of a `ConduitM`.

This mimics the behavior of `fmap` for a `Source` and `Conduit` in pre-0.4 days. It can also be simulated by fusing with the `map` conduit from Data.Conduit.List.

Since 0.4.1

mapOutputMaybe :: Monad m => (o1 -> Maybe o2) -> ConduitM i o1 m r -> ConduitM i o2 m rSource

Same as `mapOutput`, but use a function that returns `Maybe` values.

Since 0.5.0

mapInput Source

Arguments

:: Monad m |
=> (i1 -> i2) | map initial input to new input
-> (i2 -> Maybe i1) | map new leftovers to initial leftovers
-> ConduitM i2 o m r |
-> ConduitM i1 o m r |

Apply a function to all the input values of a `ConduitM`.

Since 0.5.0

# Connect-and-resume

data ResumableSource m o Source

A `Source` which has been started, but has not yet completed.

This type contains both the current state of the `Source`, and the finalizer to be run to close it.

Since 0.5.0

Instances

MFunctor ResumableSource | Since 1.0.13

($$+) :: Monad m => Source m a -> Sink a m b -> m (ResumableSource m a, b)Source

The connect-and-resume operator. This does not close the `Source`, but instead returns it to be used again. This allows a `Source` to be used incrementally in a large program, without forcing the entire program to live in the `Sink` monad.

Mnemonic: connect + do more.

Since 0.5.0

($$++) :: Monad m => ResumableSource m a -> Sink a m b -> m (ResumableSource m a, b)Source

Continue processing after usage of `$$+`.

Since 0.5.0

($$+-) :: Monad m => ResumableSource m a -> Sink a m b -> m bSource

Complete processing of a `ResumableSource`. This will run the finalizer associated with the `ResumableSource`. In order to guarantee process resource finalization, you must use this operator after using `$$+` and `$$++`.

Since 0.5.0

unwrapResumable :: MonadIO m => ResumableSource m o -> m (Source m o, m ())Source

Unwraps a `ResumableSource` into a `Source` and a finalizer.

A `ResumableSource` represents a `Source` which has already been run, and therefore has a finalizer registered. As a result, if we want to turn it into a regular `Source`, we need to ensure that the finalizer will be run appropriately. By appropriately, I mean:

- If a new finalizer is registered, the old one should not be called.
- If the old one is called, it should not be called again.

This function returns both a `Source` and a finalizer which ensures that the above two conditions hold. Once you call that finalizer, the `Source` is invalidated and cannot be used.

Since 0.5.2

# Flushing

data Flush a Source

Provide for a stream of data that can be flushed.

A number of `Conduit`s (e.g., zlib compression) need the ability to flush the stream at some point. This provides a single wrapper datatype to be used in all such circumstances.

Since 0.3.0

Constructors

Chunk a |
Flush |

Instances

Functor Flush |
Eq a => Eq (Flush a) |
Ord a => Ord (Flush a) |
Show a => Show (Flush a) |

# Newtype wrappers

## ZipSource

newtype ZipSource m o Source

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

Constructors

ZipSource |

Fields

getZipSource :: Source m o

Instances

Monad m => Functor (ZipSource m) |
Monad m => Applicative (ZipSource m) |

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

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

## ZipSink

newtype ZipSink i m r Source

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

Constructors

ZipSink |

Fields

getZipSink :: Sink i m r

Instances

Monad m => Functor (ZipSink i m) |
Monad m => Applicative (ZipSink i m) |

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

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

# Convenience re-exports

data ResourceT m a

The Resource transformer. This transformer keeps track of all registered actions, and calls them upon exit (via `runResourceT`). Actions may be registered via `register`, or resources may be allocated atomically via `allocate`. `allocate` corresponds closely to `bracket`.

Releasing may be performed before exit via the `release` function. This is a highly recommended optimization, as it will ensure that scarce resources are freed early. Note that calling `release` will deregister the action, so that a release action will only ever be called once.

Since 0.3.0

Instances

MFunctor ResourceT | Since 0.4.7
MMonad ResourceT | Since 0.4.7
MonadTrans ResourceT |
MonadTransControl ResourceT |
MonadRWS r w s m => MonadRWS r w s (ResourceT m) |
MonadBase b m => MonadBase b (ResourceT m) |
MonadBaseControl b m => MonadBaseControl b (ResourceT m) |
MonadError e m => MonadError e (ResourceT m) |
MonadReader r m => MonadReader r (ResourceT m) |
MonadState s m => MonadState s (ResourceT m) |
MonadWriter w m => MonadWriter w (ResourceT m) |
Monad m => Monad (ResourceT m) |
Functor m => Functor (ResourceT m) |
Typeable1 m => Typeable1 (ResourceT m) |
Applicative m => Applicative (ResourceT m) |
MonadIO m => MonadIO (ResourceT m) |
MonadCont m => MonadCont (ResourceT m) |
(MonadThrow m, MonadUnsafeIO m, MonadIO m, Applicative m) => MonadResource (ResourceT m) |
MonadThrow m => MonadThrow (ResourceT m) |
(MonadIO m, MonadActive m) => MonadActive (ResourceT m) |

class (MonadThrow m, MonadUnsafeIO m, MonadIO m, Applicative m) => MonadResource m

A `Monad` which allows for safe resource allocation. In theory, any monad transformer stack included a `ResourceT` can be an instance of `MonadResource`.

Note: `runResourceT` has a requirement for a `MonadBaseControl IO m` monad, which allows control operations to be lifted. A `MonadResource` does not have this requirement. This means that transformers such as `ContT` can be an instance of `MonadResource`. However, the `ContT` wrapper will need to be unwrapped before calling `runResourceT`.

Since 0.3.0

Instances

MonadResource m => MonadResource (ListT m) |
(MonadThrow m, MonadUnsafeIO m, MonadIO m, Applicative m) => MonadResource (ResourceT m) |
MonadResource m => MonadResource (ExceptionT m) |
MonadResource m => MonadResource (MaybeT m) |
MonadResource m => MonadResource (IdentityT m) |
MonadResource m => MonadResource (ContT r m) |
(Error e, MonadResource m) => MonadResource (ErrorT e m) |
MonadResource m => MonadResource (ReaderT r m) |
MonadResource m => MonadResource (StateT s m) |
MonadResource m => MonadResource (StateT s m) |
(Monoid w, MonadResource m) => MonadResource (WriterT w m) |
(Monoid w, MonadResource m) => MonadResource (WriterT w m) |
MonadResource m => MonadResource (ConduitM i o m) |
(Monoid w, MonadResource m) => MonadResource (RWST r w s m) |
(Monoid w, MonadResource m) => MonadResource (RWST r w s m) |
MonadResource m => MonadResource (Pipe l i o u m) |

class Monad m => MonadThrow m where

A `Monad` which can throw exceptions. Note that this does not work in a vanilla `ST` or `Identity` monad. Instead, you should use the `ExceptionT` transformer in your stack if you are dealing with a non-`IO` base monad.

Since 0.3.0

Methods

monadThrow :: Exception e => e -> m a

Instances

MonadThrow [] |
MonadThrow IO |
MonadThrow Maybe |
MonadThrow (Either SomeException) |
MonadThrow m => MonadThrow (ListT m) |
MonadThrow m => MonadThrow (ResourceT m) |
Monad m => MonadThrow (ExceptionT m) |
MonadThrow m => MonadThrow (MaybeT m) |
MonadThrow m => MonadThrow (IdentityT m) |
MonadThrow m => MonadThrow (ContT r m) |
(Error e, MonadThrow m) => MonadThrow (ErrorT e m) |
MonadThrow m => MonadThrow (ReaderT r m) |
MonadThrow m => MonadThrow (StateT s m) |
MonadThrow m => MonadThrow (StateT s m) |
(Monoid w, MonadThrow m) => MonadThrow (WriterT w m) |
(Monoid w, MonadThrow m) => MonadThrow (WriterT w m) |
MonadThrow m => MonadThrow (ConduitM i o m) |
(Monoid w, MonadThrow m) => MonadThrow (RWST r w s m) |
(Monoid w, MonadThrow m) => MonadThrow (RWST r w s m) |
MonadThrow m => MonadThrow (Pipe l i o u m) |

class Monad m => MonadUnsafeIO m where

A `Monad` based on some monad which allows running of some `IO` actions, via unsafe calls. This applies to `IO` and `ST`, for instance.

Since 0.3.0

Methods

unsafeLiftIO :: IO a -> m a

Instances

MonadUnsafeIO IO |
(MonadTrans t, MonadUnsafeIO m, Monad (t m)) => MonadUnsafeIO (t m) |
MonadUnsafeIO (ST s) |
MonadUnsafeIO (ST s) |

runResourceT :: MonadBaseControl IO m => ResourceT m a -> m a

Unwrap a `ResourceT` transformer, and call all registered release actions.

Note that there is some reference counting involved due to `resourceForkIO`. If multiple threads are sharing the same collection of resources, only the last call to `runResourceT` will deallocate the resources.

Since 0.3.0

newtype ExceptionT m a

The express purpose of this transformer is to allow non-`IO`-based monad stacks to catch exceptions via the `MonadThrow` typeclass.

Since 0.3.0

Constructors

ExceptionT |

Fields

runExceptionT :: m (Either SomeException a)

Instances

MonadTrans ExceptionT |
MonadTransControl ExceptionT |
MonadRWS r w s m => MonadRWS r w s (ExceptionT m) |
MonadBase b m => MonadBase b (ExceptionT m) |
MonadBaseControl b m => MonadBaseControl b (ExceptionT m) |
MonadError e m => MonadError e (ExceptionT m) |
MonadReader r m => MonadReader r (ExceptionT m) |
MonadState s m => MonadState s (ExceptionT m) |
MonadWriter w m => MonadWriter w (ExceptionT m) |
Monad m => Monad (ExceptionT m) |
Monad m => Functor (ExceptionT m) |
Monad m => Applicative (ExceptionT m) |
MonadIO m => MonadIO (ExceptionT m) |
MonadCont m => MonadCont (ExceptionT m) |
MonadResource m => MonadResource (ExceptionT m) |
Monad m => MonadThrow (ExceptionT m) |

runExceptionT_ :: Monad m => ExceptionT m a -> m a

Same as `runExceptionT`, but immediately `throw` any exception returned.

Since 0.3.0

runException :: ExceptionT Identity a -> Either SomeException a

Run an `ExceptionT Identity` stack.

Since 0.4.2

runException_ :: ExceptionT Identity a -> a

Run an `ExceptionT Identity` stack, but immediately `throw` any exception returned.

Since 0.4.2

class MonadBase b m => MonadBaseControl b m | m -> b

Instances

MonadBaseControl [] [] |
MonadBaseControl IO IO |
MonadBaseControl STM STM |
MonadBaseControl Maybe Maybe |
MonadBaseControl Identity Identity |
MonadBaseControl b m => MonadBaseControl b (MaybeT m) |
MonadBaseControl b m => MonadBaseControl b (ListT m) |
MonadBaseControl b m => MonadBaseControl b (IdentityT m) |
MonadBaseControl b m => MonadBaseControl b (ResourceT m) |
MonadBaseControl b m => MonadBaseControl b (ExceptionT m) |
(Monoid w, MonadBaseControl b m) => MonadBaseControl b (WriterT w m) |
(Monoid w, MonadBaseControl b m) => MonadBaseControl b (WriterT w m) |
MonadBaseControl b m => MonadBaseControl b (StateT s m) |
MonadBaseControl b m => MonadBaseControl b (StateT s m) |
MonadBaseControl b m => MonadBaseControl b (ReaderT r m) |
(Error e, MonadBaseControl b m) => MonadBaseControl b (ErrorT e m) |
(Monoid w, MonadBaseControl b m) => MonadBaseControl b (RWST r w s m) |
(Monoid w, MonadBaseControl b m) => MonadBaseControl b (RWST r w s m) |
MonadBaseControl ((->) r) ((->) r) |
MonadBaseControl (Either e) (Either e) |
MonadBaseControl (ST s) (ST s) |
MonadBaseControl (ST s) (ST s) |