Database.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

  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 defaultMain, see 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.

Types

newtype PayloadId

Constructors

PayloadId

Fields

• unPayloadId :: Int64

Instances

Eq PayloadId

Methods

(==) :: PayloadId -> PayloadId -> Bool

(/=) :: PayloadId -> PayloadId -> Bool

Show PayloadId

Methods

showsPrec :: Int -> PayloadId -> ShowS

show :: PayloadId -> String

showList :: [PayloadId] -> ShowS

ToRow PayloadId

Methods

toRow :: PayloadId -> [Action]

ToField PayloadId

Methods

toField :: PayloadId -> Action

FromRow PayloadId

Methods

fromRow :: RowParser PayloadId

FromField PayloadId

Methods

fromField :: FieldParser PayloadId

data State

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.

Constructors

Enqueued

Locked

Dequeued

Instances

Bounded State

Methods

minBound :: State

maxBound :: State

Enum State

Methods

succ :: State -> State

pred :: State -> State

toEnum :: Int -> State

fromEnum :: State -> Int

enumFrom :: State -> [State]

enumFromThen :: State -> State -> [State]

enumFromTo :: State -> State -> [State]

enumFromThenTo :: State -> State -> State -> [State]

Eq State

Methods

(==) :: State -> State -> Bool

(/=) :: State -> State -> Bool

Ord State

Methods

compare :: State -> State -> Ordering

(<) :: State -> State -> Bool

(<=) :: State -> State -> Bool

(>) :: State -> State -> Bool

(>=) :: State -> State -> Bool

max :: State -> State -> State

min :: State -> State -> State

Show State

Methods

showsPrec :: Int -> State -> ShowS

show :: State -> String

showList :: [State] -> ShowS

ToField State

Methods

toField :: State -> Action

FromField State

Methods

fromField :: FieldParser State

data Payload

Constructors

Payload

Fields

• pId :: PayloadId
• pValue :: Value

  The JSON value of a payload

• pState :: State
• pCreatedAt :: UTCTime
• pModifiedAt :: UTCTime

Instances

Eq Payload

Methods

(==) :: Payload -> Payload -> Bool

(/=) :: Payload -> Payload -> Bool

Show Payload

Methods

showsPrec :: Int -> Payload -> ShowS

show :: Payload -> String

showList :: [Payload] -> ShowS

FromRow Payload

Methods

fromRow :: RowParser Payload

DB API

enqueueDB :: String -> Value -> DB PayloadId

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

tryLockDB :: String -> DB (Maybe Payload)

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

unlockDB :: String -> PayloadId -> DB ()

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.

dequeueDB :: String -> PayloadId -> DB ()

Transition a Payload to the Dequeued state.

getCountDB :: String -> DB Int64

Get the number of rows in the Enqueued state.

IO API

enqueue :: String -> Connection -> Value -> IO PayloadId

Enqueue a new JSON value into the queue. See enqueueDB for a version which can be composed with other queries in a single transaction.

tryLock :: String -> Connection -> IO (Maybe Payload)

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.

lock :: String -> Connection -> IO Payload

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.

unlock :: String -> Connection -> PayloadId -> IO ()

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

dequeue :: String -> Connection -> PayloadId -> IO ()

Transition a Payload to the Dequeued state. his functions runs dequeueDB as a Serializable transaction.

getCount :: String -> Connection -> IO Int64

Get the number of rows in the Enqueued state. This function runs getCountDB in a ReadCommitted transaction.