conduit-1.0.13: Streaming data processing library.

Safe Haskell | None

Data.Conduit.Internal

Contents

- Types
- Primitives
- Finalization
- Composition
- Generalizing
- Exceptions
- Utilities

Synopsis

- data Pipe l i o u m r
  - = HaveOutput (Pipe l i o u m r) (m ()) o
  - | NeedInput (i -> Pipe l i o u m r) (u -> Pipe l i o u m r)
  - | Done r
  - | PipeM (m (Pipe l i o u m r))
  - | Leftover (Pipe l i o u m r) l
- newtype ConduitM i o m r = ConduitM {
  - unConduitM :: Pipe i i o () m r
  }
- type Source m o = ConduitM () o m ()
- type Producer m o = forall i. ConduitM i o m ()
- type Sink i = ConduitM i Void
- type Consumer i m r = forall o. ConduitM i o m r
- type Conduit i m o = ConduitM i o m ()
- data ResumableSource m o = ResumableSource (Source m o) (m ())
- await :: Pipe l i o u m (Maybe i)
- awaitE :: Pipe l i o u m (Either u i)
- awaitForever :: Monad m => (i -> Pipe l i o r m r') -> Pipe l i o r m r
- yield :: Monad m => o -> Pipe l i o u m ()
- yieldOr :: Monad m => o -> m () -> Pipe l i o u m ()
- leftover :: l -> Pipe l i o u m ()
- bracketP :: MonadResource m => IO a -> (a -> IO ()) -> (a -> Pipe l i o u m r) -> Pipe l i o u m r
- addCleanup :: Monad m => (Bool -> m ()) -> Pipe l i o u m r -> Pipe l i o u m r
- idP :: Monad m => Pipe l a a r m r
- pipe :: Monad m => Pipe l a b r0 m r1 -> Pipe Void b c r1 m r2 -> Pipe l a c r0 m r2
- pipeL :: Monad m => Pipe l a b r0 m r1 -> Pipe b b c r1 m r2 -> Pipe l a c r0 m r2
- connectResume :: Monad m => ResumableSource m o -> Sink o m r -> m (ResumableSource m o, r)
- runPipe :: Monad m => Pipe Void () Void () m r -> m r
- injectLeftovers :: Monad m => Pipe i i o u m r -> Pipe l i o u m r
- (>+>) :: Monad m => Pipe l a b r0 m r1 -> Pipe Void b c r1 m r2 -> Pipe l a c r0 m r2
- (<+<) :: Monad m => Pipe Void b c r1 m r2 -> Pipe l a b r0 m r1 -> Pipe l a c r0 m r2
- sourceToPipe :: Monad m => Source m o -> Pipe l i o u m ()
- sinkToPipe :: Monad m => Sink i m r -> Pipe l i o u m r
- conduitToPipe :: Monad m => Conduit i m o -> Pipe l i o u m ()
- toProducer :: Monad m => Source m a -> Producer m a
- toConsumer :: Monad m => Sink a m b -> Consumer a m b
- catchP :: (MonadBaseControl IO m, Exception e) => Pipe l i o u m r -> (e -> Pipe l i o u m r) -> Pipe l i o u m r
- handleP :: (MonadBaseControl IO m, Exception e) => (e -> Pipe l i o u m r) -> Pipe l i o u m r -> Pipe l i o u m r
- tryP :: (MonadBaseControl IO m, Exception e) => Pipe l i o u m r -> Pipe l i o u m (Either e r)
- 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)
- transPipe :: Monad m => (forall a. m a -> n a) -> Pipe l i o u m r -> Pipe l i o u n r
- mapOutput :: Monad m => (o1 -> o2) -> Pipe l i o1 u m r -> Pipe l i o2 u m r
- mapOutputMaybe :: Monad m => (o1 -> Maybe o2) -> Pipe l i o1 u m r -> Pipe l i o2 u m r
- mapInput :: Monad m => (i1 -> i2) -> (l2 -> Maybe l1) -> Pipe l2 i2 o u m r -> Pipe l1 i1 o u m r
- sourceList :: Monad m => [a] -> Pipe l i a u m ()
- withUpstream :: Monad m => Pipe l i o u m r -> Pipe l i o u m (u, r)
- unwrapResumable :: MonadIO m => ResumableSource m o -> m (Source m o, m ())
- enumFromTo :: (Enum o, Eq o, Monad m) => o -> o -> Pipe l i o u m ()
- zipSinks :: Monad m => Sink i m r -> Sink i m r' -> Sink i m (r, r')
- zipSources :: Monad m => Source m a -> Source m b -> Source m (a, b)
- zipSourcesApp :: Monad m => Source m (a -> b) -> Source m a -> Source m b

