stm-conduit-2.8.0: Introduces conduits to channels, and promotes using conduits concurrently.

Safe HaskellSafe




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:

instance BoundedStream (Proxy2 TBMQueue) TBMQueue where
 mkBStream _ i = atomically $ newTBMQueue i
 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 Source


:: (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 Source


:: (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.


class UnboundedStream i o | i -> o where Source

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.


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

class BoundedStream i o | i -> o where Source

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.


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

class MonadIO m => IsConduit m x where Source

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


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

mkSource :: x a -> Source m a Source


data Proxy2 a b Source

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.

proxy2 :: Proxy2 a b Source

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

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

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

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

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

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