postgresql-simple-queue 0.2.1.0 → 0.3.0.0
raw patch · 5 files changed
+58/−63 lines, 5 files
Files
- postgresql-simple-queue.cabal +2/−2
- src/Database/PostgreSQL/Simple/Queue.hs +26/−29
- src/Database/PostgreSQL/Simple/Queue/Main.hs +3/−3
- src/Database/PostgreSQL/Simple/Queue/Migrate.hs +5/−7
- test/Database/PostgreSQL/Simple/QueueSpec.hs +22/−22
postgresql-simple-queue.cabal view
@@ -1,5 +1,5 @@ name: postgresql-simple-queue-version: 0.2.1.0+version: 0.3.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.@@ -26,7 +26,7 @@ > -- Remove the payload from future processing > dequeue "queue_schema" conn $ pId payload >- > To support multiple queues in the same database, the API expects a schema name string+ > To support multiple queues in the same database, the API expects a table name string > to determine which queue tables to use. homepage: https://github.com/jfischoff/postgresql-queue#readme license: BSD3
src/Database/PostgreSQL/Simple/Queue.hs view
@@ -143,11 +143,8 @@ ------------------------------------------------------------------------------- --- DB API --------------------------------------------------------------------------------withSchema :: String -> Simple.Query -> Simple.Query-withSchema schemaName q = "SET search_path TO " <> fromString schemaName <> "; " <> q- notifyName :: IsString s => String -> s-notifyName schemaName = fromString $ schemaName <> "_enqueue"+notifyName tableName = fromString $ tableName <> "_enqueue" {-| Enqueue a new JSON value into the queue. This particularly function can be composed as part of a larger database transaction. For instance,@@ -161,10 +158,10 @@ @ -} enqueueDB :: String -> Value -> DB PayloadId-enqueueDB schemaName value =- fmap head $ query (withSchema schemaName $ [sql|- NOTIFY |] <> " " <> notifyName schemaName <> ";" <> [sql|- INSERT INTO payloads (value)+enqueueDB tableName value =+ fmap head $ query ([sql|+ NOTIFY |] <> " " <> notifyName tableName <> ";" <> [sql|+ INSERT INTO|] <> " " <> fromString tableName <> " " <> [sql|(value) VALUES (?) RETURNING id;|] )@@ -177,12 +174,12 @@ blocking version utilizing PostgreSQL's NOTIFY and LISTEN, see 'lock' -} tryLockDB :: String -> DB (Maybe Payload)-tryLockDB schemaName = fmap listToMaybe $ query_ $ withSchema schemaName- [sql| UPDATE payloads+tryLockDB tableName = fmap listToMaybe $ query_ $+ [sql| UPDATE|] <> " " <> fromString tableName <> " " <> [sql| SET state='locked' WHERE id in ( SELECT id- FROM payloads+ FROM|] <> " " <> fromString tableName <> " " <> [sql| WHERE state='enqueued' ORDER BY created_at ASC LIMIT 1@@ -196,8 +193,8 @@ useful. The DB version is provided for completeness. -} unlockDB :: String -> PayloadId -> DB ()-unlockDB schemaName payloadId = void $ execute (withSchema schemaName- [sql| UPDATE payloads+unlockDB tableName payloadId = void $ execute (+ [sql| UPDATE|] <> " " <> fromString tableName <> " " <> [sql| SET state='enqueued' WHERE id=? AND state='locked' |])@@ -205,8 +202,8 @@ -- | Transition a 'Payload' to the 'Dequeued' state. dequeueDB :: String -> PayloadId -> DB ()-dequeueDB schemaName payloadId = void $ execute (withSchema schemaName- [sql| UPDATE payloads+dequeueDB tableName payloadId = void $ execute (+ [sql| UPDATE|] <> " " <> fromString tableName <> " " <> [sql| SET state='dequeued' WHERE id=? |])@@ -214,9 +211,9 @@ -- | Get the number of rows in the 'Enqueued' state. getCountDB :: String -> DB Int64-getCountDB schemaName = fmap (fromOnly . head) $ query_ $ withSchema schemaName+getCountDB tableName = fmap (fromOnly . head) $ query_ $ [sql| SELECT count(*)- FROM payloads+ FROM|] <> " " <> fromString tableName <> " " <> [sql| WHERE state='enqueued' |] -------------------------------------------------------------------------------@@ -226,7 +223,7 @@ which can be composed with other queries in a single transaction. -} enqueue :: String -> Connection -> Value -> IO PayloadId-enqueue schemaName conn value = runDBT (enqueueDB schemaName value) ReadCommitted conn+enqueue tableName conn value = runDBT (enqueueDB tableName 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@@ -234,12 +231,12 @@ 'Serializable' transaction. -} tryLock :: String -> Connection -> IO (Maybe Payload)-tryLock schemaName conn = runDBTSerializable (tryLockDB schemaName) conn+tryLock tableName conn = runDBTSerializable (tryLockDB tableName) conn notifyPayload :: String -> Connection -> IO ()-notifyPayload schemaName conn = do+notifyPayload tableName conn = do Notification {..} <- getNotification conn- unless (notificationChannel == notifyName schemaName) $ notifyPayload schemaName conn+ unless (notificationChannel == notifyName tableName) $ notifyPayload tableName conn {-| Return the oldest 'Payload' in the 'Enqueued' state or block until a payload arrives. This function utilizes PostgreSQL's LISTEN and NOTIFY@@ -247,14 +244,14 @@ waiting for new payloads, without scarficing promptness. -} lock :: String -> Connection -> IO Payload-lock schemaName conn = bracket_- (Simple.execute_ conn $ "LISTEN " <> notifyName schemaName)- (Simple.execute_ conn $ "UNLISTEN " <> notifyName schemaName)+lock tableName conn = bracket_+ (Simple.execute_ conn $ "LISTEN " <> notifyName tableName)+ (Simple.execute_ conn $ "UNLISTEN " <> notifyName tableName) $ fix $ \continue -> do- m <- tryLock schemaName conn+ m <- tryLock tableName conn case m of Nothing -> do- notifyPayload schemaName conn+ notifyPayload tableName conn continue Just x -> return x @@ -263,15 +260,15 @@ shutdown. For a DB API version see 'unlockDB' -} unlock :: String -> Connection -> PayloadId -> IO ()-unlock schemaName conn x = runDBTSerializable (unlockDB schemaName x) conn+unlock tableName conn x = runDBTSerializable (unlockDB tableName x) conn -- | Transition a 'Payload' to the 'Dequeued' state. his functions runs -- 'dequeueDB' as a 'Serializable' transaction. dequeue :: String -> Connection -> PayloadId -> IO ()-dequeue schemaName conn x = runDBTSerializable (dequeueDB schemaName x) conn+dequeue tableName conn x = runDBTSerializable (dequeueDB tableName x) conn {-| Get the number of rows in the 'Enqueued' state. This function runs 'getCountDB' in a 'ReadCommitted' transaction. -} getCount :: String -> Connection -> IO Int64-getCount schemaName = runDBT (getCountDB schemaName) ReadCommitted+getCount tableName = runDBT (getCountDB tableName) ReadCommitted
src/Database/PostgreSQL/Simple/Queue/Main.hs view
@@ -73,7 +73,7 @@ data PartialOptions = PartialOptions { threadCount :: Last Int , dbOptions :: PostgreSQL.PartialOptions- , schemaName :: Last String+ , tableName :: Last String } deriving (Show, Eq, Typeable) instance Monoid PartialOptions where@@ -82,7 +82,7 @@ PartialOptions (threadCount x <> threadCount y) (dbOptions x <> dbOptions y)- (schemaName x <> schemaName y)+ (tableName x <> tableName y) -- | The default 'threadCount' is 1. -- The default db options are specified in@@ -107,7 +107,7 @@ -- | Convert a 'PartialOptions' to a final 'Options' completeOptions :: PartialOptions -> Either [String] Options completeOptions = \case- PartialOptions { threadCount = Last (Just oThreadCount), dbOptions, schemaName = Last (Just oSchemaName) } ->+ PartialOptions { threadCount = Last (Just oThreadCount), dbOptions, tableName = Last (Just oSchemaName) } -> Options oThreadCount <$> PostgreSQL.completeOptions dbOptions <*> pure oSchemaName _ -> Left ["Missing threadCount"]
src/Database/PostgreSQL/Simple/Queue/Migrate.hs view
@@ -12,7 +12,7 @@ @ CREATE TYPE state_t AS ENUM ('enqueued', 'locked', 'dequeued'); - CREATE TABLE payloads+ CREATE TABLE |] <> " " <> fromString tableName <> [sql| ( id uuid PRIMARY KEY , value jsonb NOT NULL , state state_t NOT NULL DEFAULT 'enqueued'@@ -20,7 +20,7 @@ , modified_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT clock_timestamp() ); - CREATE INDEX state_idx ON payloads (state);+ CREATE INDEX state_idx ON|] <> " " <> fromString tableName <> " " <> [sql|(state); CREATE OR REPLACE FUNCTION update_row_modified_function_() RETURNS TRIGGER@@ -38,12 +38,10 @@ @ -} migrate :: String -> Connection -> IO ()-migrate schemaName conn = void $ execute_ conn $- "CREATE SCHEMA IF NOT EXISTS " <> fromString schemaName <> ";" <>- "SET search_path TO " <> fromString schemaName <> ";" <> [sql|+migrate tableName conn = void $ execute_ conn $ [sql| CREATE TYPE state_t AS ENUM ('enqueued', 'locked', 'dequeued'); - CREATE TABLE payloads+ CREATE TABLE |] <> " " <> fromString tableName <> [sql| ( id SERIAL PRIMARY KEY , value jsonb NOT NULL , state state_t NOT NULL DEFAULT 'enqueued'@@ -51,7 +49,7 @@ , modified_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT clock_timestamp() ); - CREATE INDEX state_idx ON payloads (state);+ CREATE INDEX state_idx ON |] <> " " <> fromString tableName <> [sql|(state); CREATE OR REPLACE FUNCTION update_row_modified_function_() RETURNS TRIGGER
test/Database/PostgreSQL/Simple/QueueSpec.hs view
@@ -17,37 +17,37 @@ main :: IO () main = hspec spec -schemaName :: String-schemaName = "queue"+tableName :: String+tableName = "queue" spec :: Spec-spec = describeDB (migrate schemaName) "Database.Queue" $ do+spec = describeDB (migrate tableName) "Database.Queue" $ do itDB "empty locks nothing" $- tryLockDB schemaName `shouldReturn` Nothing+ tryLockDB tableName `shouldReturn` Nothing itDB "empty gives count 0" $- getCountDB schemaName `shouldReturn` 0+ getCountDB tableName `shouldReturn` 0 itDB "enqueues/tryLocks/unlocks" $ do- payloadId <- enqueueDB schemaName $ String "Hello"- getCountDB schemaName `shouldReturn` 1+ payloadId <- enqueueDB tableName $ String "Hello"+ getCountDB tableName `shouldReturn` 1 - Just Payload {..} <- tryLockDB schemaName- getCountDB schemaName `shouldReturn` 0+ Just Payload {..} <- tryLockDB tableName+ getCountDB tableName `shouldReturn` 0 pId `shouldBe` payloadId pValue `shouldBe` String "Hello"- tryLockDB schemaName `shouldReturn` Nothing+ tryLockDB tableName `shouldReturn` Nothing - unlockDB schemaName pId- getCountDB schemaName `shouldReturn` 1+ unlockDB tableName pId+ getCountDB tableName `shouldReturn` 1 itDB "tryLocks/dequeues" $ do- Just Payload {..} <- tryLockDB schemaName- getCountDB schemaName `shouldReturn` 0+ Just Payload {..} <- tryLockDB tableName+ getCountDB tableName `shouldReturn` 0 - dequeueDB schemaName pId `shouldReturn` ()- tryLockDB schemaName `shouldReturn` Nothing+ dequeueDB tableName pId `shouldReturn` ()+ tryLockDB tableName `shouldReturn` Nothing it "enqueues and dequeues concurrently tryLock" $ \testDB -> do let withPool' = withPool testDB@@ -57,18 +57,18 @@ ref <- newIORef [] loopThreads <- replicateM 10 $ async $ fix $ \next -> do- mpayload <- withPool' $ tryLock schemaName+ mpayload <- withPool' $ tryLock tableName case mpayload of Nothing -> next Just Payload {..} -> do lastCount <- atomicModifyIORef ref $ \xs -> (pValue : xs, length xs + 1)- withPool' $ \c -> dequeue schemaName c pId+ withPool' $ \c -> dequeue tableName c pId when (lastCount < elementCount) next -- Fork a hundred threads and enqueue an index forM_ [0 .. elementCount - 1] $ \i ->- forkIO $ void $ withPool' $ \c -> enqueue schemaName c $ toJSON i+ forkIO $ void $ withPool' $ \c -> enqueue tableName c $ toJSON i waitAnyCancel loopThreads Just decoded <- mapM (decode . encode) <$> readIORef ref@@ -82,15 +82,15 @@ ref <- newIORef [] loopThreads <- replicateM 10 $ async $ fix $ \next -> do- Payload {..} <- withPool' $ lock schemaName+ Payload {..} <- withPool' $ lock tableName lastCount <- atomicModifyIORef ref $ \xs -> (pValue : xs, length xs + 1)- withPool' $ \c -> dequeue schemaName c pId+ withPool' $ \c -> dequeue tableName c pId when (lastCount < elementCount) next -- Fork a hundred threads and enqueue an index forM_ [0 .. elementCount - 1] $ \i -> forkIO $ void $ withPool' $ \c ->- enqueue schemaName c $ toJSON i+ enqueue tableName c $ toJSON i waitAnyCancel loopThreads Just decoded <- mapM (decode . encode) <$> readIORef ref