{-# LANGUAGE GeneralizedNewtypeDeriving #-}
{-# LANGUAGE OverloadedStrings #-}

module Network.AMQP.Worker.Queue where

import Control.Monad.IO.Class (MonadIO, liftIO)
import Data.Default (Default (..))
import Data.String (IsString)
import Data.Text (Text)
import Network.AMQP (ExchangeOpts (..), QueueOpts)
import qualified Network.AMQP as AMQP

import Network.AMQP.Worker.Connection (Connection, exchange, withChannel)
import Network.AMQP.Worker.Key (Bind, Key (..), keyText, toBindKey)

type QueueName = Text

newtype QueuePrefix = QueuePrefix Text
    deriving (Int -> QueuePrefix -> ShowS
[QueuePrefix] -> ShowS
QueuePrefix -> String
(Int -> QueuePrefix -> ShowS)
-> (QueuePrefix -> String)
-> ([QueuePrefix] -> ShowS)
-> Show QueuePrefix
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> QueuePrefix -> ShowS
showsPrec :: Int -> QueuePrefix -> ShowS
$cshow :: QueuePrefix -> String
show :: QueuePrefix -> String
$cshowList :: [QueuePrefix] -> ShowS
showList :: [QueuePrefix] -> ShowS
Show, QueuePrefix -> QueuePrefix -> Bool
(QueuePrefix -> QueuePrefix -> Bool)
-> (QueuePrefix -> QueuePrefix -> Bool) -> Eq QueuePrefix
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: QueuePrefix -> QueuePrefix -> Bool
== :: QueuePrefix -> QueuePrefix -> Bool
$c/= :: QueuePrefix -> QueuePrefix -> Bool
/= :: QueuePrefix -> QueuePrefix -> Bool
Eq, String -> QueuePrefix
(String -> QueuePrefix) -> IsString QueuePrefix
forall a. (String -> a) -> IsString a
$cfromString :: String -> QueuePrefix
fromString :: String -> QueuePrefix
IsString)

instance Default QueuePrefix where
    def :: QueuePrefix
def = Text -> QueuePrefix
QueuePrefix Text
"main"

-- | A queue is an inbox for messages to be delivered
data Queue msg
    = Queue (Key Bind msg) QueueName
    deriving (Int -> Queue msg -> ShowS
[Queue msg] -> ShowS
Queue msg -> String
(Int -> Queue msg -> ShowS)
-> (Queue msg -> String)
-> ([Queue msg] -> ShowS)
-> Show (Queue msg)
forall msg. Int -> Queue msg -> ShowS
forall msg. [Queue msg] -> ShowS
forall msg. Queue msg -> String
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: forall msg. Int -> Queue msg -> ShowS
showsPrec :: Int -> Queue msg -> ShowS
$cshow :: forall msg. Queue msg -> String
show :: Queue msg -> String
$cshowList :: forall msg. [Queue msg] -> ShowS
showList :: [Queue msg] -> ShowS
Show, Queue msg -> Queue msg -> Bool
(Queue msg -> Queue msg -> Bool)
-> (Queue msg -> Queue msg -> Bool) -> Eq (Queue msg)
forall msg. Queue msg -> Queue msg -> Bool
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: forall msg. Queue msg -> Queue msg -> Bool
== :: Queue msg -> Queue msg -> Bool
$c/= :: forall msg. Queue msg -> Queue msg -> Bool
/= :: Queue msg -> Queue msg -> Bool
Eq)

-- | Create a queue to receive messages matching the 'Key' with a name prefixed via `queueName`.
--
-- > q <- Worker.queue conn "main" $ key "messages" & any1
-- > Worker.worker conn def q onError onMessage
queue :: (MonadIO m) => Connection -> QueuePrefix -> Key a msg -> m (Queue msg)
queue :: forall (m :: * -> *) a msg.
MonadIO m =>
Connection -> QueuePrefix -> Key a msg -> m (Queue msg)
queue Connection
conn QueuePrefix
pre Key a msg
key = do
    Connection -> Text -> Key a msg -> m (Queue msg)
forall (m :: * -> *) a msg.
MonadIO m =>
Connection -> Text -> Key a msg -> m (Queue msg)
queueNamed Connection
conn (QueuePrefix -> Key a msg -> Text
forall a msg. QueuePrefix -> Key a msg -> Text
queueName QueuePrefix
pre Key a msg
key) Key a msg
key

-- | Create a queue to receive messages matching the binding key. Each queue with a unique name
-- will be delivered a separate copy of the messsage. Workers operating on the same queue,
-- or on queues with the same name will load balance
queueNamed :: (MonadIO m) => Connection -> QueueName -> Key a msg -> m (Queue msg)
queueNamed :: forall (m :: * -> *) a msg.
MonadIO m =>
Connection -> Text -> Key a msg -> m (Queue msg)
queueNamed Connection
conn Text
name Key a msg
key = do
    let q :: Queue msg
q = Key Bind msg -> Text -> Queue msg
forall msg. Key Bind msg -> Text -> Queue msg
Queue (Key a msg -> Key Bind msg
forall a msg. Key a msg -> Key Bind msg
toBindKey Key a msg
key) Text
name
    Connection -> Queue msg -> m ()
forall (m :: * -> *) msg.
MonadIO m =>
Connection -> Queue msg -> m ()
bindQueue Connection
conn Queue msg
q
    Queue msg -> m (Queue msg)
forall a. a -> m a
forall (m :: * -> *) a. Monad m => a -> m a
return Queue msg
q

-- | Name a queue with a prefix and the binding key name. Useful for seeing at
-- a glance which queues are receiving which messages
--
-- > -- "main messages.new"
-- > queueName "main" (key "messages" & word "new")
queueName :: QueuePrefix -> Key a msg -> QueueName
queueName :: forall a msg. QueuePrefix -> Key a msg -> Text
queueName (QueuePrefix Text
pre) Key a msg
key = Text
pre Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Key a msg -> Text
forall a msg. Key a msg -> Text
keyText Key a msg
key

-- | Queues must be bound before you publish messages to them, or the messages will not be saved.
-- Use `queue` or `queueNamed` instead
bindQueue :: (MonadIO m) => Connection -> Queue msg -> m ()
bindQueue :: forall (m :: * -> *) msg.
MonadIO m =>
Connection -> Queue msg -> m ()
bindQueue Connection
conn (Queue Key Bind msg
key Text
name) =
    IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ Connection -> (Channel -> IO ()) -> IO ()
forall b. Connection -> (Channel -> IO b) -> IO b
withChannel Connection
conn ((Channel -> IO ()) -> IO ()) -> (Channel -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Channel
chan -> do
        let options :: QueueOpts
options = QueueOpts
AMQP.newQueue{AMQP.queueName = name}
        let exg :: ExchangeOpts
exg = ExchangeOpts
AMQP.newExchange{exchangeName = exchange conn, exchangeType = "topic"}
        _ <- Channel -> ExchangeOpts -> IO ()
AMQP.declareExchange Channel
chan ExchangeOpts
exg
        _ <- AMQP.declareQueue chan options
        _ <- AMQP.bindQueue chan name (exchange conn) (keyText key)
        return ()