postgresql-simple-queue (empty) → 0.1.0.0
raw patch · 11 files changed
+886/−0 lines, 11 filesdep +aesondep +amazonkadep +amazonka-sessetup-changed
Dependencies added: aeson, amazonka, amazonka-ses, async, base, bytestring, data-default, exceptions, hspec, hspec-discover, hspec-expectations-lifted, hspec-pg-transact, lens, lifted-async, lifted-base, monad-control, optparse-generic, pg-transact, postgresql-simple, postgresql-simple-opts, postgresql-simple-queue, random, resource-pool, text, time, transformers, uuid
Files
- LICENSE +30/−0
- README.md +28/−0
- Setup.hs +2/−0
- examples/EmailQueue.hs +48/−0
- postgresql-simple-queue.cabal +102/−0
- src/Database/PostgreSQL/Simple/Queue.hs +282/−0
- src/Database/PostgreSQL/Simple/Queue/Examples/EmailQueue.hs +59/−0
- src/Database/PostgreSQL/Simple/Queue/Main.hs +175/−0
- src/Database/PostgreSQL/Simple/Queue/Migrate.hs +65/−0
- test/Database/PostgreSQL/Simple/QueueSpec.hs +94/−0
- test/Spec.hs +1/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Author name here (c) 2017++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Author name here nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,28 @@+# postgresql-simple-queue++This module utilize PostgreSQL to implement a durable queue for efficently processing arbitrary payloads which can be represented as JSON.++Typically a producer would enqueue a new payload as part of larger database transaction++```haskell+createAccount userRecord = do+ runDBTSerializable $ do+ createUserDB userRecord+ enqueueDB $ makeVerificationEmail userRecord+```++In another thread or process, the consumer would drain the queue.++```haskell+ forever $ do+ -- Attempt get a payload or block until one is available+ payload <- lock conn++ -- Perform application specifc parsing of the payload value+ case fromJSON $ pValue payload of+ Success x -> sendEmail x -- Perform application specific processing+ Error err -> logErr err++ -- Remove the payload from future processing+ dequeue conn $ pId payload+```
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ examples/EmailQueue.hs view
@@ -0,0 +1,48 @@+{-# 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
@@ -0,0 +1,102 @@+name: postgresql-simple-queue+version: 0.1.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.+ .+ Typically a producer would enqueue a new payload as part of larger database+ transaction+ .+ > createAccount userRecord = do+ > 'runDBTSerializable' $ do+ > createUserDB userRecord+ > 'enqueueDB' $ makeVerificationEmail userRecord+ .+ In another thread or process, the consumer would drain the queue.+ .+ > forever $ do+ > -- Attempt get a payload or block until one is available+ > payload <- 'lock' conn+ >+ > -- Perform application specifc parsing of the payload value+ > case fromJSON $ 'pValue' payload of+ > Success x -> sendEmail x -- Perform application specific processing+ > Error err -> logErr err+ >+ > -- Remove the payload from future processing+ > 'dequeue' conn $ 'pId' payload+homepage: https://github.com/jfischoff/postgresql-queue#readme+license: BSD3+license-file: LICENSE+author: Jonathan Fischoff+maintainer: jonathangfischoff@gmail.com+copyright: 2017 Jonathan Fischoff+category: Web+build-type: Simple+extra-source-files: README.md+cabal-version: >=1.10++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+ , 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 db-testing-example-test+ type: exitcode-stdio-1.0+ hs-source-dirs: test+ main-is: Spec.hs+ other-modules: Database.PostgreSQL.Simple.QueueSpec+ build-depends: base+ , postgresql-simple-queue+ , hspec+ , hspec-discover+ , postgresql-simple+ , pg-transact+ , bytestring+ , aeson+ , hspec-expectations-lifted+ , hspec-pg-transact+ , async+ ghc-options: -Wall -Wno-unused-do-bind -O2 -threaded -rtsopts -with-rtsopts=-N+ default-language: Haskell2010++source-repository head+ type: git+ location: https://github.com/jfischoff/postgresql-queue
+ src/Database/PostgreSQL/Simple/Queue.hs view
@@ -0,0 +1,282 @@+{-| This module utilize PostgreSQL to implement a durable queue for efficently processing+ arbitrary payloads which can be represented as JSON.++ Typically a producer would enqueue a new payload as part of larger database+ transaction++ @+ createAccount userRecord = do+ 'runDBTSerializable' $ do+ createUserDB userRecord+ 'enqueueDB' $ makeVerificationEmail userRecord+ @++In another thread or process, the consumer would drain the queue.++ @+ forever $ do+ -- Attempt get a payload or block until one is available+ payload <- 'lock' conn++ -- Perform application specifc parsing of the payload value+ case fromJSON $ 'pValue' payload of+ Success x -> sendEmail x -- Perform application specific processing+ Error err -> logErr err++ -- Remove the payload from future processing+ 'dequeue' conn $ 'pId' payload+ @++ For a more complete example or a consumer, utilizing the provided+ 'Database.PostgreSQL.Simple.Queue.Main.defaultMain', see+ 'Database.PostgreSQL.Simple.Queue.Examples.EmailQueue.EmailQueue'.++This modules provides two flavors of functions, a DB API and an IO API.+Most operations are provided in both flavors, with the exception of 'lock'.+'lock' blocks and would not be that useful as part of a larger transaction+since it would keep the transaction open for a potentially long time. Although+both flavors are provided, in general one versions is more useful for typical+use cases.++-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+module Database.PostgreSQL.Simple.Queue+ ( -- * Types+ PayloadId (..)+ , State (..)+ , Payload (..)+ -- * DB API+ , enqueueDB+ , tryLockDB+ , unlockDB+ , dequeueDB+ , getCountDB+ -- * IO API+ , enqueue+ , tryLock+ , lock+ , unlock+ , dequeue+ , getCount+ ) where+import Control.Monad+import Control.Monad.Catch+import Control.Monad.IO.Class+import Data.Aeson+import qualified Data.ByteString as BS+import Data.Function+import Data.Int+import Data.Maybe+import Data.Text (Text)+import Data.Time+import Data.UUID+import Database.PostgreSQL.Simple (Connection, Only (..))+import qualified Database.PostgreSQL.Simple as Simple+import Database.PostgreSQL.Simple.FromField+import Database.PostgreSQL.Simple.FromRow+import Database.PostgreSQL.Simple.Notification+import Database.PostgreSQL.Simple.SqlQQ+import Database.PostgreSQL.Simple.ToField+import Database.PostgreSQL.Simple.ToRow+import Database.PostgreSQL.Simple.Transaction+import Database.PostgreSQL.Transact+import System.Random+-------------------------------------------------------------------------------+--- Types+-------------------------------------------------------------------------------+newtype PayloadId = PayloadId { unPayloadId :: UUID }+ deriving (Eq, Show, FromField, ToField)++instance FromRow PayloadId where+ fromRow = fromOnly <$> fromRow++instance ToRow PayloadId where+ toRow = toRow . Only++-- | A 'Payload' can exist in three states in the queue, 'Enqueued', 'Locked'+-- and 'Dequeued'. A 'Payload' starts in the 'Enqueued' state and is 'Locked'+-- so some sort of process can occur with it, usually something in 'IO'.+-- Once the processing is complete, the `Payload' is moved the 'Dequeued'+-- state, which is the terminal state.+data State = Enqueued | Locked | Dequeued+ deriving (Show, Eq, Ord, Enum, Bounded)++instance ToField State where+ toField = toField . \case+ Enqueued -> "enqueued" :: Text+ Locked -> "locked"+ Dequeued -> "dequeued"++-- Converting from enumerations is annoying :(+instance FromField State where+ fromField f y = do+ n <- typename f+ if n == "state_t" then case y of+ Nothing -> returnError UnexpectedNull f "state can't be NULL"+ Just y' -> case y' of+ "enqueued" -> return Enqueued+ "locked" -> return Locked+ "dequeued" -> return Dequeued+ x -> returnError ConversionFailed f (show x)+ else+ returnError Incompatible f $+ "Expect type name to be state but it was " ++ show n++-- The fundemental record stored in the queue. The queue is a single table+-- and each row consists of a 'Payload'+data Payload = Payload+ { pId :: PayloadId+ , pValue :: Value+ -- ^ The JSON value of a payload+ , pState :: State+ , pCreatedAt :: UTCTime+ , pModifiedAt :: UTCTime+ } deriving (Show, Eq)++instance FromRow Payload where+ fromRow = Payload <$> field <*> field <*> field <*> field <*> field+-------------------------------------------------------------------------------+--- DB API+-------------------------------------------------------------------------------+retryOnUniqueViolation :: MonadCatch m => m a -> m a+retryOnUniqueViolation act = try act >>= \case+ Right x -> return x+ Left e ->+ if Simple.sqlState e == "23505" &&+ "duplicate key" `BS.isPrefixOf` Simple.sqlErrorMsg e then+ act+ else+ throwM e++{-| Enqueue a new JSON value into the queue. This particularly function+ can be composed as part of a larger database transaction. For instance,+ a single transaction could create a user and enqueue a email message.++ @+ createAccount userRecord = do+ 'runDBTSerializable' $ do+ createUserDB userRecord+ 'enqueueDB' $ makeVerificationEmail userRecord+ @+-}+enqueueDB :: Value -> DB PayloadId+enqueueDB value = retryOnUniqueViolation $ do+ pid <- liftIO randomIO+ execute [sql| INSERT INTO payloads (id, value)+ VALUES (?, ?);+ NOTIFY enqueue;+ |]+ (pid, value)+ return $ PayloadId pid++{-| Return a the oldest 'Payload' in the 'Enqueued' state, or 'Nothing'+ if there are no payloads. This function is not necessarily useful by+ itself, since there are not many use cases where it needs to be combined+ with other transactions. See 'tryLock' the IO API version, or for a+ blocking version utilizing PostgreSQL's NOTIFY and LISTEN, see 'lock'+-}+tryLockDB :: DB (Maybe Payload)+tryLockDB = listToMaybe <$> query_+ [sql| UPDATE payloads+ SET state='locked'+ WHERE id in+ ( SELECT id+ FROM payloads+ WHERE state='enqueued'+ ORDER BY created_at ASC+ LIMIT 1+ )+ RETURNING id, value, state, created_at, modified_at+ |]++{-| Transition a 'Payload' from the 'Locked' state to the 'Enqueued' state.+ Useful for responding to asynchronous exceptions during a unexpected+ shutdown. In general the IO API version, 'unlock', is probably more+ useful. The DB version is provided for completeness.+-}+unlockDB :: PayloadId -> DB ()+unlockDB payloadId = void $ execute+ [sql| UPDATE payloads+ SET state='enqueued'+ WHERE id=? AND state='locked'+ |]+ payloadId++-- | Transition a 'Payload' to the 'Dequeued' state.+dequeueDB :: PayloadId -> DB ()+dequeueDB payloadId = void $ execute+ [sql| UPDATE payloads+ SET state='dequeued'+ WHERE id=?+ |]+ payloadId++-- | Get the number of rows in the 'Enqueued' state.+getCountDB :: DB Int64+getCountDB = fromOnly . head <$> query_+ [sql| SELECT count(*)+ FROM payloads+ WHERE state='enqueued'+ |]+-------------------------------------------------------------------------------+--- IO API+-------------------------------------------------------------------------------+{-| Enqueue a new JSON value into the queue. See 'enqueueDB' for a version+ which can be composed with other queries in a single transaction.+-}+enqueue :: Connection -> Value -> IO PayloadId+enqueue conn value = runDBT (enqueueDB value) ReadCommitted conn++{-| Return a the oldest 'Payload' in the 'Enqueued' state or 'Nothing'+ if there are no payloads. For a blocking version utilizing PostgreSQL's+ NOTIFY and LISTEN, see 'lock'. This functions runs 'tryLockDB' as a+ 'Serializable' transaction.+-}+tryLock :: Connection -> IO (Maybe Payload)+tryLock = runDBTSerializable tryLockDB++notifyPayload :: Connection -> IO ()+notifyPayload conn = do+ Notification {..} <- getNotification conn+ unless (notificationChannel == "enqueue") $ notifyPayload conn++{-| Return the oldest 'Payload' in the 'Enqueued' state or block until a+ payload arrives. This function utilizes PostgreSQL's LISTEN and NOTIFY+ functionality to avoid excessively polling of the DB while+ waiting for new payloads, without scarficing promptness.+-}+lock :: Connection -> IO Payload+lock conn = bracket_+ (Simple.execute_ conn "LISTEN enqueue")+ (Simple.execute_ conn "UNLISTEN enqueue")+ $ fix $ \continue -> do+ m <- tryLock conn+ case m of+ Nothing -> do+ notifyPayload conn+ continue+ Just x -> return x++{-| Transition a 'Payload' from the 'Locked' state to the 'Enqueued' state.+ Useful for responding to asynchronous exceptions during a unexpected+ shutdown. For a DB API version see 'unlockDB'+-}+unlock :: Connection -> PayloadId -> IO ()+unlock conn x = runDBTSerializable (unlockDB x) conn++-- | Transition a 'Payload' to the 'Dequeued' state. his functions runs+-- 'dequeueDB' as a 'Serializable' transaction.+dequeue :: Connection -> PayloadId -> IO ()+dequeue conn x = runDBTSerializable (dequeueDB x) conn++{-| Get the number of rows in the 'Enqueued' state. This function runs+ 'getCountDB' in a 'ReadCommitted' transaction.+-}+getCount :: Connection -> IO Int64+getCount = runDBT getCountDB ReadCommitted
+ src/Database/PostgreSQL/Simple/Queue/Examples/EmailQueue.hs view
@@ -0,0 +1,59 @@+{-|+ 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 view
@@ -0,0 +1,175 @@+{-# 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+ } deriving (Show, Eq, Typeable)++instance Monoid PartialOptions where+ mempty = PartialOptions mempty mempty+ mappend x y =+ PartialOptions+ (threadCount x <> threadCount y)+ (dbOptions x <> dbOptions 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++instance ParseRecord PartialOptions where+ parseRecord+ = PartialOptions+ <$> parseFields Nothing (Just "thread-count")+ <*> parseRecord++-- | Final Options used by 'run'.+data Options = Options+ { oThreadCount :: Int+ , oDBOptions :: PostgreSQL.Options+ } deriving (Show, Eq)++-- | Convert a 'PartialOptions' to a final 'Options'+completeOptions :: PartialOptions -> Either [String] Options+completeOptions = \case+ PartialOptions { threadCount = Last (Just oThreadCount), dbOptions } ->+ Options oThreadCount <$> PostgreSQL.completeOptions dbOptions+ _ -> 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+ count <- liftIO $ withResource connectionPool getCount+ f payload count+ liftIO $ withResource connectionPool $ flip dequeue (pId payload)++ _ :: (Async (StM m ()), ()) <- waitAnyCancel threads+ return ()
+ src/Database/PostgreSQL/Simple/Queue/Migrate.hs view
@@ -0,0 +1,65 @@+{-# LANGUAGE QuasiQuotes #-}+module Database.PostgreSQL.Simple.Queue.Migrate where+import Control.Monad+import Database.PostgreSQL.Simple+import Database.PostgreSQL.Simple.SqlQQ++{-| This function creates a table and enumeration type that is+ appriopiate for the queue. The following sql is used.++ @+ CREATE TYPE state_t AS ENUM ('enqueued', 'locked', 'dequeued');++ CREATE TABLE payloads+ ( id uuid PRIMARY KEY+ , value jsonb NOT NULL+ , state state_t NOT NULL DEFAULT 'enqueued'+ , created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT clock_timestamp()+ , modified_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT clock_timestamp()+ );++ CREATE INDEX state_idx ON payloads (state);++ CREATE OR REPLACE FUNCTION update_row_modified_function_()+ RETURNS TRIGGER+ AS+ $$+ BEGIN+ -- ASSUMES the table has a column named exactly "modified_at".+ -- Fetch date-time of actual current moment from clock,+ -- rather than start of statement or start of transaction.+ NEW.modified_at = clock_timestamp();+ RETURN NEW;+ END;+ $$+ language 'plpgsql';+ @+-}+migrate :: Connection -> IO ()+migrate conn = void $ execute_ conn [sql|+ CREATE TYPE state_t AS ENUM ('enqueued', 'locked', 'dequeued');++ CREATE TABLE payloads+ ( id uuid PRIMARY KEY+ , value jsonb NOT NULL+ , state state_t NOT NULL DEFAULT 'enqueued'+ , created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT clock_timestamp()+ , modified_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT clock_timestamp()+ );++ CREATE INDEX state_idx ON payloads (state);++ CREATE OR REPLACE FUNCTION update_row_modified_function_()+ RETURNS TRIGGER+ AS+ $$+ BEGIN+ -- ASSUMES the table has a column named exactly "modified_at".+ -- Fetch date-time of actual current moment from clock,+ -- rather than start of statement or start of transaction.+ NEW.modified_at = clock_timestamp();+ RETURN NEW;+ END;+ $$+ language 'plpgsql';+ |]
+ test/Database/PostgreSQL/Simple/QueueSpec.hs view
@@ -0,0 +1,94 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+module Database.PostgreSQL.Simple.QueueSpec (spec, main) where+import Control.Concurrent+import Control.Concurrent.Async+import Control.Monad+import Data.Aeson+import Data.Function+import Data.IORef+import Data.List+import Database.PostgreSQL.Simple.Queue+import Database.PostgreSQL.Simple.Queue.Migrate+import Test.Hspec (Spec, hspec, it)+import Test.Hspec.Expectations.Lifted+import Test.Hspec.DB++main :: IO ()+main = hspec spec++spec :: Spec+spec = describeDB migrate "Database.Queue" $ do+ itDB "empty locks nothing" $+ tryLockDB `shouldReturn` Nothing++ itDB "empty gives count 0" $+ getCountDB `shouldReturn` 0++ itDB "enqueues/tryLocks/unlocks" $ do+ payloadId <- enqueueDB $ String "Hello"+ getCountDB `shouldReturn` 1++ Just Payload {..} <- tryLockDB+ getCountDB `shouldReturn` 0++ pId `shouldBe` payloadId+ pValue `shouldBe` String "Hello"+ tryLockDB `shouldReturn` Nothing++ unlockDB pId+ getCountDB `shouldReturn` 1++ itDB "tryLocks/dequeues" $ do+ Just Payload {..} <- tryLockDB+ getCountDB `shouldReturn` 0++ dequeueDB pId `shouldReturn` ()+ tryLockDB `shouldReturn` Nothing++ it "enqueues and dequeues concurrently tryLock" $ \testDB -> do+ let withPool' = withPool testDB+ elementCount = 1000 :: Int+ expected = [0 .. elementCount - 1]++ ref <- newIORef []++ loopThreads <- replicateM 10 $ async $ fix $ \next -> do+ mpayload <- withPool' tryLock+ case mpayload of+ Nothing -> next+ Just Payload {..} -> do+ lastCount <- atomicModifyIORef ref+ $ \xs -> (pValue : xs, length xs + 1)+ withPool' $ flip dequeue pId+ when (lastCount < elementCount) next++ -- Fork a hundred threads and enqueue an index+ forM_ [0 .. elementCount - 1] $ \i ->+ forkIO $ void $ withPool' $ flip enqueue $ toJSON i++ waitAnyCancel loopThreads+ Just decoded <- mapM (decode . encode) <$> readIORef ref+ sort decoded `shouldBe` sort expected++ it "enqueues and dequeues concurrently lock" $ \testDB -> do+ let withPool' = withPool testDB+ elementCount = 1000 :: Int+ expected = [0 .. elementCount - 1]++ ref <- newIORef []++ loopThreads <- replicateM 10 $ async $ fix $ \next -> do+ Payload {..} <- withPool' lock+ lastCount <- atomicModifyIORef ref+ $ \xs -> (pValue : xs, length xs + 1)+ withPool' $ flip dequeue pId+ when (lastCount < elementCount) next++ -- Fork a hundred threads and enqueue an index+ forM_ [0 .. elementCount - 1] $ \i -> forkIO $ void $ withPool' $+ flip enqueue $ toJSON i++ waitAnyCancel loopThreads+ Just decoded <- mapM (decode . encode) <$> readIORef ref+ sort decoded `shouldBe` sort expected
+ test/Spec.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}