diff --git a/examples/EmailQueue.hs b/examples/EmailQueue.hs deleted file mode 100644 --- a/examples/EmailQueue.hs +++ /dev/null @@ -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 diff --git a/postgresql-simple-queue.cabal b/postgresql-simple-queue.cabal --- a/postgresql-simple-queue.cabal +++ b/postgresql-simple-queue.cabal @@ -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 diff --git a/src/Database/PostgreSQL/Simple/Queue/Examples/EmailQueue.hs b/src/Database/PostgreSQL/Simple/Queue/Examples/EmailQueue.hs deleted file mode 100644 --- a/src/Database/PostgreSQL/Simple/Queue/Examples/EmailQueue.hs +++ /dev/null @@ -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 - diff --git a/src/Database/PostgreSQL/Simple/Queue/Main.hs b/src/Database/PostgreSQL/Simple/Queue/Main.hs deleted file mode 100644 --- a/src/Database/PostgreSQL/Simple/Queue/Main.hs +++ /dev/null @@ -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 ()