sqsd-local 0.1.1 → 0.2.0
raw patch · 4 files changed
+207/−56 lines, 4 filesdep +bytestringdep +http-clientdep +lifted-base
Dependencies added: bytestring, http-client, lifted-base
Files
- README.md +13/−6
- sqsd-local.cabal +4/−1
- src/Main.hs +96/−10
- src/Network/SQS/Daemon/Local.hs +94/−39
README.md view
@@ -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
sqsd-local.cabal view
@@ -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
src/Main.hs view
@@ -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
src/Network/SQS/Daemon/Local.hs view
@@ -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))