pipes-concurrency-1.1.0: Concurrency for the pipes ecosystem

Safe HaskellTrustworthy




Asynchronous communication between proxies


Spawn mailboxes

spawn :: Size -> IO (Input a, Output a)Source

Spawn a mailbox of the specified Size that has an Input and Output end

data Size Source

Size specifies how many messages to store in the mailbox before send blocks.



Store an Unbounded number of messages

Bounded Int

Store a Bounded number of messages specified by the Int argument


Store only a Single message (like Bounded 1, but more efficient)

data Input a Source

Accepts messages for the mailbox

data Output a Source

Retrieves messages from the mailbox

Send and receive messages

send :: Input a -> a -> STM BoolSource

Send a message to the mailbox

  • Fails and returns False if the mailbox's Output has been garbage collected (even if the mailbox is not full), otherwise it:
  • Retries if the mailbox is full, or:
  • Succeeds if the mailbox is not full and returns True.

recv :: Output a -> STM (Maybe a)Source

Receive a message from the mailbox

  • Succeeds and returns a Just if the mailbox is not empty, otherwise it:
  • Retries if mailbox's Input has not been garbage collected, or:
  • Fails if the mailbox's Input has been garbage collected and returns Nothing.

Proxy utilities

sendD :: Proxy p => Input a -> x -> p x a x a IO ()Source

Writes all messages flowing 'D'ownstream to the given Input

sendD terminates when the corresponding Output is garbage collected.

recvS :: Proxy p => Output a -> () -> Producer p a IO ()Source

Convert an Output to a Producer

recvS terminates when the corresponding Input is garbage collected.


Control.Concurrent re-exports forkIO, although I recommend using the async library instead.

Control.Concurrent.STM re-exports atomically and STM.

System.Mem re-exports performGC.

module System.Mem