postgresql-simple-queue 0.4.0.0 → 0.5.0.0
raw patch · 4 files changed
+1/−312 lines, 4 filesdep −amazonkadep −amazonka-sesdep −data-defaultdep ~base
Dependencies removed: amazonka, amazonka-ses, data-default, lens, lifted-async, lifted-base, optparse-generic, postgresql-simple-opts, resource-pool, uuid
Dependency ranges changed: base
Files
- examples/EmailQueue.hs +0/−48
- postgresql-simple-queue.cabal +1/−25
- src/Database/PostgreSQL/Simple/Queue/Examples/EmailQueue.hs +0/−59
- src/Database/PostgreSQL/Simple/Queue/Main.hs +0/−180
− examples/EmailQueue.hs
@@ -1,48 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE DeriveAnyClass #-}-{-# LANGUAGE RecordWildCards #-}-import Control.Exception.Lifted-import Control.Lens-import Control.Monad-import Control.Monad.IO.Class-import Data.Aeson as Aeson-import Data.Text-import Database.PostgreSQL.Simple.Queue-import Database.PostgreSQL.Simple.Queue.Main-import GHC.Generics-import Network.AWS as AWS-import Network.AWS.SES.SendEmail as AWS-import Network.AWS.SES.Types as AWS--main :: IO ()-main = do- env <- newEnv Discover- runResourceT $ runAWS env $ defaultMain "aws-email-queue-consumer" $ \payload _ -> do- case fromJSON $ pValue payload of- Aeson.Success email -> do- resp <- AWS.send $ makeEmail email- logFailedRequest resp- Aeson.Error x -> throwIO- $ userError- $ "Failed to decode payload as an Email: " ++ show x--data Email = Email- { emailAddress :: Text- , emailSubject :: Text- , emailBody :: Text- } deriving (Show, Eq, Generic, FromJSON, ToJSON)--makeEmail :: Email -> SendEmail-makeEmail Email {..}- = sendEmail emailAddress- (set dToAddresses [emailAddress] destination)- $ message (content emailSubject)- $ set bText (Just $ content emailBody) AWS.body--logFailedRequest :: MonadIO m => SendEmailResponse -> m ()-logFailedRequest resp = do- let stat = view sersResponseStatus resp-- unless (stat >= 200 && stat < 300) $- liftIO $ putStrLn $ "SES failed with status: " ++ show stat
postgresql-simple-queue.cabal view
@@ -1,5 +1,5 @@ name: postgresql-simple-queue-version: 0.4.0.0+version: 0.5.0.0 synopsis: A PostgreSQL backed queue description: This module utilize PostgreSQL to implement a durable queue for efficently processing arbitrary payloads which can be represented as JSON.@@ -42,44 +42,20 @@ library hs-source-dirs: src exposed-modules: Database.PostgreSQL.Simple.Queue- , Database.PostgreSQL.Simple.Queue.Examples.EmailQueue , Database.PostgreSQL.Simple.Queue.Migrate- , Database.PostgreSQL.Simple.Queue.Main build-depends: base >= 4.7 && < 5 , postgresql-simple , pg-transact , aeson , time- , uuid , transformers , random , text , monad-control- , postgresql-simple-opts- , resource-pool- , optparse-generic >= 1.2.1- , data-default- , monad-control- , lifted-base- , lifted-async , exceptions , bytestring default-language: Haskell2010 ghc-options: -Wall -Wno-unused-do-bind--executable async-email-example- hs-source-dirs: examples- main-is: EmailQueue.hs- ghc-options: -threaded -rtsopts -with-rtsopts=-N- build-depends: base- , postgresql-simple-queue- , amazonka- , amazonka-ses- , lens- , text- , aeson- , lifted-base- default-language: Haskell2010 test-suite unit-tests type: exitcode-stdio-1.0
− src/Database/PostgreSQL/Simple/Queue/Examples/EmailQueue.hs
@@ -1,59 +0,0 @@-{-|- This is a documentation only module that exposes the source for- the async-email-example. The main file is located in examples/EmailQueue- and is reprinted here to show an example of using 'defaultMain' from- 'Database.PostgreSQL.Simple.Queue.Main' to quickly build a queue consumer executable.-- @- import Control.Exception.Lifted- import Control.Lens- import Control.Monad- import Control.Monad.IO.Class- import Data.Aeson as Aeson- import Data.Text- import Database.PostgreSQL.Simple.Queue- import Database.PostgreSQL.Simple.Queue.Main- import GHC.Generics- import Network.AWS as AWS- import Network.AWS.SES.SendEmail as AWS- import Network.AWS.SES.Types as AWS-- main :: IO ()- main = do- env <- newEnv Discover- runResourceT $ runAWS env $ 'defaultMain' "aws-email-queue-consumer" $ \payload _ -> do- case fromJSON $ 'pValue' payload of- Aeson.Success email -> do- resp <- AWS.send $ makeEmail email- logFailedRequest resp- Aeson.Error x -> throwIO- $ userError- $ "Failed to decode payload as an Email: " ++ show x-- data Email = Email- { emailAddress :: Text- , emailSubject :: Text- , emailBody :: Text- } deriving (Show, Eq, Generic, FromJSON, ToJSON)-- makeEmail :: Email -> SendEmail- makeEmail Email {..}- = sendEmail emailAddress- (set dToAddresses [emailAddress] destination)- $ message (content emailSubject)- $ set bText (Just $ content emailBody) AWS.body-- logFailedRequest :: MonadIO m => SendEmailResponse -> m ()- logFailedRequest resp = do- let stat = view sersResponseStatus resp-- unless (stat >= 200 && stat < 300) $- liftIO $ putStrLn $ "SES failed with status: " ++ show stat- @---}-{-# OPTIONS_GHC -w #-}-module Database.PostgreSQL.Simple.Queue.Examples.EmailQueue where-import Database.PostgreSQL.Simple.Queue-import Database.PostgreSQL.Simple.Queue.Main-
− src/Database/PostgreSQL/Simple/Queue/Main.hs
@@ -1,180 +0,0 @@-{-# LANGUAGE DeriveDataTypeable #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NamedFieldPuns #-}-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE OverloadedStrings #-}-{-|- This module provides a simple way to create executables for consuming queue- payloads. It provides a `defaultMain' function which takes in a executable- name and processing function. It returns a main function which will parse- database options on from the command line and spawn a specified number of- worker threads.-- Here is a simple example that logs out queue payloads-- @- defaultMain "queue-logger" $ \payload count -> do- print payload- print count- @-- The worker threads do not attempt to handle exceptions. If an exception is- thrown on any threads, all threads are cancel and 'defaultMain' returns. The- assumption is the queue executable will get run by a process watcher that- can log failures.-- For a more complicated example, see the code for the async-email-example- documented in 'Database.PostgreSQL.Simple.Queue.Examples.EmailQueue.EmailQueue'.--}-module Database.PostgreSQL.Simple.Queue.Main- ( -- * Options- PartialOptions- , Options- , completeOptions- -- * Entry Points- , defaultMain- , run- ) where-import Control.Concurrent.Async.Lifted-import Control.Monad-import Control.Monad.IO.Class-import Control.Monad.Trans.Control-import Data.Default-import Data.Int-import Data.Monoid-import Data.Pool-import Data.Typeable-import Database.PostgreSQL.Simple-import qualified Database.PostgreSQL.Simple.Options as PostgreSQL-import Database.PostgreSQL.Simple.Queue-import Options.Generic-import System.Exit--{-| The 'PartialOptions' provide a 'Monoid' for combining options used by- 'defaultMain'. The following command line arguments are used to- construct a 'PartialOptions'-- @- --thread-count INT-- Either a connection string- --connectString ARG- or individual db properties are provided- --host STRING- --port INT- --user STRING- --password STRING- --database STRING- @--}-data PartialOptions = PartialOptions- { threadCount :: Last Int- , dbOptions :: PostgreSQL.PartialOptions- , schemaName :: Last String- } deriving (Show, Eq, Typeable)--instance Monoid PartialOptions where- mempty = PartialOptions mempty mempty mempty- mappend x y =- PartialOptions- (threadCount x <> threadCount y)- (dbOptions x <> dbOptions y)- (schemaName x <> schemaName y)---- | The default 'threadCount' is 1.--- The default db options are specified in--- 'Database.PostgreSQL.Simple.Options.PartialOptions'-instance Default PartialOptions where- def = PartialOptions (return 1) def $ return "queue"--instance ParseRecord PartialOptions where- parseRecord- = PartialOptions- <$> parseFields Nothing (Just "thread-count") (Just 't')- <*> parseRecord- <*> parseFields Nothing (Just "schema-name") (Just 's')---- | Final Options used by 'run'.-data Options = Options- { oThreadCount :: Int- , oDBOptions :: PostgreSQL.Options- , oSchemaName :: String- } deriving (Show, Eq)---- | Convert a 'PartialOptions' to a final 'Options'-completeOptions :: PartialOptions -> Either [String] Options-completeOptions = \case- PartialOptions { threadCount = Last (Just oThreadCount), dbOptions, schemaName = Last (Just oSchemaName) } ->- Options oThreadCount <$> PostgreSQL.completeOptions dbOptions <*> pure oSchemaName-- _ -> Left ["Missing threadCount"]--{-| This function is a helper for creating queue consumer executables.- It takes in a executable- name and processing function. It returns a main function which will parse- database options on from the command line and spawn a specified number of- worker threads. See 'PartialOptions' for command line documentation.-- Here is a simple example that logs out queue payloads-- @- defaultMain "queue-logger" $ \payload count -> do- print payload- print count- @-- The worker threads do not attempt to handle exceptions. If an exception is- thrown on any threads, all threads are cancel and 'defaultMain' returns. The- assumption is the queue executable will get run by a process watcher that- can log failures.-- For a more complicated example, see the code for the async-email-example- documented in 'Database.PostgreSQL.Simple.Queue.Examples.EmailQueue.EmailQueue'.---}-defaultMain :: (MonadIO m, MonadBaseControl IO m)- => Text- -- ^ Executable name. Used when command line help- -- is printed- -> (Payload -> Int64 -> m ())- -- ^ Processing function. Takes a 'Payload' to process- -- and the current number of 'Enqueued' 'Payload's- -- remaining.- -> m ()-defaultMain name f = do- poptions <- liftIO $ getRecord name- options <- liftIO- $ either (\err -> putStrLn (unlines err) >> exitWith (ExitFailure 64))- return- $ completeOptions $ def <> poptions- run f options---- | 'run' is called by 'defaultMain' after command line argument parsing--- is performed. Useful is wants to embed consumer threads inside a--- larger application.-run :: forall m. (MonadIO m, MonadBaseControl IO m)- => (Payload -> Int64 -> m ())- -- ^ Processing function. Takes a 'Payload' to process- -- and the current number of 'Enqueued' 'Payload's- -- remaining.- -> Options- -- ^ Options for thread creation and db connections.- -> m ()-run f Options {..} = do- -- Creates a pool with half as many connections as threads. Should- -- probably make the number of connection configurable.- connectionPool <- liftIO $ createPool- (PostgreSQL.run oDBOptions) close 1 60 (max 1 $ oThreadCount `div` 2)-- threads :: [Async (StM m ())] <- replicateM oThreadCount $ async $ void $- forever $ do- payload <- liftIO $ withResource connectionPool $ lock oSchemaName- count <- liftIO $ withResource connectionPool $ getCount oSchemaName- f payload count- liftIO $ withResource connectionPool $ \c -> dequeue oSchemaName c (pId payload)-- _ :: (Async (StM m ()), ()) <- waitAnyCancel threads- return ()