MessageQueue.hs

{-# LANGUAGE BangPatterns #-}
{-# LANGUAGE CPP #-}
{-# LANGUAGE TemplateHaskell #-}
{-# LANGUAGE Trustworthy #-}
-- For Empty and :<
{-# OPTIONS_GHC -Wno-incomplete-patterns #-}

-- |
-- Module      :  Games.ECS.MessageQueue
-- Description : A simple message queue implementation.
-- Copyright   :  (C) 2020 Sophie Taylor
-- License     :  AGPL-3.0-or-later
-- Maintainer  :  Sophie Taylor <sophie@spacekitteh.moe>
-- Stability   :  experimental
-- Portability: GHC
--
-- Implements a simple message queuing system.
module Games.ECS.MessageQueue
  ( MessageQueue,
    --    queue,
    newMessageQueue,
    MonadAtomicMessageQueue (..),
    queueMessage,
    queueImmediateMessage,
    queueMessages,
    readMessage,
    processMessageQueue,
    MessagesProcessed (..),
  )
where

import Control.Lens
import Control.Monad
import Data.Hashable
import Data.Sequence (Seq, singleton)
import Data.Tuple (swap)
import GHC.Generics
import GHC.Types (SPEC (..))
import Games.ECS.Util.Misc
import Games.ECS.World

data DelayedOrNot
  = NotDelayed
  | Delayed
  deriving stock (Eq, Generic)
  deriving anyclass (Hashable)

-- | The core message queue type.
newtype MessageQueue a = MessageQueue {_queue :: Seq (DelayedOrNot, a)} deriving stock (Eq, Generic)

makeLenses ''MessageQueue

-- | Construct a new, empty t'MessageQueue'.
{-# INLINE newMessageQueue #-}
newMessageQueue :: MessageQueue a
newMessageQueue = let q = Empty in MessageQueue q

instance (Hashable a) => Hashable (MessageQueue a) where
  {-# INLINE hashWithSalt #-}
  hashWithSalt salt (MessageQueue q) = hashWithSalt salt (HashableSeq q)

-- | Monads which can atomically modify a t'MessageQueue'.
class (Monad m) => MonadAtomicMessageQueue a m where
  -- | Monadically perform a given state transition atomically on the t'MessageQueue', and return the result.
  stateMessageQueue :: (MessageQueue a -> (MessageQueue a, val)) -> m val

  -- | Atomically modify a t'MessageQueue'.
  modifyMessageQueue :: (MessageQueue a -> MessageQueue a) -> m ()

{-# INLINEABLE queueMessage #-}

-- | Queue message
queueMessage :: forall a m. (MonadAtomicMessageQueue a m) => a -> m ()
queueMessage msg = do
  modifyMessageQueue (\q -> q & queue %~ (|> (NotDelayed, msg)))

{-# INLINEABLE queueImmediateMessage #-}

-- | Push a message on to the front of the queue.
queueImmediateMessage :: forall a m. (MonadAtomicMessageQueue a m) => a -> m ()
queueImmediateMessage msg = do
  modifyMessageQueue (\q -> q & queue %~ ((NotDelayed, msg) <|))

{-# INLINEABLE queueMessages #-}

-- | Queue a series of messages
queueMessages :: (Foldable f, MonadAtomicMessageQueue a m) => f a -> m ()
queueMessages msgs = forM_ msgs queueMessage

{-# INLINEABLE readMessage #-}

-- | Pops the message queue
readMessage :: (MonadAtomicMessageQueue a m) => (a -> Bool) -> m (Maybe a)
readMessage shouldSkip =
  stateMessageQueue
    ( \(MessageQueue msgs) -> swap . fmap (MessageQueue . fmap clearDelayed) . go $ msgs
    )
  where
    clearDelayed (Delayed, a) = (NotDelayed, a)
    clearDelayed x = x
    go Empty = (Nothing, Empty)
    go ((NotDelayed, a) :< rest)
      | shouldSkip a = fmap ((Delayed, a) <|) (go rest)
      | otherwise = (Just a, rest)
    go ((Delayed, a) :< rest) = fmap ((Delayed, a) <|) (go rest)

-- | Which messages were processed.
data MessagesProcessed a
  = AllMessagesProcessed
  | NewMessagesProcessed !(Seq a)
  deriving stock (Eq, Generic, Show)

instance (Hashable a) => Hashable (MessagesProcessed a) where
  {-# INLINE hashWithSalt #-}
  hashWithSalt salt AllMessagesProcessed = hashWithSalt salt (0 :: Int)
  hashWithSalt salt (NewMessagesProcessed msgs) = (salt `hashWithSalt` (1 :: Int)) `hashWithSalt` HashableSeq msgs

instance Semigroup (MessagesProcessed a) where
  {-# INLINE (<>) #-}
  AllMessagesProcessed <> (NewMessagesProcessed Empty) = AllMessagesProcessed
  AllMessagesProcessed <> b = b
  (NewMessagesProcessed Empty) <> AllMessagesProcessed = AllMessagesProcessed
  a <> AllMessagesProcessed = a
  (NewMessagesProcessed a) <> (NewMessagesProcessed b) = NewMessagesProcessed (a <> b)

instance Monoid (MessagesProcessed a) where
  {-# INLINEABLE mempty #-}
  mempty = AllMessagesProcessed

{-# INLINEABLE processMessageQueue #-}

-- | Processes a message queue until it's empty.
processMessageQueue ::
  forall world a m.
  (MonadAtomicMessageQueue a m) =>
  -- | Predicate on whether a message should be skipped this time round, and left in the queue.
  (world Storing -> a -> Bool) ->
  -- | How to process an individual message.
  (world Storing -> a -> m (world Storing)) ->
  -- | The world.
  world Storing ->
  m (MessagesProcessed a, world Storing)
processMessageQueue shouldSkip processMsg oldWorld = processNextMessage SPEC mempty oldWorld
  where
    -- Process the message queue one at a time. Because it only ever locks the message queue long enough to
    -- pop it, messages can safely enqueue other messages.
    processNextMessage !sPEC doneMessages world = do
      nextMessage <- readMessage (shouldSkip world)
      case nextMessage of
        -- The queue is empty or has no messages available to process; return the list of effects we processed and the updated world
        Nothing -> pure (doneMessages, world)
        -- We've popped another effect off of the queue. Process it, then process the rest of the queue
        Just msg -> do
          processedNewWorld <- processMsg world msg
          processNextMessage sPEC (doneMessages <> (NewMessagesProcessed (singleton msg))) processedNewWorld