diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -3,15 +3,22 @@ A local version of `sqsd`, the daemon that runs in Elastic Beanstalk's _Worker Environments_. -**Current limitations:** - -* Only supports `application/json` content type. -* Local SQS endpoint (host/port) is hardcoded. Only tested with ElasticMQ. - ## Usage ```bash -sqsd-local http://localhost:8080 my-worker-queue-name my-deadletter-queue-name +Usage: sqsd-local WORKER_QUEUE_NAME [options] + +Options: + + -h --help Print this help message + -V --version Print the CLI version + -u URL --worker-url=URL Specify the worker URL to send POST requests to (default: http://localhost:5000) + -d NAME --dead-letter-queue-name=NAME Name of the SQS queue to send messages to which the worker failed processing (no default) + -T NAME --http-timeout=NAME Timeout in seconds for the HTTP POST request to worker (default: 30) + -C MEDIA_TYPE --content-type=MEDIA_TYPE Content-Type header value to use in HTTP POST request to worker (default: application/octet-stream) + --sqs-host=MEDIA_TYPE SQS endpoint host (default: localhost) + --sqs-port=MEDIA_TYPE SQS endpoint port (default: 9324) + --forked If messages should be POSTed to the worker concurrently (default: false) ``` ## Install/Build diff --git a/sqsd-local.cabal b/sqsd-local.cabal --- a/sqsd-local.cabal +++ b/sqsd-local.cabal @@ -1,5 +1,5 @@ name: sqsd-local -version: 0.1.1 +version: 0.2.0 synopsis: Initial project template from stack description: A local version of sqsd, the daemon that runs in Elastic Beanstalk's Worker Environments. homepage: https://github.com/owickstrom/sqsd-local#readme @@ -20,9 +20,12 @@ build-depends: amazonka , amazonka-sqs , base >= 4.7 && < 5 + , bytestring >= 0.10.8.1 , case-insensitive >= 1.2.0.7 , exceptions >= 0.8.3 + , http-client >= 0.4.31.2 , lens + , lifted-base >= 0.2.3.8 , resourcet >= 1.1.9 , text , unordered-containers >= 0.2.7.2 diff --git a/src/Main.hs b/src/Main.hs --- a/src/Main.hs +++ b/src/Main.hs @@ -1,19 +1,105 @@ +{-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE TemplateHaskell #-} module Main where +import Control.Lens +import qualified Data.ByteString.Char8 as BS +import Data.Text (Text) import qualified Data.Text as Text import Network.SQS.Daemon.Local +import System.Console.GetOpt import System.Environment +import System.IO +import Text.Printf +import qualified Data.Version as Version +import Paths_sqsd_local + +data Options + = Options { _daemonOptions :: DaemonOptions + , _showHelp :: Bool + , _showVersion :: Bool + } deriving (Show, Eq, Ord) + +makeLenses ''Options + +defaultOptions :: Text -> Options +defaultOptions workerQueueName' = + Options { _daemonOptions = DaemonOptions { _workerUrl = "http://localhost:5000" + , _workerQueueName = workerQueueName' + , _deadLetterQueueName = Nothing + , _httpTimeout = Just 30 + , _contentType = BS.pack "application/octet-stream" + , _sqsHost = BS.pack "localhost" + , _sqsPort = 9324 + , _forked = False + } + , _showHelp = False + , _showVersion = False + } + +readMaybe :: Read a => String -> Maybe a +readMaybe s = + case reads s of + [(val, "")] -> Just val + _ -> Nothing + +options :: [OptDescr (Options -> Options)] +options = + [ Option ['h'] ["help"] + (NoArg (set showHelp True)) + "Print this help message" + , Option ['V'] ["version"] + (NoArg (set showVersion True)) + "Print the CLI version" + , Option "u" ["worker-url"] + (ReqArg (set (daemonOptions . workerUrl) . Text.pack) "URL") + "Specify the worker URL to send POST requests to (default: http://localhost:5000)" + , Option "d" ["dead-letter-queue-name"] + (ReqArg (set (daemonOptions . deadLetterQueueName) . Just . Text.pack) "NAME") + "Name of the SQS queue to send messages to which the worker failed processing (no default)" + , Option "T" ["http-timeout"] + (ReqArg (set (daemonOptions . httpTimeout) . readMaybe) "NAME") + "Timeout in seconds for the HTTP POST request to worker (default: 30)" + , Option "C" ["content-type"] + (ReqArg (set (daemonOptions . contentType) . BS.pack) "MEDIA_TYPE") + "Content-Type header value to use in HTTP POST request to worker (default: application/octet-stream)" + , Option "" ["sqs-host"] + (ReqArg (set (daemonOptions . sqsHost) . BS.pack) "MEDIA_TYPE") + "SQS endpoint host (default: localhost)" + , Option "" ["sqs-port"] + (ReqArg (set (daemonOptions . sqsPort) . maybe 9324 id . readMaybe) "MEDIA_TYPE") + "SQS endpoint port (default: 9324)" + , Option "" ["forked"] + (NoArg (set (daemonOptions . forked) True)) + "If messages should be POSTed to the worker concurrently (default: false)" + ] + +usage :: String +usage = "Usage: sqsd-local WORKER_QUEUE_NAME [options]" + +help :: String +help = usageInfo (usage ++ "\n\n" ++ "Options:\n") options + +parseOptions :: [String] -> IO (Either String Options) +parseOptions argv = + case getOpt Permute options argv of + (o, [queueName], []) -> + return (Right (foldl (&) (defaultOptions (Text.pack queueName)) o)) + (_, _, errs) -> + return (Left (concat errs ++ help)) + +runDaemon :: Options -> IO () +runDaemon = + \case + opts | opts ^. showHelp -> hPutStr stderr help + opts | opts ^. showVersion -> hPutStrLn stderr (Version.showVersion version) + opts -> startDaemon (opts ^. daemonOptions) + main :: IO () main = do - args <- map Text.pack <$> getArgs - case args of - [workerUrl', workerQueueName', deadletterQueueName'] -> - let options = DaemonOptions { workerUrl = workerUrl' - , workerQueueName = workerQueueName' - , deadletterQueueName = deadletterQueueName' - } - in startDaemon options - _ -> - putStrLn "Usage: sqsd-local WORKER_URL WORKER_QUEUE_NAME DEADLETTER_QUEUE_NAME" + res <- getArgs >>= parseOptions + case res of + Left err -> hPutStr stderr err + Right opts -> runDaemon opts diff --git a/src/Network/SQS/Daemon/Local.hs b/src/Network/SQS/Daemon/Local.hs --- a/src/Network/SQS/Daemon/Local.hs +++ b/src/Network/SQS/Daemon/Local.hs @@ -1,39 +1,74 @@ +{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} module Network.SQS.Daemon.Local ( startDaemon , DaemonOptions(..) + , workerUrl + , workerQueueName + , deadLetterQueueName + , httpTimeout + , contentType + , sqsHost + , sqsPort + , forked ) where +import Control.Concurrent.Lifted import Control.Lens import Control.Monad import Control.Monad.Catch import Control.Monad.IO.Class import Control.Monad.Trans.AWS import Control.Monad.Trans.Resource +import Data.ByteString (ByteString) import Data.CaseInsensitive import Data.HashMap.Strict (HashMap) import qualified Data.HashMap.Strict as HM +import Data.Maybe import Data.Monoid -import Data.Text (Text, unpack) +import Data.Text (Text) +import qualified Data.Text as Text import Data.Text.Encoding import Data.Text.IO as Text import Network.AWS.SQS +import Network.HTTP.Client (defaultManagerSettings, + managerResponseTimeout) import Network.Wreq import System.IO import Text.Printf -data QueueUrls = QueueUrls { workerQueueUrl :: Text - , deadLetterQueueUrl :: Text - } +data DaemonOptions + = DaemonOptions { _workerUrl :: Text + , _workerQueueName :: Text + , _deadLetterQueueName :: Maybe Text + , _httpTimeout :: Maybe Int + , _contentType :: ByteString + , _sqsHost :: ByteString + , _sqsPort :: Int + , _forked :: Bool + } deriving (Show, Eq, Ord) +makeLenses ''DaemonOptions + +data WithUrls o + = WithUrls { _daemonOptions :: o + , _workerQueueUrl :: Text + , _deadLetterQueueUrl :: Maybe Text + } deriving (Show, Eq, Ord) + +makeLenses ''WithUrls + say :: MonadIO m => Text -> m () say = liftIO . Text.putStrLn + getQueueUrl :: (MonadCatch a, MonadResource a) => Text -> AWST a Text getQueueUrl = fmap (view gqursQueueURL) . send . getQueueURL + addAttributeHeaders :: Options -> HashMap Text MessageAttributeValue -> Options @@ -47,6 +82,7 @@ in opts' & header headerName .~ [encodeUtf8 value] Nothing -> opts' + deleteMessageByReceiptHandle :: (MonadCatch m, MonadResource m) => Text -> Text @@ -55,74 +91,93 @@ void (send (deleteMessage url receiptHandle)) liftIO (printf "Deleted message from queue: %s\n" url) + sendToDeadLetterQueueAndDelete :: (MonadCatch m, MonadResource m) - => QueueUrls + => WithUrls DaemonOptions -> Text -> Text -> HashMap Text MessageAttributeValue -> AWST m () -sendToDeadLetterQueueAndDelete urls receiptHandle body attrs = do - let dlqMsg = sendMessage (deadLetterQueueUrl urls) body - & smMessageAttributes .~ attrs - void (send dlqMsg) - say "Sent message to dead letter queue, deleting." - void (send (deleteMessage (workerQueueUrl urls) receiptHandle)) +sendToDeadLetterQueueAndDelete opts receiptHandle body attrs = do + -- Only send to dead letter queue if specified. + case opts ^. deadLetterQueueUrl of + Just dlqUrl -> do + let dlqMsg = sendMessage dlqUrl body + & smMessageAttributes .~ attrs + void (send dlqMsg) + say "Sent message to dead letter queue." + Nothing -> return () + say "Deleting handled message from worker queue." + void (send (deleteMessage (opts ^. workerQueueUrl) receiptHandle)) + + forwardMessage :: (MonadCatch m, MonadResource m) - => Text - -> QueueUrls + => WithUrls DaemonOptions -> Message -> AWST m () -forwardMessage workerUrl urls msg = +forwardMessage o msg = case (msg ^. mReceiptHandle, msg ^. mBody) of (Just receiptHandle, Just body) -> do let attrs = msg ^. mMessageAttributes + timeout' = (* 1000000) <$> o ^. daemonOptions . httpTimeout opts = defaults `addAttributeHeaders` attrs - & header "content-type" .~ ["application/json"] - req = postWith opts (unpack workerUrl) (encodeUtf8 body) + & header "content-type" .~ [o ^. daemonOptions . contentType] + & manager .~ Left (defaultManagerSettings { managerResponseTimeout = timeout' }) + req = postWith opts (Text.unpack (o ^. daemonOptions . workerUrl)) (encodeUtf8 body) result <- liftIO (try req) case result of Right res | res ^. responseStatus . statusCode == 200 -> - deleteMessageByReceiptHandle (workerQueueUrl urls) receiptHandle + deleteMessageByReceiptHandle (o ^. workerQueueUrl) receiptHandle Right res -> do liftIO (printf "Worker failed with status code: %d\n" (res ^. responseStatus . statusCode)) - sendToDeadLetterQueueAndDelete urls receiptHandle body attrs + sendToDeadLetterQueueAndDelete o receiptHandle body attrs Left (e :: HttpException) -> do liftIO (printf "Worker request failed: %s\n" (show e)) - sendToDeadLetterQueueAndDelete urls receiptHandle body attrs + sendToDeadLetterQueueAndDelete o receiptHandle body attrs (Nothing, _) -> say "Message had no reciept handle, ignoring." (_, Nothing) -> say "Message had no body, ignoring." -receiveNext :: (MonadCatch m, MonadResource m) - => Text - -> QueueUrls + +receiveNext :: (MonadCatch m, MonadResource m, MonadBaseControl IO m) + => WithUrls DaemonOptions -> AWST m () -receiveNext workerUrl urls = do +receiveNext o = do msgs <- view rmrsMessages - <$> send (receiveMessage (workerQueueUrl urls) & rmWaitTimeSeconds ?~ 20) - unless (null msgs) $ - liftIO (printf "Received %d new messages, posting to worker.\n" (length msgs)) - forM_ msgs (forwardMessage workerUrl urls) + <$> send (receiveMessage (o ^. workerQueueUrl) + & rmMaxNumberOfMessages ?~ 10 + & rmWaitTimeSeconds ?~ 20) -data DaemonOptions - = DaemonOptions { workerUrl :: Text - , workerQueueName :: Text - , deadletterQueueName :: Text - } + unless (Prelude.null msgs) $ + let forkedMsg = if o ^. daemonOptions . forked + then "concurrent" + else "serialized" + in liftIO (printf "Received %d new messages, posting to worker (%s).\n" + (Prelude.length msgs) + (forkedMsg :: String)) + let process = if o ^. daemonOptions . forked then void . fork else void + forM_ msgs (process . forwardMessage o) + + startDaemon :: DaemonOptions -> IO () -startDaemon (DaemonOptions {workerUrl, workerQueueName, deadletterQueueName}) = do - let region = Ireland +startDaemon opts = do lgr <- newLogger Info stdout - env <- newEnv region Discover <&> set envLogger lgr + env <- newEnv Ireland Discover <&> set envLogger lgr - let sqsEndpoint = setEndpoint False "localhost" 9324 sqs + let sqsEndpoint = setEndpoint False (opts ^. sqsHost) (opts ^. sqsPort) sqs - runResourceT . runAWST env . within region $ + runResourceT . runAWST env $ reconfigure sqsEndpoint $ do + o <- WithUrls opts + <$> getQueueUrl (opts ^. workerQueueName) + <*> maybe (return Nothing) (fmap Just . getQueueUrl) (opts ^. deadLetterQueueName) + + when (isNothing (o ^. deadLetterQueueUrl)) $ + say "No dead letter queue specified." + say "Starting local SQSD. Awaiting messages..." - urls <- QueueUrls <$> getQueueUrl workerQueueName <*> getQueueUrl deadletterQueueName - void (forever (receiveNext workerUrl urls)) + void (forever (receiveNext o))