Scatter.hs

{-|
Module:             STM.Scatter
Description:        Broadcast channel with multiple listeners and memory safety.
Copyright:          © 2017 All rights reserved.
License:            GPL-3
Maintainer:         Evan Cofsky <evan@theunixman.com>
Stability:          experimental
Portability:        POSIX
-}

module STM.Scatter (
    Scatter,
    scatter,
    scatterMsg,

    Gather,
    gather,
    gatherMsg
    )where

import Lawless
import STM.Base
import Control.Concurrent.STM.TChan

-- * The 'Scatter'

-- | A 'Scatter' is a single source that can broadcast messages
-- to any number of subscribers.
newtype Scatter a = Scatter {unScatter ∷ TChan a}

-- | Creates a new 'Scatter'.
scatter ∷ MonadBase IO m ⇒ m (Scatter a)
scatter = liftBase $ Scatter <$> newTChanIO

-- | Post a message to a 'Scatter' to all the subscribed 'Gather's.
scatterMsg ∷ MonadBase IO m ⇒ Scatter a → a → m ()
scatterMsg s a = atomically $ writeTChan (unScatter s) a

-- * The 'Gather'

-- | A 'Gather' receives copies of all messages posted to the
-- 'Scatter' it was derived from.
newtype Gather a = Gather {unGather ∷ TChan a}

-- | Create a new 'Gather' from a given 'Scatter'.
gather ∷ MonadBase IO m ⇒ Getter (Scatter a) (m (Gather a))
gather = to $ \s → Gather <$> (atomically ∘ dupTChan ∘ unScatter) s

-- | Read a message posted to a 'Gather' by a 'Scatter'.
gatherMsg ∷ MonadBase IO m ⇒ Gather a → m a
gatherMsg g = atomically $ readTChan (unGather g)