[broker] getMessageById

This is to just fetch the message, given it's message id.

This is for querying reasons, to see what is available.

cabal.project

@@ -8,7 +8,7 @@ packages:
 source-repository-package
     type: git
     location: https://gitlab.iscpif.fr/gargantext/haskell-pgmq
-    tag: b45423def18390818c19861a458d6cd7964c5b60
+    tag: 1dd92f0aa8e9f8096064e5656c336e562680f4e3
     
 tests: true
 

src/Async/Worker/Broker/PGMQ.hs

@@ -137,6 +137,9 @@ instance (SerializableMessage a, Show a) => MessageBroker PGMQBroker a where
     msgIds <- PGMQ.queueAvailableIds conn queue
     pure $ PGMQMid <$> msgIds
 
+  getMessageById (PGMQBroker' { conn }) (renderQueue -> queue) (PGMQMid msgId) = do
+    mMsg <- PGMQ.readMessageById conn queue msgId
+    pure $ PGMQBM <$> mMsg
 
 
 instance ToJSON (MessageId PGMQBroker) where

src/Async/Worker/Broker/Redis.hs

@@ -184,6 +184,9 @@ instance (SerializableMessage a, Show a) => MessageBroker RedisBroker a where
       Left _ -> return []
       Right msgIds -> return $ catMaybes (bsToId <$> msgIds)
 
+  getMessageById b queue msgId = do
+    getRedisMessage b queue msgId
+
 -- Helper functions for getting redis keys
 
 -- | Redis counter is an 'Int', while sets can only store strings
@@ -223,7 +226,7 @@ archiveKey queue = BS.pack $ beePrefix <> "archive-" <> renderQueue queue
               
 
 getRedisMessage :: FromJSON a
-               => Broker RedisBroker a
+                => Broker RedisBroker a
                 -> Queue
                 -> MessageId RedisBroker
                 -> IO (Maybe (BrokerMessage RedisBroker a))

src/Async/Worker/Broker/STM.hs

@@ -163,6 +163,14 @@ instance (Show a) => MessageBroker STMBroker a where
     msgIds <- Map.keys <$> filterAvailableMessages broker queue
     pure $ STMMid <$> msgIds
 
+  getMessageById (STMBroker' { stmMap }) queue (STMMid msgId) = do
+    atomically $ do
+      map' <- readTVar stmMap
+      case Map.lookup queue map' of
+        Nothing -> pure Nothing
+        Just qm ->
+          pure $ STMBM <$> (Map.lookup msgId qm)
+
 
 filterAvailableMessages :: Broker STMBroker a -> Queue -> IO (Map.Map Int (STMWithMsgId a))
 filterAvailableMessages (STMBroker' { stmMap }) queue = do

src/Async/Worker/Broker/Types.hs

@@ -166,3 +166,6 @@ class (
 
   {-| List all pending message ids -}
   listPendingMessageIds :: Broker b a -> Queue -> IO [MessageId b]
+
+  {-| Get message by it's id -}
+  getMessageById :: Broker b a -> Queue -> MessageId b -> IO (Maybe (BrokerMessage b a))

tests/Test/Integration/Broker.hs

@@ -152,6 +152,13 @@ brokerTests bInitParams =
       qs <- BT.getQueueSize broker queue
       qs `shouldBe` 0
 
+    it "can get message by it's id (from queue)" $ \(TestEnv { broker, queue }) -> do
+      text <- randomString (onlyAlphaNum randomASCII) 20
+      let msg = Message { text }
+      msgId <- BT.sendMessage broker queue (BT.toMessage msg)
+      mMsg <- BT.getMessageById broker queue msgId
+      (BT.toA . BT.getMessage <$> mMsg) `shouldBe` (Just msg)
+
 
 pgmqBrokerInitParams :: IO (BT.BrokerInitParams PGMQ.PGMQBroker Message)
 pgmqBrokerInitParams = do