EventBus.hs

module Data.CQRS.Internal.EventBus
    ( EventBus(..)
    , Subscription(..)
    , newEventBus
    ) where

-- External imports
import           Control.Concurrent.STM (atomically)
import qualified Control.Concurrent.STM.TChan as C
import           Control.DeepSeq (NFData, ($!!))
import           Control.Monad.IO.Class (MonadIO, liftIO)

-- Internal imports
import           Data.CQRS.PersistedEvent

-- | Event bus that broadcasts events of a certain type.
data EventBus e = EventBus
    { publishEventsToBus :: [PersistedEvent e] -> IO ()
    , subscribeToEventBus :: IO (Subscription e)
    , unsubscribeFromEventBus :: Subscription e -> IO ()
    }

-- | A subscription to an event bus.
newtype Subscription e = Subscription
    { readEventFromSubscription :: IO [PersistedEvent e]
    }

-- | Create a new event bus.
newEventBus :: (MonadIO m, Show e, NFData e) => m (EventBus e)
newEventBus = do
  c <- liftIO $ atomically $ C.newBroadcastTChan
  let publish events = do
        atomically $ C.writeTChan c $!! events
      subscribe = do
        c' <- atomically $ C.dupTChan c
        return $ Subscription (atomically $ C.readTChan c')
      unsubscribe _ = do
        -- TODO: Should we set a flag to detect post-unsubscribe use?
        return () -- Needs no cleanup.
  return $ EventBus publish subscribe unsubscribe