antiope-sqs 1.0.0 → 3.1.0

raw patch · 3 files changed

+126/−27 lines, 3 filesdep +unliftio-corePVP ok

version bump matches the API change (PVP)

Dependencies added: unliftio-core

API changes (from Hackage documentation)

Files

• antiope-sqs.cabal +7/−5
• src/Antiope/SQS.hs +36/−22
• src/Antiope/SQS/Classy.hs +83/−0

@@ -1,23 +1,22 @@--- This file has been generated from package.yaml by hpack version 0.20.0.+-- This file has been generated from package.yaml by hpack version 0.28.2. -- -- see: https://github.com/sol/hpack ----- hash: ac64958cc7efc69597ec92dfaf4373cb4d2525a8efafa7a89526a55e027a8434+-- hash: f45ea86f5796546a0006488266f03623cd4e14683965ef76ea04c9d4f0ef0ea4  name:           antiope-sqs-version:        1.0.0+version:        3.1.0 description:    Please see the README on Github at <https://github.com/arbor/antiope#readme> category:       Services homepage:       https://github.com/arbor/antiope#readme bug-reports:    https://github.com/arbor/antiope/issues-author:         Tyler Durden+author:         Arbor Networks maintainer:     mayhem@arbor.net copyright:      Arbor Networks license:        MIT license-file:   LICENSE build-type:     Simple cabal-version:  >= 1.10- extra-source-files:     ChangeLog.md     README.md@@ -45,8 +44,10 @@     , monad-loops     , network-uri     , text+    , unliftio-core   exposed-modules:       Antiope.SQS+      Antiope.SQS.Classy       Antiope.SQS.Types   other-modules:       Paths_antiope_sqs@@ -74,6 +75,7 @@     , monad-loops     , network-uri     , text+    , unliftio-core   other-modules:       Paths_antiope_sqs   default-language: Haskell2010

