diff --git a/antiope-sqs.cabal b/antiope-sqs.cabal --- a/antiope-sqs.cabal +++ b/antiope-sqs.cabal @@ -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 +version: 7.0.0 +synopsis: Please see the README on Github at +description: Please see the README on Github at . 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 diff --git a/src/Antiope/SQS.hs b/src/Antiope/SQS.hs --- a/src/Antiope/SQS.hs +++ b/src/Antiope/SQS.hs @@ -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 () diff --git a/src/Antiope/SQS/Messages.hs b/src/Antiope/SQS/Messages.hs --- a/src/Antiope/SQS/Messages.hs +++ b/src/Antiope/SQS/Messages.hs @@ -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) diff --git a/src/Antiope/SQS/Types.hs b/src/Antiope/SQS/Types.hs new file mode 100644 --- /dev/null +++ b/src/Antiope/SQS/Types.hs @@ -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 diff --git a/test/Antiope/SQS/MessagesSpec.hs b/test/Antiope/SQS/MessagesSpec.hs new file mode 100644 --- /dev/null +++ b/test/Antiope/SQS/MessagesSpec.hs @@ -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 diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -1,2 +1,1 @@ -main :: IO () -main = putStrLn "Test suite not yet implemented" +{-# OPTIONS_GHC -F -pgmF hspec-discover #-}