postgresql-simple-queue 0.2.0.0 → 0.2.1.0

raw patch · 3 files changed

+10/−26 lines, 3 files

Files

• postgresql-simple-queue.cabal +1/−1
• src/Database/PostgreSQL/Simple/Queue.hs +8/−24
• src/Database/PostgreSQL/Simple/Queue/Migrate.hs +1/−1

@@ -1,5 +1,5 @@ name:                postgresql-simple-queue-version:             0.2.0.0+version:             0.2.1.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.

@@ -68,15 +68,12 @@   ) 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                               (UUID) import           Database.PostgreSQL.Simple              (Connection, Only (..)) import qualified Database.PostgreSQL.Simple              as Simple import           Database.PostgreSQL.Simple.FromField@@ -87,13 +84,12 @@ import           Database.PostgreSQL.Simple.ToRow import           Database.PostgreSQL.Simple.Transaction import           Database.PostgreSQL.Transact-import           System.Random import           Data.Monoid import           Data.String ------------------------------------------------------------------------------- ---  Types --------------------------------------------------------------------------------newtype PayloadId = PayloadId { unPayloadId :: UUID }+newtype PayloadId = PayloadId { unPayloadId :: Int64 }   deriving (Eq, Show, FromField, ToField)  instance FromRow PayloadId where@@ -153,16 +149,6 @@ notifyName :: IsString s => String -> s notifyName schemaName = fromString $ schemaName <> "_enqueue" -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.@@ -175,16 +161,14 @@  @ -} enqueueDB :: String -> Value -> DB PayloadId-enqueueDB schemaName value = retryOnUniqueViolation $ do-  pid <- liftIO randomIO-  execute (withSchema schemaName $ [sql|-    INSERT INTO payloads (id, value)-    VALUES (?, ?);-    NOTIFY |] <> " " <> notifyName schemaName <>-    ";"+enqueueDB schemaName value =+  fmap head $ query (withSchema schemaName $ [sql|+    NOTIFY |] <> " " <> notifyName schemaName <> ";" <> [sql|+    INSERT INTO payloads (value)+    VALUES (?)+    RETURNING id;|]     )-    (pid, value)-  return $ PayloadId pid+    $ Only value  {-| Return a the oldest 'Payload' in the 'Enqueued' state, or 'Nothing'     if there are no payloads. This function is not necessarily useful by

@@ -44,7 +44,7 @@   CREATE TYPE state_t AS ENUM ('enqueued', 'locked', 'dequeued');    CREATE TABLE payloads-  ( id uuid PRIMARY KEY+  ( id SERIAL PRIMARY KEY   , value jsonb NOT NULL   , state state_t NOT NULL DEFAULT 'enqueued'   , created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT clock_timestamp()