# Types

data Pipe l i o u m r Source

The underlying datatype for all the types in this package. In has six type parameters:

- l is the type of values that may be left over from this `Pipe`. A `Pipe` with no leftovers would use `Void` here, and one with leftovers would use the same type as the i parameter. Leftovers are automatically provided to the next `Pipe` in the monadic chain.
- i is the type of values for this `Pipe`'s input stream.
- o is the type of values for this `Pipe`'s output stream.
- u is the result type from the upstream `Pipe`.
- m is the underlying monad.
- r is the result type.

A basic intuition is that every `Pipe` produces a stream of output values (o), and eventually indicates that this stream is terminated by sending a result (r). On the receiving end of a `Pipe`, these become the i and u parameters.

Since 0.5.0

Constructors

HaveOutput (Pipe l i o u m r) (m ()) o | Provide new output to be sent downstream. This constructor has three fields: the next `Pipe` to be used, a finalization function, and the output value.
NeedInput (i -> Pipe l i o u m r) (u -> Pipe l i o u m r) | Request more input from upstream. The first field takes a new input value and provides a new `Pipe`. The second takes an upstream result value, which indicates that upstream is producing no more results.
Done r | Processing with this `Pipe` is complete, providing the final result.
PipeM (m (Pipe l i o u m r)) | Require running of a monadic action to get the next `Pipe`.
Leftover (Pipe l i o u m r) l | Return leftover input, which should be provided to future operations.

Instances

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

newtype 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

Constructors

ConduitM |

Fields

unConduitM :: Pipe i i o () m r

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 ()) |

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

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

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

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

Constructors

ResumableSource (Source m o) (m ()) |

Instances

MFunctor ResumableSource | Since 1.0.13

# Primitives

await :: Pipe l i o u m (Maybe i)Source

Wait for a single input value from upstream.

Since 0.5.0

awaitE :: Pipe l i o u m (Either u i)Source

This is similar to `await`, but will return the upstream result value as `Left` if available.

Since 0.5.0

awaitForever :: Monad m => (i -> Pipe l i o r m r') -> Pipe l i o r m rSource

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

Since 0.5.0

yield Source

Arguments

:: Monad m |
=> o | output value
-> Pipe l i o u m () |

Send a single output value downstream. If the downstream `Pipe` terminates, this `Pipe` will terminate as well.

Since 0.5.0

yieldOr Source

Arguments

:: Monad m |
=> o |
-> m () | finalizer
-> Pipe l i o u m () |

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

Since 0.5.0

leftover :: l -> Pipe l i o u m ()Source

Provide a single piece of leftover input to be consumed by the next pipe 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 -> Pipe l i o u m r) -> Pipe l i o u m rSource

Perform some allocation and run an inner `Pipe`. 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 Source

Arguments

:: Monad m |
=> (Bool -> m ()) | `True` if `Pipe` ran to completion, `False` for early termination.
-> Pipe l i o u m r |
-> Pipe l i o u m r |

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

Since 0.4.1

# Composition

idP :: Monad m => Pipe l a a r m rSource

The identity `Pipe`.

Since 0.5.0

pipe :: Monad m => Pipe l a b r0 m r1 -> Pipe Void b c r1 m r2 -> Pipe l a c r0 m r2Source

Compose a left and right pipe together into a complete pipe. The left pipe will be automatically closed when the right pipe finishes.

Since 0.5.0

