antiope-sqs 6.4.0 → 7.0.0

raw patch · 6 files changed

+165/−71 lines, 6 filesdep +bytestringdep +hedgehogdep +hspecdep −amazonka-s3dep −antiope-messagesdep −antiope-s3PVP ok

version bump matches the API change (PVP)

Dependencies added: bytestring, hedgehog, hspec, hw-hspec-hedgehog, time, unordered-containers

Dependencies removed: amazonka-s3, antiope-messages, antiope-s3

API changes (from Hackage documentation)

Files

• antiope-sqs.cabal +19/−26
• src/Antiope/SQS.hs +39/−11
• src/Antiope/SQS/Messages.hs +49/−32
• src/Antiope/SQS/Types.hs +39/−0
• test/Antiope/SQS/MessagesSpec.hs +18/−0
• test/Spec.hs +1/−2

@@ -1,14 +1,9 @@-cabal-version: 1.12---- This file has been generated from package.yaml by hpack version 0.31.1.------ see: https://github.com/sol/hpack------ hash: 443e48287c77512b5c25fec64ac9d6cf85fec8c2c04eaf787c0ac373596918bd+cabal-version: 2.2  name:           antiope-sqs-version:        6.4.0-description:    Please see the README on Github at <https://github.com/arbor/antiope#readme>+version:        7.0.0+synopsis:       Please see the README on Github at <https://github.com/arbor/antiope#readme>+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@@ -27,24 +22,16 @@   location: https://github.com/arbor/antiope  library-  exposed-modules:-      Antiope.SQS-      Antiope.SQS.Messages-  other-modules:-      Paths_antiope_sqs-  hs-source-dirs:-      src+  hs-source-dirs: src   default-extensions: BangPatterns GeneralizedNewtypeDeriving OverloadedStrings TupleSections   ghc-options: -Wall -Wcompat -Wincomplete-record-updates -Wincomplete-uni-patterns -Wredundant-constraints -msse4.2   build-depends:       aeson     , amazonka     , amazonka-core-    , amazonka-s3     , amazonka-sqs-    , antiope-messages-    , antiope-s3     , base >=4.7 && <5+    , bytestring     , conduit     , generic-lens     , lens@@ -54,34 +41,40 @@     , network-uri     , text     , unliftio-core+    , unordered-containers+  exposed-modules:+      Antiope.SQS+      Antiope.SQS.Messages+      Antiope.SQS.Types   default-language: Haskell2010  test-suite antiope-sqs-test   type: exitcode-stdio-1.0   main-is: Spec.hs-  other-modules:-      Paths_antiope_sqs-  hs-source-dirs:-      test+  hs-source-dirs: test   default-extensions: BangPatterns GeneralizedNewtypeDeriving OverloadedStrings TupleSections   ghc-options: -Wall -Wcompat -Wincomplete-record-updates -Wincomplete-uni-patterns -Wredundant-constraints -msse4.2 -threaded -rtsopts -with-rtsopts=-N   build-depends:       aeson     , amazonka     , amazonka-core-    , amazonka-s3     , amazonka-sqs-    , antiope-messages-    , antiope-s3     , antiope-sqs     , base >=4.7 && <5+    , bytestring     , conduit     , generic-lens+    , hedgehog >=0.5 && <0.7+    , hspec >=2.4 && <2.7+    , hw-hspec-hedgehog >=0.1 && <0.3     , lens     , lens-aeson     , monad-loops     , mtl     , network-uri     , text+    , time     , unliftio-core+  other-modules:+      Antiope.SQS.MessagesSpec   default-language: Haskell2010

