TransactionalQueue.hs

module Queue.TransactionalQueue
  ( newQueue
  ) where

import           Control.Concurrent.STM
import qualified Control.Monad
import           Queue.API              (Queue (Queue), QueueItem)

data TransactionalQueue = TransactionalQueue
                            { transactionalQueueItems  :: TBQueue QueueItem
                            , transactionalQueueClosed :: TMVar ()
                            }

newQueue :: IO Queue
newQueue = do
  queueItems <- newTBQueueIO 1000
  queueClosed <- newEmptyTMVarIO
  let transactionalQueue =
        TransactionalQueue
          { transactionalQueueItems = queueItems
          , transactionalQueueClosed = queueClosed
          }
  pure $
    Queue
      (enqueue transactionalQueue)
      (dequeue transactionalQueue)
      (close transactionalQueue)
      (open transactionalQueue)

enqueue :: TransactionalQueue -> QueueItem -> IO (Either String ())
enqueue transactionalQueue item = do
  isOpen <- atomically $ isEmptyTMVar (transactionalQueueClosed transactionalQueue)
  if isOpen
    then atomically $ do
      writeTBQueue (transactionalQueueItems transactionalQueue) item
      pure (Right ())
    else pure (Left "Queue is closed")

dequeue :: TransactionalQueue -> IO (Either String QueueItem)
dequeue transactionalQueue =
  atomically $
    (Right <$> readTBQueue (transactionalQueueItems transactionalQueue))
    `orElse`
    (do
      isOpen <- isEmptyTMVar (transactionalQueueClosed transactionalQueue)
      check (not isOpen)
      pure (Left "Queue is closed"))

close :: TransactionalQueue -> IO ()
close transactionalQueue = atomically $ do
  _ <- tryTakeTMVar (transactionalQueueClosed transactionalQueue)
  putTMVar (transactionalQueueClosed transactionalQueue) ()

open :: TransactionalQueue -> IO ()
open transactionalQueue =
  Control.Monad.void (atomically (tryTakeTMVar (transactionalQueueClosed transactionalQueue)))