pipeL :: Monad m => Pipe l a b r0 m r1 -> Pipe b b c r1 m r2 -> Pipe l a c r0 m r2Source

Same as `pipe`, but automatically applies `injectLeftovers` to the right `Pipe`.

Since 0.5.0

connectResume :: Monad m => ResumableSource m o -> Sink o m r -> m (ResumableSource m o, r)Source

Connect a `Source` to a `Sink` until the latter closes. Returns both the most recent state of the `Source` and the result of the `Sink`.

We use a `ResumableSource` to keep track of the most recent finalizer provided by the `Source`.

Since 0.5.0

runPipe :: Monad m => Pipe Void () Void () m r -> m rSource

Run a pipeline until processing completes.

Since 0.5.0

injectLeftovers :: Monad m => Pipe i i o u m r -> Pipe l i o u m rSource

Transforms a `Pipe` that provides leftovers to one which does not, allowing it to be composed.

This function will provide any leftover values within this `Pipe` to any calls to `await`. If there are more leftover values than are demanded, the remainder are discarded.

Since 0.5.0

(>+>) :: Monad m => Pipe l a b r0 m r1 -> Pipe Void b c r1 m r2 -> Pipe l a c r0 m r2Source

Fuse together two `Pipe`s, connecting the output from the left to the input of the right.

Notice that the leftover parameter for the `Pipe`s must be `Void`. This ensures that there is no accidental data loss of leftovers during fusion. If you have a `Pipe` with leftovers, you must first call `injectLeftovers`.

Since 0.5.0

(<+<) :: Monad m => Pipe Void b c r1 m r2 -> Pipe l a b r0 m r1 -> Pipe l a c r0 m r2Source

Same as `>+>`, but reverse the order of the arguments.

Since 0.5.0

# Generalizing

sourceToPipe :: Monad m => Source m o -> Pipe l i o u m ()Source

sinkToPipe :: Monad m => Sink i m r -> Pipe l i o u m rSource

conduitToPipe :: Monad m => Conduit i m o -> Pipe l i o u m ()Source

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

# Exceptions

catchP :: (MonadBaseControl IO m, Exception e) => Pipe l i o u m r -> (e -> Pipe l i o u m r) -> Pipe l i o u m rSource

See `catchC` for more details.

Since 1.0.11

handleP :: (MonadBaseControl IO m, Exception e) => (e -> Pipe l i o u m r) -> Pipe l i o u m r -> Pipe l i o u m rSource

The same as `flip catchP`.

Since 1.0.11

tryP :: (MonadBaseControl IO m, Exception e) => Pipe l i o u m r -> Pipe l i o u m (Either e r)Source

See `tryC` for more details.

Since 1.0.11

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

# Utilities

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

Transform the monad that a `Pipe` 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) -> Pipe l i o1 u m r -> Pipe l i o2 u m rSource

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

This mimics the behavior of `fmap` for a `Source` and `Conduit` in pre-0.4 days.

Since 0.4.1

mapOutputMaybe :: Monad m => (o1 -> Maybe o2) -> Pipe l i o1 u m r -> Pipe l i o2 u 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
-> (l2 -> Maybe l1) | map new leftovers to initial leftovers
-> Pipe l2 i2 o u m r |
-> Pipe l1 i1 o u m r |

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

Since 0.5.0

sourceList :: Monad m => [a] -> Pipe l i a u m ()Source

Convert a list into a source.

Since 0.3.0

withUpstream :: Monad m => Pipe l i o u m r -> Pipe l i o u m (u, r)Source

Returns a tuple of the upstream and downstream results. Note that this will force consumption of the entire input stream.

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

enumFromTo :: (Enum o, Eq o, Monad m) => o -> o -> Pipe l i o u m ()Source

zipSinks :: Monad m => Sink i m r -> Sink i m r' -> Sink i m (r, r')Source

Combines two sinks. The new sink will complete when both input sinks have completed.

Any leftovers are discarded.

Since 0.4.1

zipSources :: Monad m => Source m a -> Source m b -> Source m (a, b)Source

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 bSource

Combines two sources. The new source will stop producing once either source has been exhausted.

Since 1.0.13