@@ -1,26 +1,36 @@ module Antiope.SQS   ( QueueUrl(..)   , SQSError(..)+  , HasReceiptHandle+  , ConsumerMode(..)+  , ConsumerResult(..)   , readQueue   , drainQueue   , ackMessage   , ackMessages-  , mBody   , queueSource+  , forAllMessages++  -- * Re-exports+  , Message+  , mBody, mMD5OfBody, mMessageId, mReceiptHandle, mAttributes   ) where -import Antiope.Messages         (QueueUrl (QueueUrl), SQSError (DeleteMessageBatchError)) import Control.Lens-import Control.Monad            (join)+import Control.Monad            (forM_, join, void)+import Control.Monad.IO.Unlift  (MonadUnliftIO) import Control.Monad.Loops      (unfoldWhileM) import Control.Monad.Trans      (lift)+import Data.Coerce              (coerce) import Data.Conduit import Data.Conduit.Combinators (yieldMany) import Data.Maybe               (catMaybes) import Data.Text                (pack)-import Network.AWS              (MonadAWS)+import Network.AWS              (HasEnv, MonadAWS, runAWS, runResourceT) import Network.AWS.SQS +import Antiope.SQS.Types+ import qualified Network.AWS as AWS  -- | Reads the specified SQS queue once returning a bath of messages@@ -39,21 +49,21 @@ drainQueue queueUrl = join <$> unfoldWhileM (not . null) (readQueue queueUrl)  -- | Acknowledges a single SQS message-ackMessage :: MonadAWS m+ackMessage :: (MonadAWS m, HasReceiptHandle msg)   => QueueUrl-  -> Message+  -> msg   -> m (Either SQSError ()) ackMessage q msg = ackMessages q [msg]  -- | Acknowledges a group of SQS messages-ackMessages :: MonadAWS m+ackMessages :: (MonadAWS m, HasReceiptHandle msg)   => QueueUrl-  -> [Message]+  -> [msg]   -> m (Either SQSError ()) ackMessages (QueueUrl queueUrl) msgs = do-  let receipts = catMaybes $ msgs ^.. each . mReceiptHandle+  let receipts = msgs ^.. each . to getReceiptHandle & catMaybes   -- each dmbr needs an ID. just use the list index.-  let dmbres = (\(r, i) -> deleteMessageBatchRequestEntry (pack (show i)) r) <$> zip receipts ([0..] :: [Int])+  let dmbres = (\(r, i) -> deleteMessageBatchRequestEntry (pack (show i)) r) <$> zip (coerce receipts) ([0..] :: [Int])   resp <- AWS.send $ deleteMessageBatch queueUrl & dmbEntries .~ dmbres   -- only acceptable if no errors.   if resp ^. dmbrsResponseStatus == 200@@ -69,4 +79,22 @@   yieldMany (m ^. rmrsMessages)   queueSource (QueueUrl queueUrl) -+forAllMessages :: (MonadUnliftIO m, HasEnv env)+  => env+  -> QueueUrl+  -> ConsumerMode+  -> (Message -> m ConsumerResult)+  -> m ()+forAllMessages env queue mode process = go+  where+    go = do+      msgs <- runResourceT $ runAWS env (readQueue queue)+      case (mode, msgs) of+        (Drain, []) -> pure ()+        _           -> processBatch msgs >> go+    processBatch msgs =+      forM_ msgs $ \msg -> do+        res <- process msg+        case res of+          Ack  -> void . runResourceT . runAWS env $ (ackMessage queue msg)+          Nack -> pure ()