@@ -1,6 +1,5 @@ module Antiope.SQS-( MonadAWS-, FromText(..), fromText+( FromText(..), fromText , ToText(..) , QueueUrl(..) , SQSError(..)@@ -13,42 +12,57 @@ , s3Location' ) where -import Antiope.S3            (S3Uri (..))-import Antiope.SQS.Types     (QueueUrl (QueueUrl), SQSError (DeleteMessageBatchError))+import Antiope.S3              (S3Uri (..))+import Antiope.SQS.Types       (QueueUrl (QueueUrl), SQSError (DeleteMessageBatchError)) import Control.Lens-import Control.Monad         (join)-import Control.Monad.Loops   (unfoldWhileM)+import Control.Monad           (join)+import Control.Monad.IO.Unlift (MonadUnliftIO)+import Control.Monad.Loops     (unfoldWhileM) import Data.Aeson.Lens-import Data.Maybe            (catMaybes)-import Data.Text             (Text, pack, unpack)-import Network.AWS           (MonadAWS, send)-import Network.AWS.Data.Text (FromText (..), ToText (..), fromText, toText)-import Network.AWS.S3        hiding (s3Location)+import Data.Maybe              (catMaybes)+import Data.Text               (Text, pack, unpack)+import Network.AWS             (HasEnv)+import Network.AWS.Data.Text   (FromText (..), ToText (..), fromText, toText)+import Network.AWS.S3          hiding (s3Location) import Network.AWS.SQS +import qualified Network.AWS as AWS import qualified Network.URI as URI  -- | Reads the specified SQS queue once returning a bath of messages-readQueue :: MonadAWS m => QueueUrl -> m [Message]-readQueue (QueueUrl queueUrl) = do+readQueue :: (HasEnv e, MonadUnliftIO m)+  => e+  -> QueueUrl+  -> m [Message]+readQueue e (QueueUrl queueUrl) = AWS.runResourceT . AWS.runAWS e $ do   -- max wait time allowed by aws is 20sec, max number messages to recieve is 10-  resp <- send $ receiveMessage queueUrl & rmWaitTimeSeconds ?~ 10 & rmMaxNumberOfMessages ?~ 10+  resp <- AWS.send $ receiveMessage queueUrl & rmWaitTimeSeconds ?~ 10 & rmMaxNumberOfMessages ?~ 10   return $ resp ^. rmrsMessages  -- | Reads the specified SQS queue until it is empty and returns a list of messages-drainQueue :: MonadAWS m => QueueUrl -> m [Message]-drainQueue queueUrl =-  join <$> unfoldWhileM (not . null) (readQueue queueUrl)+drainQueue :: (HasEnv e, MonadUnliftIO m)+  => e+  -> QueueUrl+  -> m [Message]+drainQueue e queueUrl = join <$> unfoldWhileM (not . null) (readQueue e queueUrl) -ackMessage :: MonadAWS m => QueueUrl -> Message -> m (Either SQSError ())-ackMessage q msg = ackMessages q [msg]+ackMessage :: (HasEnv e, MonadUnliftIO m)+  => e+  -> QueueUrl+  -> Message+  -> m (Either SQSError ())+ackMessage e q msg = ackMessages e q [msg] -ackMessages :: MonadAWS m => QueueUrl -> [Message] -> m (Either SQSError ())-ackMessages (QueueUrl queueUrl) msgs = do+ackMessages :: (HasEnv e, MonadUnliftIO m)+  => e+  -> QueueUrl+  -> [Message]+  -> m (Either SQSError ())+ackMessages e (QueueUrl queueUrl) msgs = AWS.runResourceT . AWS.runAWS e $ do   let receipts = catMaybes $ msgs ^.. each . mReceiptHandle   -- each dmbr needs an ID. just use the list index.   let dmbres = (\(r, i) -> deleteMessageBatchRequestEntry (pack (show i)) r) <$> zip receipts ([0..] :: [Int])-  resp <- send $ deleteMessageBatch queueUrl & dmbEntries .~ dmbres+  resp <- AWS.send $ deleteMessageBatch queueUrl & dmbEntries .~ dmbres   -- only acceptable if no errors.   if resp ^. dmbrsResponseStatus == 200     then case resp ^. dmbrsFailed of

@@ -0,0 +1,83 @@+module Antiope.SQS.Classy+( FromText(..), fromText+, ToText(..)+, QueueUrl(..)+, SQSError(..)+, readQueue+, drainQueue+, ackMessage+, ackMessages+, messageInBody+, s3Location+, s3Location'+) where++import Antiope.S3            (S3Uri (..))+import Antiope.SQS.Types     (QueueUrl (QueueUrl), SQSError (DeleteMessageBatchError))+import Control.Lens+import Control.Monad         (join)+import Control.Monad.Loops   (unfoldWhileM)+import Data.Aeson.Lens+import Data.Maybe            (catMaybes)+import Data.Text             (Text, pack, unpack)+import Network.AWS           (MonadAWS)+import Network.AWS.Data.Text (FromText (..), ToText (..), fromText, toText)+import Network.AWS.S3        hiding (s3Location)+import Network.AWS.SQS++import qualified Network.AWS as AWS+import qualified Network.URI as URI++-- | Reads the specified SQS queue once returning a bath of messages+readQueue :: MonadAWS m+  => QueueUrl+  -> m [Message]+readQueue (QueueUrl queueUrl) = do+  -- max wait time allowed by aws is 20sec, max number messages to recieve is 10+  resp <- AWS.send $ receiveMessage queueUrl & rmWaitTimeSeconds ?~ 10 & rmMaxNumberOfMessages ?~ 10+  return $ resp ^. rmrsMessages++-- | Reads the specified SQS queue until it is empty and returns a list of messages+drainQueue :: MonadAWS m+  => QueueUrl+  -> m [Message]+drainQueue queueUrl = join <$> unfoldWhileM (not . null) (readQueue queueUrl)++ackMessage :: MonadAWS m+  => QueueUrl+  -> Message+  -> m (Either SQSError ())+ackMessage q msg = ackMessages q [msg]++ackMessages :: MonadAWS m+  => QueueUrl+  -> [Message]+  -> m (Either SQSError ())+ackMessages (QueueUrl queueUrl) msgs = do+  let receipts = catMaybes $ msgs ^.. each . mReceiptHandle+  -- each dmbr needs an ID. just use the list index.+  let dmbres = (\(r, i) -> deleteMessageBatchRequestEntry (pack (show i)) r) <$> zip receipts ([0..] :: [Int])+  resp <- AWS.send $ deleteMessageBatch queueUrl & dmbEntries .~ dmbres+  -- only acceptable if no errors.+  if resp ^. dmbrsResponseStatus == 200+    then case resp ^. dmbrsFailed of+        [] -> return $ Right ()+        _  -> return $ Left DeleteMessageBatchError+    else return $ Left DeleteMessageBatchError++-- Extract the "Message" content in the body if have+messageInBody :: Text -> Maybe Text+messageInBody body = body ^? key "Message" . _String++s3Location :: Message -> Maybe S3Uri+s3Location msg = join $ s3Location' <$> msg ^. mBody++s3Location' :: Text -> Maybe S3Uri+s3Location' msg = do+  s3m <- messageInBody msg ^? _Just . key "Records" . nth 0 . key "s3"+  b   <- s3m ^? key "bucket" . key "name" . _String+  k   <- s3m ^? key "object" . key "key" . _String+  pure $ S3Uri (BucketName b) (ObjectKey $ uriDecode k)++uriDecode :: Text -> Text+uriDecode = pack . URI.unEscapeString . unpack