Data.Conduit.Utils

This module provide different utility functions that allow to use safe higher level usage.

Conduit pairs allow creation of an internal datastructure that acts as a bridge, and provides input and output conduits. The structure itself is hidden internally and can't be used directly, this provide an additional safeness.

In order to create a bridge from your own datastructures you need to do the following:

• Make it an instance of UnboundedStream or BoundedStream depending on it's properties:

instance BoundedStream (Proxy2 TBMQueue) TBMQueue where
 mkBStream _ i = atomically $ newTBMQueue i

• Add IsConduit instance.

 instance MonadIO m => IsConduit m TBMQueue where
   mkSource = sourceTBMQueue
   mkSink   = flip sinkTBMQueue True

• Use "pair" or "pairBounded" to create a bridge. Because bridge data structure is hidden and not seen in parameters, we need proxy type to help compiler to choose type, we use Proxy2 for that.

 pairTBMQueue = pairBounded (proxy2 :: Proxy2 TBMQueue a)

• Now we can create a pair of conduits:

(src, snk) <- pairTBMQueue 32
Control.Concurrent.Async.concurrently (sender src) (receviver snk)

As channel is not visible we can close it or send additional messages bypassing conduit code.

This package provides predefined pairs for all STM types that are used in the package.

Conduit pairs

Low level functions

pairBounded

Arguments

:: (MonadIO m, IsConduit m o, BoundedStream i o)

=> i a

Type description.

-> Int

Conduit size.

-> m (Source m a, Sink a m ())

Create bounded conduit pair, see BoundedStream class description.

pair

Arguments

:: (MonadIO m, IsConduit m o, UnboundedStream i o)

=> i a

Type description.

-> m (Source m a, Sink a m ())

Create unbounded pair, see UnboundedStream class description.

Classes

class UnboundedStream i o | i -> o where

Class for structures that can handle unbounded stream of values. Such streams break conduit assumptions that constant memory will be used, because if receiver is slower then sender than values will be accumulated.

Methods

mkUStream :: i a -> IO (o a)

Instances

UnboundedStream (Proxy2 TQueue) TQueue

UnboundedStream (Proxy2 TMChan) TMChan

UnboundedStream (Proxy2 TMQueue) TMQueue

class BoundedStream i o | i -> o where

Class for structures that can handle bounded stream of values i.e. there is exists Int value that sets an upper limit on the number of values that can be handled by structure. Exact meaning of this limit may depend on the carrier type.

Methods

mkBStream :: i a -> Int -> IO (o a)

Instances

BoundedStream (Proxy2 TBQueue) TBQueue

BoundedStream (Proxy2 TBMChan) TBMChan

BoundedStream (Proxy2 TBMQueue) TBMQueue

class MonadIO m => IsConduit m x where

Class that describes how we can make conduit out of the carrier value.

Methods

mkSink :: x a -> Sink a m ()

mkSource :: x a -> Source m a

Instances

MonadIO m => IsConduit m TMChan

MonadIO m => IsConduit m TBMChan

MonadIO m => IsConduit m TQueue

MonadIO m => IsConduit m TMQueue

MonadIO m => IsConduit m TBMQueue

MonadIO m => IsConduit m TBQueue

Types

data Proxy2 a b

Proxy type that can be used to create opaque values.

This proxy type is required because pair hides internal data structure and proxy is used to help compiler infer internal type.

Instances

BoundedStream (Proxy2 TBQueue) TBQueue

BoundedStream (Proxy2 TBMChan) TBMChan

BoundedStream (Proxy2 TBMQueue) TBMQueue

UnboundedStream (Proxy2 TQueue) TQueue

UnboundedStream (Proxy2 TMChan) TMChan

UnboundedStream (Proxy2 TMQueue) TMQueue

proxy2 :: Proxy2 a b

Construct Proxy2 value.

(proxy2 :: Proxy2 TChan a)

Specialized functions

List of specialized functions, that can create a bridges over STM types, where *B* stands for bounded *M* stands for closable. If data structure is not closable then there is no way to notify receiver side that bridge is closed, so it's possible to use it only in infinite streams of when some other mechanism of notification is used.

pairTQueue :: MonadIO m => m (Source m a, Sink a m ())

pairTMQueue :: MonadIO m => m (Source m a, Sink a m ())

pairTMChan :: MonadIO m => m (Source m a, Sink a m ())

pairTBQueue :: MonadIO m => Int -> m (Source m a, Sink a m ())

pairTBMQueue :: MonadIO m => Int -> m (Source m a, Sink a m ())

pairTBMChan :: MonadIO m => Int -> m (Source m a, Sink a m ())