@@ -1,45 +1,62 @@+{-# LANGUAGE DeriveFunctor #-} {-# LANGUAGE DeriveGeneric #-}  module Antiope.SQS.Messages-  ( Many (..)-  , SqsMessage (..)+  ( SqsMessage (..)+  , Message, mBody+  , QueueUrl+  , fromMessage, fromMessageMaybe+  , decodeBody   ) where -import Data.Aeson   as Aeson-import Data.Text    (Text)-import GHC.Generics (Generic)--import qualified Data.Aeson.Types   as Aeson-import qualified Data.Text.Encoding as T+import Antiope.SQS.Types+import Control.Lens      ((&), (<&>), (^.))+import Data.Aeson        as Aeson+import Data.Text         (Text)+import GHC.Generics      (Generic)+import Network.AWS.SQS   (Message, mAttributes, mBody, mMD5OfBody, mMessageId, mReceiptHandle) -newtype Many a = Many { records :: [a] } deriving (Show, Eq, Generic)+import qualified Data.HashMap.Strict as Hash+import qualified Data.Text.Encoding  as Text+import qualified Network.AWS.SQS     as SQS -instance FromJSON a => FromJSON (Many a) where-  parseJSON =-    withObject "Many" $ \obj ->-      Many <$> obj .: "Records"+-- | Converts 'Message` into 'SqsMessage' if its body is decodable as 'a'+fromMessageMaybe :: FromJSON a => Message -> Maybe (SqsMessage a)+fromMessageMaybe msg =+  let smsg = fromMessage msg+  in body smsg <&> (\a -> smsg { body = a}) -instance ToJSON a => ToJSON (Many a) where-  toJSON (Many a) =-    object ["Records" Aeson..= a]+-- | Converts a 'Message' into 'SqsMessage'+fromMessage :: FromJSON a => Message -> SqsMessage (Maybe a)+fromMessage msg =+  let attr a = msg ^. mAttributes & Hash.lookup a+  in SqsMessage+        { messageId                         = msg ^. mMessageId+        , receiptHandle                     = msg ^. mReceiptHandle <&> ReceiptHandle+        , md5OfBody                         = msg ^. mMD5OfBody+        , senderId                          = attr SQS.SenderId+        , sendTimestamp                     = attr SQS.SentTimestamp+        , approximateReceiveCount           = attr SQS.ApproximateReceiveCount+        , approximateFirstReceiveTimestamp  = attr SQS.ApproximateFirstReceiveTimestamp+        , body                              = decodeBody msg+        }  data SqsMessage a = SqsMessage-  { senderId      :: !(Maybe Text)-  , messageId     :: !(Maybe Text)-  , md5OfBody     :: !(Maybe Text)-  , receiptHandle :: !(Maybe Text)-  , body          :: !(Maybe a)-  } deriving (Show, Eq, Generic)+  { senderId                         :: !(Maybe Text)+  , messageId                        :: !(Maybe Text)+  , md5OfBody                        :: !(Maybe Text)+  , receiptHandle                    :: !(Maybe ReceiptHandle)+  , sendTimestamp                    :: !(Maybe Text)+  , approximateReceiveCount          :: !(Maybe Text)+  , approximateFirstReceiveTimestamp :: !(Maybe Text)+  , body                             :: !a+  } deriving (Show, Eq, Generic, Functor) -instance FromJSON a => FromJSON (SqsMessage a) where-  parseJSON = withObject "SqsMessage" $ \obj -> SqsMessage-    <$> obj .:? "SenderId"-    <*> obj .:? "MessageId"-    <*> obj .:? "Md5OfBody"-    <*> obj .:? "ReceiptHandle"-    <*> decodeEscaped obj "Body"+instance HasReceiptHandle (SqsMessage a) where+  getReceiptHandle = receiptHandle -decodeEscaped :: FromJSON b => Object -> Text -> Aeson.Parser b-decodeEscaped o t =-  (o .: t) >>= (either fail pure . eitherDecodeStrict . T.encodeUtf8)+------------------------------------------------------------------------------- +decodeBody :: FromJSON a => Message -> Maybe a+decodeBody msg =+  msg ^. mBody >>= (Aeson.decodeStrict . Text.encodeUtf8)

@@ -0,0 +1,39 @@+{-# LANGUAGE DeriveGeneric #-}+module Antiope.SQS.Types+where++import Control.Lens          ((<&>), (^.))+import Data.String           (IsString)+import Data.Text             (Text)+import GHC.Generics          (Generic)+import Network.AWS.Data.Text (FromText (..), ToText (..))+import Network.AWS.SQS       (Message, mReceiptHandle)++-- | Queue consuming mode+data ConsumerMode+  = Drain   -- ^ Keep consuming the queue until there are no messages+  | Forever -- ^ Keep consuming the queue forever+  deriving (Show, Eq, Generic)++data ConsumerResult+  = Ack+  | Nack+  deriving (Show, Eq, Generic)++data SQSError = DeleteMessageBatchError+  deriving (Eq, Show, Generic)++newtype QueueUrl = QueueUrl Text+  deriving (Show, Eq, IsString, FromText, ToText, Generic)++newtype ReceiptHandle = ReceiptHandle Text+  deriving (Show, Read, Eq, Ord, FromText, ToText)++class HasReceiptHandle a where+  getReceiptHandle :: a -> Maybe ReceiptHandle++instance HasReceiptHandle Message where+  getReceiptHandle msg = msg ^. mReceiptHandle <&> ReceiptHandle++instance HasReceiptHandle ReceiptHandle where+  getReceiptHandle = Just . id

@@ -0,0 +1,18 @@+module Antiope.SQS.MessagesSpec+where++import Antiope.SQS.Messages+import Data.Aeson++import HaskellWorks.Hspec.Hedgehog+import Hedgehog+import Hedgehog.Gen                as Gen+import Hedgehog.Range              as Range+import Test.Hspec++{-# ANN module ("HLint: Ignore Redundant do" :: String) #-}++spec :: Spec+spec = describe "Antiope.SQS.MessagesSpec" $ do+  it "Implement me" $ require $ property $ do+    True === True

@@ -1,2 +1,1 @@-main :: IO ()-main = putStrLn "Test suite not yet implemented"+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}