Control.Concurrent.STM.Firehose

A subscription based messaging system, with non-blocking bounded write.

fh <- atomically newFirehose
-- the following line will not block, even though nobody subscribed
atomically (mapM_ (writeEvent fh) [1..100])
-- let's subscribe a single client
sub <- atomically (subscribe 10 fh)
forkIO (forever (atomically (readEvent sub) >>= print))
atomically $ writeEvent fh 1

Documentation

data Firehose a

data Subscription a

newFirehose :: STM (Firehose a)

Creates a new Firehose item.

writeEvent :: Firehose a -> a -> STM ()

Sends a piece of data in the fire hose.

subscribe

Arguments

:: Int

Number of elements buffered. If set too high, it will increase memory usage. If set too low, activity spikes will result in message loss.

-> Firehose a

-> STM (Subscription a)

Get a subscription from the fire hose, that will be used to read events.

unsubscribe :: Subscription a -> STM ()

Unsubscribe from the fire hose. Subsequent calls to readEvent will return Nothing. This runs in O(n), where n is the current number of subscriptions. Please contact the maintainer if you need better performance.

readEvent :: Subscription a -> STM (Maybe a)

Read an event from a Subscription. This will return Nothing if the firehose is shut, or the subscription removed.

getQueue :: Subscription a -> TBMQueue a

Gets the underlying queue from a subscription.