hasql-queue (empty) → 1.0.1
raw patch · 13 files changed
+1397/−0 lines, 13 filesdep +aesondep +asyncdep +basesetup-changed
Dependencies added: aeson, async, base, base64-bytestring, bytestring, cryptohash-sha1, exceptions, hasql, hasql-notifications, hasql-queue, here, hspec, hspec-core, hspec-expectations-lifted, monad-control, random, resource-pool, split, stm, text, time, tmp-postgres, transformers
Files
- CHANGELOG.md +3/−0
- LICENSE +30/−0
- README.md +33/−0
- Setup.hs +2/−0
- benchmarks/Main.hs +102/−0
- hasql-queue.cabal +149/−0
- src/Hasql/Queue/IO.hs +128/−0
- src/Hasql/Queue/Internal.hs +248/−0
- src/Hasql/Queue/Migrate.hs +81/−0
- src/Hasql/Queue/Session.hs +180/−0
- test/Database/Hasql/Queue/IOSpec.hs +206/−0
- test/Database/Hasql/Queue/SessionSpec.hs +226/−0
- test/Main.hs +9/−0
+ CHANGELOG.md view
@@ -0,0 +1,3 @@+Changelog for hasql-queue+- 1.0.0.0+ - First release!
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Author Jonathan Fischoff (c) 2020++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,33 @@+[](http://travis-ci.org/jfischoff/hasql-queue)++# hasql-queue++This module utilizes PostgreSQL to implement a durable queue for efficently processing payloads.++Typically a producer would enqueue a new payload as part of larger database transaction++```haskell+createAccount userRecord = transaction Serializable Write $ do+ createUser userRecord+ enqueueNotification "queue_channel" emailEncoder [makeVerificationEmail userRecord]+```++In another thread or process the consumer would drain the queue.++```haskell+ -- Wait for a single new record and try to send the email 5 times for giving+ -- up and marking the payload as failed.+ forever $ withDequeue "queue_channel" conn emailDecoder 5 1 $+ mapM_ sendEmail+```++In the example above we used the `Session` API for enqueuing and the `IO` for+dequeuing.++The `Session` API is useful for composing larger transactions and the `IO` utilizes PostgreSQL notifications to avoid polling.++## Installation++```bash+stack install hasql-queue+```
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ benchmarks/Main.hs view
@@ -0,0 +1,102 @@+module Main where+import System.Environment+import Hasql.Queue.Migrate+import Data.IORef+import Control.Exception+import Crypto.Hash.SHA1 (hash)+import qualified Data.ByteString.Base64.URL as Base64+import qualified Data.ByteString.Char8 as BSC+import Data.Pool+import Database.Postgres.Temp+import Control.Concurrent+import Control.Monad (replicateM, forever, void)+import Hasql.Session+import Hasql.Connection+import Data.Function+import qualified Hasql.Encoders as E+import qualified Hasql.Decoders as D+import Hasql.Statement+import qualified Hasql.Queue.Internal as I+import qualified Hasql.Queue.Session as S+import Data.Int++-- TODO need to make sure the number of producers and consumers does not go over the number of connections++withConn :: DB -> (Connection -> IO a) -> IO a+withConn db f = do+ let connStr = toConnectionString db+ bracket (either (throwIO . userError . show) pure =<< acquire connStr) release f++withSetup :: (Pool Connection -> IO ()) -> IO ()+withSetup f = do+ -- Helper to throw exceptions+ let throwE x = either throwIO pure =<< x++ throwE $ withDbCache $ \dbCache -> do+ --let combinedConfig = autoExplainConfig 15 <> cacheConfig dbCache+ let combinedConfig = defaultConfig <> cacheConfig dbCache+ migratedConfig <- throwE $ cacheAction (("~/.tmp-postgres/" <>) . BSC.unpack . Base64.encode . hash+ $ BSC.pack $ migrationQueryString "int4")+ (flip withConn $ flip migrate "int4")+ combinedConfig+ withConfig migratedConfig $ \db -> do+ print $ toConnectionString db++ f =<< createPool+ (either (throwIO . userError . show) pure =<< acquire (toConnectionString db)+ ) release 2 60 49++payload :: Int32+payload = 1++main :: IO ()+main = do+ [producerCount, consumerCount, time, initialDequeueCount, initialEnqueueCount, batchCount] <- map read <$> getArgs+ -- create a temporary database+ enqueueCounter <- newIORef (0 :: Int)+ dequeueCounter <- newIORef (0 :: Int)++ let printCounters = do+ finalEnqueueCount <- readIORef enqueueCounter+ finalDequeueCount <- readIORef dequeueCounter+ putStrLn $ "Time " <> show time <> " secs"+ putStrLn $ "Enqueue Count: " <> show finalEnqueueCount+ putStrLn $ "Dequeue Count: " <> show finalDequeueCount++ flip finally printCounters $ withSetup $ \pool -> do+ -- enqueue the enqueueCount + dequeueCount+ let enqueueAction = void $ withResource pool $ \conn -> I.runThrow (S.enqueue E.int4 [payload]) conn+ dequeueAction = void $ withResource pool $ \conn -> fix $ \next ->+ I.runThrow (S.dequeue D.int4 batchCount) conn >>= \case+ [] -> next+ _ -> pure ()++ let enqueueInsertSql = "INSERT INTO payloads (attempts, value) SELECT 0, g.value FROM generate_series(1, $1) AS g (value)"+ enqueueInsertStatement =+ statement (fromIntegral initialEnqueueCount) $ Statement enqueueInsertSql (E.param $ E.nonNullable E.int4) D.noResult False++ withResource pool $ run enqueueInsertStatement++ let dequeueInsertSql = "INSERT INTO payloads (attempts, state, value) SELECT 0, 'dequeued', g.value FROM generate_series(1, $1) AS g (value)"+ dequeueInsertStatement =+ statement (fromIntegral initialDequeueCount) $ Statement dequeueInsertSql (E.param $ E.nonNullable E.int4) D.noResult False++ withResource pool $ run dequeueInsertStatement++ withResource pool $ \conn -> void $ run (sql "VACUUM FULL ANALYZE") conn+ putStrLn "Finished VACUUM FULL ANALYZE"++ let enqueueLoop = forever $ do+ enqueueAction+ atomicModifyIORef' enqueueCounter $ \x -> (x+1, ())++ dequeueLoop = forever $ do+ dequeueAction+ atomicModifyIORef' dequeueCounter $ \x -> (x+1, ())++ -- Need better exception behavior ... idk ... I'll deal with this later+ _enqueueThreads <- replicateM producerCount $ forkIO enqueueLoop+ _dequeueThreads <- replicateM consumerCount $ forkIO dequeueLoop++ threadDelay $ time * 1000000+ throwIO $ userError "Finished"
+ hasql-queue.cabal view
@@ -0,0 +1,149 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.31.2.+--+-- see: https://github.com/sol/hpack+--+-- hash: 2c8109092a14882df219b99da03aea13b68292c07f409a58fa68028d6f263eb1++name: hasql-queue+version: 1.0.1+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' "queue_schema" $ 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 "queue" 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 "queue" conn $ pId payload+ >+ > To support multiple queues in the same database, the API expects a table name string+ > to determine which queue tables to use.+category: Web+homepage: https://github.com/jfischoff/postgresql-queue#readme+bug-reports: https://github.com/jfischoff/postgresql-queue/issues+author: Jonathan Fischoff+maintainer: jonathangfischoff@gmail.com+copyright: 2020 Jonathan Fischoff+license: BSD3+license-file: LICENSE+tested-with: GHC ==8.8.1+build-type: Simple+extra-source-files:+ README.md+ CHANGELOG.md++source-repository head+ type: git+ location: https://github.com/jfischoff/postgresql-queue++library+ exposed-modules:+ Hasql.Queue.Internal+ Hasql.Queue.IO+ Hasql.Queue.Migrate+ Hasql.Queue.Session+ other-modules:+ Paths_hasql_queue+ hs-source-dirs:+ src+ default-extensions: OverloadedStrings LambdaCase RecordWildCards TupleSections GeneralizedNewtypeDeriving QuasiQuotes ScopedTypeVariables TypeApplications AllowAmbiguousTypes+ ghc-options: -Wall -Wno-unused-do-bind -Wno-unused-foralls+ build-depends:+ aeson+ , base >=4.7 && <5+ , bytestring+ , exceptions+ , hasql+ , hasql-notifications+ , here+ , monad-control+ , random+ , stm+ , text+ , time+ , transformers+ default-language: Haskell2010++executable benchmark+ main-is: Main.hs+ other-modules:+ Paths_hasql_queue+ hs-source-dirs:+ benchmarks+ default-extensions: OverloadedStrings LambdaCase RecordWildCards TupleSections GeneralizedNewtypeDeriving QuasiQuotes ScopedTypeVariables TypeApplications AllowAmbiguousTypes+ ghc-options: -Wall -Wno-unused-do-bind -Wno-unused-foralls -O2 -threaded -rtsopts -with-rtsopts=-N -g2+ build-depends:+ aeson+ , async+ , base >=4.7 && <5+ , base64-bytestring+ , bytestring+ , cryptohash-sha1+ , exceptions+ , hasql+ , hasql-notifications+ , hasql-queue+ , here+ , monad-control+ , random+ , resource-pool+ , stm+ , text+ , time+ , tmp-postgres+ , transformers+ default-language: Haskell2010++test-suite unit-tests+ type: exitcode-stdio-1.0+ main-is: Main.hs+ other-modules:+ Database.Hasql.Queue.IOSpec+ Database.Hasql.Queue.SessionSpec+ Paths_hasql_queue+ hs-source-dirs:+ test+ default-extensions: OverloadedStrings LambdaCase RecordWildCards TupleSections GeneralizedNewtypeDeriving QuasiQuotes ScopedTypeVariables TypeApplications AllowAmbiguousTypes+ ghc-options: -Wall -Wno-unused-do-bind -Wno-unused-foralls -O2 -threaded -rtsopts -with-rtsopts=-N+ build-depends:+ aeson+ , async+ , base >=4.7 && <5+ , base64-bytestring+ , bytestring+ , cryptohash-sha1+ , exceptions+ , hasql+ , hasql-notifications+ , hasql-queue+ , here+ , hspec+ , hspec-core+ , hspec-expectations-lifted+ , monad-control+ , random+ , resource-pool+ , split+ , stm+ , text+ , time+ , tmp-postgres+ , transformers+ default-language: Haskell2010
+ src/Hasql/Queue/IO.hs view
@@ -0,0 +1,128 @@+{-|+IO based API for a PostgreSQL backed queue. The API utilizes PostgreSQL+notifications.+-}+module Hasql.Queue.IO+ ( enqueue+ , withDequeue+ -- ** Listing API+ , I.PayloadId+ , failed+ , dequeued+ -- ** Advanced API+ , withDequeueWith+ , I.WithNotifyHandlers (..)+ ) where++import qualified Hasql.Queue.Session as S+import qualified Hasql.Queue.Internal as I+import Hasql.Connection+import qualified Hasql.Encoders as E+import qualified Hasql.Decoders as D+import Control.Exception+import Data.Function+import Data.ByteString (ByteString)++{-|Enqueue a payload.+-}+enqueue :: ByteString+ -- ^ Notification channel name. Any valid PostgreSQL identifier+ -> Connection+ -- ^ Connection+ -> E.Value a+ -- ^ Payload encoder+ -> [a]+ -- ^ List of payloads to enqueue+ -> IO ()+enqueue channel conn encoder xs = I.runThrow (S.enqueueNotify channel encoder xs) conn++{-|+Wait for the next payload and process it. If the continuation throws an+exception the payloads are put back in the queue. An 'IOError' is caught+and 'withDequeue' will retry up to the retry count. If 'withDequeue' fails+after too many retries the final exception is rethrown. If individual payloads are+are attempted more than the retry count they are set as "failed". See 'failed'+to receive the list of failed payloads.++If the queue is empty 'withDequeue' will block until it recieves a notification+from the PostgreSQL server.+-}+withDequeue :: ByteString+ -- ^ Notification channel name. Any valid PostgreSQL identifier+ -> Connection+ -- ^ Connection+ -> D.Value a+ -- ^ Payload decoder+ -> Int+ -- ^ Retry count+ -> Int+ -- ^ Element count+ -> ([a] -> IO b)+ -- ^ Continuation+ -> IO b+withDequeue = withDequeueWith @IOError mempty++{-|+Retrieve the payloads that have entered a failed state. See 'withDequeue' for how that+occurs. The function returns a list of values and an id. The id is used the starting+place for the next batch of values. If 'Nothing' is passed the list starts at the+beginning.+-}+failed :: Connection+ -> D.Value a+ -- ^ Payload decoder+ -> Maybe I.PayloadId+ -- ^ Starting position of payloads. Pass 'Nothing' to+ -- start at the beginning+ -> Int+ -- ^ Count+ -> IO (I.PayloadId, [a])+failed conn decoder mPayload count = I.runThrow (S.failed decoder mPayload count) conn++{-|+Retrieve the payloads that have been successfully dequeued.+The function returns a list of values and an id. The id is used the starting+place for the next batch of values. If 'Nothing' is passed the list starts at the+beginning.+-}+dequeued :: Connection+ -> D.Value a+ -- ^ Payload decoder+ -> Maybe I.PayloadId+ -- ^ Starting position of payloads. Pass 'Nothing' to+ -- start at the beginning+ -> Int+ -- ^ Count+ -> IO (I.PayloadId, [a])+dequeued conn decoder mPayload count = I.runThrow (S.dequeued decoder mPayload count) conn++{-|+A more general configurable version of 'withDequeue'. Unlike 'withDequeue' one+can specify the+-}+withDequeueWith :: forall e a b+ . Exception e+ => I.WithNotifyHandlers+ -- Event handlers for events that occur as 'withDequeWith' loops+ -> ByteString+ -- ^ Notification channel name. Any valid PostgreSQL identifier+ -> Connection+ -- ^ Connection+ -> D.Value a+ -- ^ Payload decoder+ -> Int+ -- ^ Retry count+ -> Int+ -- ^ Element count+ -> ([a] -> IO b)+ -- ^ Continuation+ -> IO b+withDequeueWith withNotifyHandlers channel conn decoder retryCount count f = (fix $ \restart i -> do+ try (I.withNotifyWith withNotifyHandlers channel conn (I.withDequeue decoder retryCount count f) id) >>= \case+ Right x -> pure x+ Left (e :: e) ->+ if i < retryCount then+ restart $ i + 1+ else+ throwIO e+ ) 0
+ src/Hasql/Queue/Internal.hs view
@@ -0,0 +1,248 @@+module Hasql.Queue.Internal where+import qualified Hasql.Encoders as E+import qualified Hasql.Decoders as D+import Hasql.Session+import Hasql.Notification+import Control.Monad (unless)+import Data.Function(fix)+import Hasql.Connection+import Data.Int+import Data.Functor.Contravariant+import Data.String.Here.Uninterpolated+import Hasql.Statement+import Data.ByteString (ByteString)+import Control.Exception+import Control.Monad.IO.Class+import Data.Typeable++-- | A 'Payload' can exist in three states in the queue, 'Enqueued',+-- 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 | Dequeued | Failed+ deriving (Show, Eq, Ord, Enum, Bounded)++state :: E.Params a -> D.Result b -> ByteString -> Statement a b+state enc dec theSql = Statement theSql enc dec True++stateDecoder :: D.Value State+stateDecoder = D.enum $ \txt ->+ if txt == "enqueued" then+ pure Enqueued+ else if txt == "dequeued" then+ pure Dequeued+ else if txt == "failed" then+ pure Failed+ else Nothing++stateEncoder :: E.Value State+stateEncoder = E.enum $ \case+ Enqueued -> "enqueued"+ Dequeued -> "dequeued"+ Failed -> "failed"++initialPayloadId :: PayloadId+initialPayloadId = PayloadId (-1)++{-|+Internal payload id. Used by the public api as continuation token+for pagination.+-}+newtype PayloadId = PayloadId { unPayloadId :: Int64 }+ deriving (Eq, Show)++-- | The fundemental record stored in the queue. The queue is a single table+-- and each row consists of a 'Payload'+data Payload a = Payload+ { pId :: PayloadId+ , pState :: State+ -- TODO do I need this?+ , pAttempts :: Int+ , pModifiedAt :: Int+ -- TODO rename. I don't need this either.+ , pValue :: a+ } deriving (Show, Eq)++-- | 'Payload' decoder+payloadDecoder :: D.Value a -> D.Row (Payload a)+payloadDecoder thePayloadDecoder+ = Payload+ <$> payloadIdRow+ <*> D.column (D.nonNullable stateDecoder)+ <*> D.column (D.nonNullable $ fromIntegral <$> D.int4)+ <*> D.column (D.nonNullable $ fromIntegral <$> D.int4)+ <*> D.column (D.nonNullable thePayloadDecoder)++payloadIdEncoder :: E.Value PayloadId+payloadIdEncoder = unPayloadId >$< E.int8++payloadIdDecoder :: D.Value PayloadId+payloadIdDecoder = PayloadId <$> D.int8++payloadIdRow :: D.Row PayloadId+payloadIdRow = D.column (D.nonNullable payloadIdDecoder)++-- TODO include special cases for single element insertion+enqueuePayload :: E.Value a -> [a] -> Session [PayloadId]+enqueuePayload theEncoder values = do+ let theQuery = [here|+ INSERT INTO payloads (attempts, value)+ SELECT 0, * FROM unnest($1)+ RETURNING id+ |]+ encoder = E.param $ E.nonNullable $ E.foldableArray $ E.nonNullable theEncoder+ decoder = D.rowList (D.column (D.nonNullable payloadIdDecoder))+ theStatement = Statement theQuery encoder decoder True++ statement values theStatement++dequeuePayload :: D.Value a -> Int -> Session [Payload a]+dequeuePayload valueDecoder count = do+ let multipleQuery = [here|+ UPDATE payloads+ SET state='dequeued'+ WHERE id in+ ( SELECT p1.id+ FROM payloads AS p1+ WHERE p1.state='enqueued'+ ORDER BY p1.modified_at ASC+ FOR UPDATE SKIP LOCKED+ LIMIT $1+ )+ RETURNING id, state, attempts, modified_at, value+ |]+ multipleEncoder = E.param $ E.nonNullable $ fromIntegral >$< E.int4++ singleQuery = [here|+ UPDATE payloads+ SET state='dequeued'+ WHERE id in+ ( SELECT p1.id+ FROM payloads AS p1+ WHERE p1.state='enqueued'+ ORDER BY p1.modified_at ASC+ FOR UPDATE SKIP LOCKED+ LIMIT 1+ )+ RETURNING id, state, attempts, modified_at, value+ |]++ singleEncoder = mempty++ decoder = D.rowList $ payloadDecoder valueDecoder++ theStatement = case count of+ 1 -> Statement singleQuery singleEncoder decoder True+ _ -> Statement multipleQuery multipleEncoder decoder True+ statement count theStatement++-- | Get the 'Payload' given a 'PayloadId'+getPayload :: D.Value a -> PayloadId -> Session (Maybe (Payload a))+getPayload decoder payloadId = do+ let theQuery = [here|+ SELECT id, state, attempts, modified_at, value+ FROM payloads+ WHERE id = $1+ |]++ encoder = E.param (E.nonNullable payloadIdEncoder)+ statement payloadId $ Statement theQuery encoder (D.rowMaybe $ payloadDecoder decoder) True+++-- | Get the number of rows in the 'Enqueued' state.+getCount :: Session Int64+getCount = do+ let decoder = D.singleRow (D.column (D.nonNullable D.int8))+ theSql = [here|+ SELECT count(*)+ FROM payloads+ WHERE state='enqueued';+ |]+ theStatement = Statement theSql mempty decoder True+ statement () theStatement++incrementAttempts :: Int -> [PayloadId] -> Session ()+incrementAttempts retryCount pids = do+ let theQuery = [here|+ UPDATE payloads+ SET state=CASE WHEN attempts >= $1 THEN 'failed' :: state_t ELSE 'enqueued' END+ , attempts=attempts+1+ WHERE id = ANY($2)+ |]+ encoder = (fst >$< E.param (E.nonNullable E.int4)) <>+ (snd >$< E.param (E.nonNullable $ E.foldableArray $ E.nonNullable payloadIdEncoder))++ theStatement = Statement theQuery encoder D.noResult True++ statement (fromIntegral retryCount, pids) theStatement++-- | Dequeue and+-- Move to Internal+-- This should use bracketOnError+withDequeue :: D.Value a -> Int -> Int -> ([a] -> IO b) -> Session (Maybe b)+withDequeue decoder retryCount count f = do+ sql "BEGIN;SAVEPOINT temp"+ dequeuePayload decoder count >>= \case+ [] -> Nothing <$ sql "COMMIT"+ xs -> fmap Just $ do+ liftIO (try $ f $ fmap pValue xs) >>= \case+ Left (e :: SomeException) -> do+ sql "ROLLBACK TO SAVEPOINT temp; RELEASE SAVEPOINT temp"+ let pids = fmap pId xs+ incrementAttempts retryCount pids+ sql "COMMIT"+ liftIO (throwIO e)+ Right x -> x <$ sql "COMMIT"++-- TODO remove+newtype QueryException = QueryException QueryError+ deriving (Eq, Show, Typeable)++instance Exception QueryException++runThrow :: Session a -> Connection -> IO a+runThrow sess conn = either (throwIO . QueryException) pure =<< run sess conn++execute :: Connection -> ByteString -> IO ()+execute conn theSql = runThrow (sql theSql) conn++-- Block until a payload notification is fired. Fired during insertion.+notifyPayload :: ByteString -> Connection -> IO ()+notifyPayload channel conn = fix $ \restart -> do+ Notification {..} <- either throwIO pure =<< getNotification conn+ unless (notificationChannel == channel) restart++-- | To aid in observability and white box testing+data WithNotifyHandlers = WithNotifyHandlers+ { withNotifyHandlersAfterAction :: IO ()+ -- ^ An event that is trigger after the initial action, e.g.+ -- before dequeue is called.+ , withNotifyHandlersBeforeNotification :: IO ()+ -- ^ An event that is triggered before the blocking on a+ -- notification.+ }++instance Semigroup WithNotifyHandlers where+ x <> y = WithNotifyHandlers+ { withNotifyHandlersAfterAction = withNotifyHandlersAfterAction x <> withNotifyHandlersAfterAction y+ , withNotifyHandlersBeforeNotification = withNotifyHandlersBeforeNotification x <> withNotifyHandlersBeforeNotification y+ }++instance Monoid WithNotifyHandlers where+ mempty = WithNotifyHandlers mempty mempty++withNotifyWith :: WithNotifyHandlers -> ByteString -> Connection -> Session a -> (a -> Maybe b) -> IO b+withNotifyWith WithNotifyHandlers {..} channel conn action theCast = bracket_+ (execute conn $ "LISTEN " <> channel)+ (execute conn $ "UNLISTEN " <> channel)+ $ fix $ \restart -> do+ x <- runThrow action conn+ withNotifyHandlersAfterAction+ case theCast x of+ Nothing -> do+ -- TODO record the time here+ withNotifyHandlersBeforeNotification+ notifyPayload channel conn+ restart+ Just xs -> pure xs
+ src/Hasql/Queue/Migrate.hs view
@@ -0,0 +1,81 @@+{-|+Functions for migrating the database to create the necessary+functions for the package.++Users can use these functions or copy and paste the tables+to create these tables through a standalone migration+system.+-}+{-# OPTIONS_HADDOCK prune #-}+{-# LANGUAGE QuasiQuotes, OverloadedStrings #-}+module Hasql.Queue.Migrate where+import Control.Monad+import Data.String+import Data.String.Here.Interpolated+import Hasql.Connection+import Hasql.Session+++{-|+The DDL statements to create the schema given a value type.+-}+migrationQueryString :: String+ -- ^ @value@ column type, e.g. @int4@ or+ -- @jsonb@.+ -> String+migrationQueryString valueType = [i|+ DO $$+ BEGIN+ IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'state_t') THEN+ CREATE TYPE state_t AS ENUM ('enqueued', 'dequeued', 'failed');+ END IF;+ END$$;++ CREATE SEQUENCE IF NOT EXISTS modified_index START 1;++ CREATE TABLE IF NOT EXISTS payloads+ ( id BIGSERIAL PRIMARY KEY+ , attempts int NOT NULL DEFAULT 0+ , state state_t NOT NULL DEFAULT 'enqueued'+ , modified_at int8 NOT NULL DEFAULT nextval('modified_index')+ , value ${valueType} NOT NULL+ ) WITH (fillfactor = 50);++ CREATE INDEX IF NOT EXISTS active_modified_at_idx ON payloads USING btree (modified_at)+ WHERE (state = 'enqueued');++|]++{-| This function creates a table and enumeration type that is+ appriopiate for the queue. The following sql is used.++ @+ DO $$+ BEGIN+ IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'state_t') THEN+ CREATE TYPE state_t AS ENUM ('enqueued', 'dequeued', 'failed');+ END IF;+ END$$;++ CREATE SEQUENCE IF NOT EXISTS modified_index START 1;++ CREATE TABLE IF NOT EXISTS payloads+ ( id BIGSERIAL PRIMARY KEY+ , attempts int NOT NULL DEFAULT 0+ , state state_t NOT NULL DEFAULT 'enqueued'+ , modified_at int8 NOT NULL DEFAULT nextval('modified_index')+ , value VALUE_TYPE NOT NULL+ ) WITH (fillfactor = 50);++ CREATE INDEX IF NOT EXISTS active_modified_at_idx ON payloads USING btree (modified_at)+ WHERE (state = 'enqueued');+ @++The @VALUE_TYPE@ needs to passed in through the second argument.+-}+migrate :: Connection+ -> String+ -- ^ The type of the @value@ column+ -> IO ()+migrate conn valueType = void $+ run (sql $ fromString $ migrationQueryString valueType) conn
+ src/Hasql/Queue/Session.hs view
@@ -0,0 +1,180 @@+{-|+'Session' based API for a PostgreSQL backed queue.+-}+module Hasql.Queue.Session+ ( enqueue+ , enqueueNotify+ , dequeue+ -- ** Listing API+ , PayloadId+ , failed+ , dequeued+ ) where+import qualified Hasql.Encoders as E+import qualified Hasql.Decoders as D+import Hasql.Session+import Data.Functor.Contravariant+import Data.String.Here.Uninterpolated+import Hasql.Statement+import Hasql.Queue.Internal+import Data.Maybe+import Data.ByteString (ByteString)++{-|Enqueue a payload.+-}+enqueue :: E.Value a+ -- ^ Payload encoder+ -> [a]+ -- ^ List of payloads to enqueue+ -> Session ()+enqueue theEncoder = \case+ [x] -> do+ let theQuery =+ [here|+ INSERT INTO payloads (attempts, value)+ VALUES (0, $1)+ |]++ encoder = E.param $ E.nonNullable theEncoder++ statement x $ Statement theQuery encoder D.noResult True++ xs -> do+ let theQuery =+ [here|+ INSERT INTO payloads (attempts, value)+ SELECT 0, * FROM unnest($1)+ |]++ encoder = E.param $ E.nonNullable $ E.foldableArray $ E.nonNullable theEncoder++ statement xs $ Statement theQuery encoder D.noResult True++{-|Enqueue a payload send a notification on the+specified channel.+-}+enqueueNotify :: ByteString+ -- ^ Notification channel name. Any valid PostgreSQL identifier+ -> E.Value a+ -- ^ Payload encoder+ -> [a]+ -- ^ List of payloads to enqueue+ -> Session ()+enqueueNotify channel theEncoder values = do+ enqueue theEncoder values+ sql $ "NOTIFY " <> channel++{-|+Dequeue a list of payloads+-}+dequeue :: D.Value a+ -- ^ Payload decoder+ -> Int+ -- ^ Element count+ -> Session [a]+dequeue valueDecoder count = do+ let multipleQuery = [here|+ UPDATE payloads+ SET state='dequeued'+ WHERE id in+ ( SELECT p1.id+ FROM payloads AS p1+ WHERE p1.state='enqueued'+ ORDER BY p1.modified_at ASC+ FOR UPDATE SKIP LOCKED+ LIMIT $1+ )+ RETURNING value+ |]+ multipleEncoder = E.param $ E.nonNullable $ fromIntegral >$< E.int4++ singleQuery = [here|+ UPDATE payloads+ SET state='dequeued'+ WHERE id in+ ( SELECT p1.id+ FROM payloads AS p1+ WHERE p1.state='enqueued'+ ORDER BY p1.modified_at ASC+ FOR UPDATE SKIP LOCKED+ LIMIT 1+ )+ RETURNING value+ |]++ singleEncoder = mempty++ decoder = D.rowList $ D.column $ D.nonNullable $ valueDecoder++ theStatement = case count of+ 1 -> Statement singleQuery singleEncoder decoder True+ _ -> Statement multipleQuery multipleEncoder decoder True+ statement count theStatement++fst3 :: (a, b, c) -> a+fst3 (x, _, _) = x++snd3 :: (a, b, c) -> b+snd3 (_, x, _) = x++trd3 :: (a, b, c) -> c+trd3 (_, _, x) = x++listState :: State -> D.Value a -> Maybe PayloadId -> Int -> Session (PayloadId, [a])+listState theState valueDecoder mPayloadId count = do+ let theQuery = [here|+ SELECT id, value+ FROM payloads+ WHERE state = ($1 :: state_t)+ AND id > $2+ ORDER BY id ASC+ LIMIT $3+ |]+ encoder = (fst3 >$< E.param (E.nonNullable stateEncoder))+ <> (snd3 >$< E.param (E.nonNullable payloadIdEncoder))+ <> (trd3 >$< E.param (E.nonNullable E.int4))++ decoder = D.rowList+ $ (,)+ <$> D.column (D.nonNullable payloadIdDecoder)+ <*> D.column (D.nonNullable valueDecoder)+ theStatement = Statement theQuery encoder decoder True++ defaultPayloadId = fromMaybe initialPayloadId mPayloadId++ idsAndValues <- statement (theState, defaultPayloadId, fromIntegral count) theStatement+ pure $ case idsAndValues of+ [] -> (defaultPayloadId, [])+ xs -> (fst $ last xs, map snd xs)++{-|+Retrieve the payloads that have entered a failed state. See 'withDequeue' for how that+occurs. The function returns a list of values and an id. The id is used the starting+place for the next batch of values. If 'Nothing' is passed the list starts at the+beginning.+-}+failed :: D.Value a+ -- ^ Payload decoder+ -> Maybe PayloadId+ -- ^ Starting position of payloads. Pass 'Nothing' to+ -- start at the beginning+ -> Int+ -- ^ Count+ -> Session (PayloadId, [a])+failed = listState Failed++{-|+Retrieve the payloads that have been successfully dequeued.+The function returns a list of values and an id. The id is used the starting+place for the next batch of values. If 'Nothing' is passed the list starts at the+beginning.+-}+dequeued :: D.Value a+ -- ^ Payload decoder+ -> Maybe PayloadId+ -- ^ Starting position of payloads. Pass 'Nothing' to+ -- start at the beginning+ -> Int+ -- ^ Count+ -> Session (PayloadId, [a])+dequeued = listState Dequeued
+ test/Database/Hasql/Queue/IOSpec.hs view
@@ -0,0 +1,206 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+module Database.Hasql.Queue.IOSpec where+import Control.Concurrent+import Control.Concurrent.STM+import Control.Concurrent.Async+import Control.Exception as E+import Control.Monad+import Data.Aeson+import Data.Function+import Data.IORef+import Data.List+import Hasql.Queue.IO+import Hasql.Queue.Migrate+import Test.Hspec (SpecWith, Spec, describe, it, afterAll, beforeAll, runIO)+import Test.Hspec.Expectations.Lifted+import Data.List.Split+import Database.Postgres.Temp as Temp+import Data.Pool+import Data.Foldable+import Crypto.Hash.SHA1 (hash)+import qualified Data.ByteString.Base64.URL as Base64+import qualified Data.ByteString.Char8 as BSC+import Data.ByteString (ByteString)+import Hasql.Connection+import Hasql.Session+import qualified Hasql.Encoders as E+import qualified Hasql.Decoders as D+import Data.Int+import Data.Typeable+import qualified Hasql.Queue.Internal as I+import Hasql.Queue.Internal (Payload (..))++aroundAll :: forall a. ((a -> IO ()) -> IO ()) -> SpecWith a -> Spec+aroundAll withFunc specWith = do+ (var, stopper, asyncer) <- runIO $+ (,,) <$> newEmptyMVar <*> newEmptyMVar <*> newIORef Nothing+ let theStart :: IO a+ theStart = do++ thread <- async $ do+ withFunc $ \x -> do+ putMVar var x+ takeMVar stopper+ pure $ error "Don't evaluate this"++ writeIORef asyncer $ Just thread++ either pure pure =<< (wait thread `race` takeMVar var)++ theStop :: a -> IO ()+ theStop _ = do+ putMVar stopper ()+ traverse_ cancel =<< readIORef asyncer++ beforeAll theStart $ afterAll theStop $ specWith++withConn :: Temp.DB -> (Connection -> IO a) -> IO a+withConn db f = do+ let connStr = toConnectionString db+ E.bracket (either (throwIO . userError . show) pure =<< acquire connStr) release f++runThrow :: Session a -> Connection -> IO a+runThrow sess conn = either (throwIO . I.QueryException) pure =<< run sess conn++getCount :: Connection -> IO Int64+getCount = runThrow I.getCount++getPayload :: Connection -> D.Value a -> I.PayloadId -> IO (Maybe (I.Payload a))+getPayload conn decoder payloadId = runThrow (I.getPayload decoder payloadId) conn++withSetup :: (Pool Connection -> IO ()) -> IO ()+withSetup f = either throwIO pure <=< withDbCache $ \dbCache -> do+ migratedConfig <- either throwIO pure =<<+ cacheAction+ (("~/.tmp-postgres/" <>) . BSC.unpack . Base64.encode . hash+ $ BSC.pack $ migrationQueryString "int4")+ (flip withConn $ flip migrate "int4")+ (verboseConfig <> cacheConfig dbCache)+ withConfig migratedConfig $ \db -> do+ f =<< createPool+ (either (throwIO . userError . show) pure =<< acquire (toConnectionString db))+ release+ 2+ 60+ 50++channel :: ByteString+channel = "hey"++withConnection :: (Connection -> IO ()) -> Pool Connection -> IO ()+withConnection = flip withResource++runReadCommitted :: Pool Connection -> Session a -> IO a+runReadCommitted = flip withReadCommitted++withReadCommitted :: Session a -> Pool Connection -> IO a+withReadCommitted action pool = do+ let wrappedAction = do+ sql "BEGIN"+ r <- action+ sql "ROLLBACK"+ pure r+ withResource pool $ \conn ->+ either (throwIO . userError . show) pure =<< run wrappedAction conn++{-+runNoTransaction :: Pool Connection -> Session a -> IO a+runNoTransaction pool session = withResource pool $ \conn ->+ either (throwIO . userError . show) pure =<< run action conn+-}++data FailedwithDequeue = FailedwithDequeue+ deriving (Show, Eq, Typeable)++instance Exception FailedwithDequeue++spec :: Spec+spec = describe "Hasql.Queue.IO" $ do+ aroundAll withSetup $ describe "basic" $ do+ it "withDequeue blocks until something is enqueued: before" $ withConnection $ \conn -> do+ void $ enqueue channel conn E.int4 [1]+ res <- withDequeueWith @IOException mempty channel conn D.int4 1 1 pure+ res `shouldBe` [1]+ getCount conn `shouldReturn` 0++ it "withDequeue blocks until something is enqueued: during" $ withConnection $ \conn -> do+ afterActionMVar <- newEmptyMVar+ beforeNotifyMVar <- newEmptyMVar++ let handlers = WithNotifyHandlers+ { withNotifyHandlersAfterAction = putMVar afterActionMVar ()+ , withNotifyHandlersBeforeNotification = takeMVar beforeNotifyMVar+ }++ -- This is the definition of IO.dequeue+ resultThread <- async $ withDequeueWith @IOException handlers channel conn D.int4 1 1 pure+ takeMVar afterActionMVar++ void $ enqueue "hey" conn E.int4 [1]++ putMVar beforeNotifyMVar ()++ wait resultThread `shouldReturn` [1]++ it "withDequeue blocks until something is enqueued: after" $ withConnection $ \conn -> do+ resultThread <- async $ withDequeueWith @IOException mempty channel conn D.int4 1 1 pure+ void $ enqueue channel conn E.int4 [1]++ wait resultThread `shouldReturn` [1]++ it "withDequeue fails and sets the retries to +1" $ withConnection $ \conn -> do+ [payloadId] <- runThrow (I.enqueuePayload E.int4 [1]) conn+ handle (\FailedwithDequeue -> pure ()) $ withDequeue channel conn D.int4 0 1 $ \_ -> throwIO FailedwithDequeue+ Just Payload {..} <- getPayload conn D.int4 payloadId++ pState `shouldBe` I.Failed+ pAttempts `shouldBe` 1++ it "withDequeue succeeds even if the first attempt fails" $ withConnection $ \conn -> do+ [payloadId] <- runThrow (I.enqueuePayload E.int4 [1]) conn++ ref <- newIORef (0 :: Int)++ withDequeueWith @FailedwithDequeue mempty channel conn D.int4 1 1 (\_ -> do+ count <- readIORef ref+ writeIORef ref $ count + 1+ when (count < 1) $ throwIO FailedwithDequeue+ pure '!') `shouldReturn` '!'++ Just Payload {..} <- getPayload conn D.int4 payloadId++ pState `shouldBe` I.Dequeued+ -- Failed attempts I guess+ pAttempts `shouldBe` 1++ it "enqueues and dequeues concurrently withDequeue" $ \testDB -> do+ let withPool' = flip withConnection testDB+ elementCount = 1000 :: Int+ expected = [0 .. elementCount - 1]++ ref <- newTVarIO []++ loopThreads <- replicateM 35 $ async $ withPool' $ \c -> fix $ \next -> do+ lastCount <- withDequeue channel c D.int4 1 1 $ \[x] -> do+ atomically $ do+ xs <- readTVar ref+ writeTVar ref $ x : xs+ return $ length xs + 1++ when (lastCount < elementCount) next++ forM_ (chunksOf (elementCount `div` 11) expected) $ \xs -> forkIO $ void $ withPool' $ \c ->+ forM_ xs $ \i -> enqueue channel c E.int4 [fromIntegral i]++ _ <- waitAnyCancel loopThreads+ xs <- atomically $ readTVar ref+ let Just decoded = mapM (decode . encode) xs+ sort decoded `shouldBe` sort expected++ it "enqueue returns a PayloadId that cooresponds to the entry it added" $ withConnection $ \conn -> do+ [payloadId] <- runThrow (I.enqueuePayload E.int4 [1]) conn+ Just actual <- getPayload conn D.int4 payloadId++ pValue actual `shouldBe` 1
+ test/Database/Hasql/Queue/SessionSpec.hs view
@@ -0,0 +1,226 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+module Database.Hasql.Queue.SessionSpec where+import Hasql.Queue.Internal+import Control.Concurrent+import Control.Concurrent.Async+import Control.Exception as E+import Control.Monad+import Data.IORef+import Hasql.Queue.Session+import Hasql.Queue.Migrate+import Test.Hspec (SpecWith, Spec, describe, parallel, it, afterAll, beforeAll, runIO)+import Test.Hspec.Expectations.Lifted+import Control.Monad.Catch+import Control.Monad.IO.Class+import Database.Postgres.Temp as Temp+import Data.Pool+import Data.Foldable+import Test.Hspec.Core.Spec (sequential)+import Crypto.Hash.SHA1 (hash)+import qualified Data.ByteString.Base64.URL as Base64+import qualified Data.ByteString.Char8 as BSC+import Hasql.Connection+import Hasql.Session+import qualified Hasql.Encoders as E+import qualified Hasql.Decoders as D+import Data.Typeable+import Data.Int++aroundAll :: forall a. ((a -> IO ()) -> IO ()) -> SpecWith a -> Spec+aroundAll withFunc specWith = do+ (var, stopper, asyncer) <- runIO $+ (,,) <$> newEmptyMVar <*> newEmptyMVar <*> newIORef Nothing+ let theStart :: IO a+ theStart = do++ thread <- async $ do+ withFunc $ \x -> do+ putMVar var x+ takeMVar stopper+ pure $ error "Don't evaluate this"++ writeIORef asyncer $ Just thread++ either pure pure =<< (wait thread `race` takeMVar var)++ theStop :: a -> IO ()+ theStop _ = do+ putMVar stopper ()+ traverse_ cancel =<< readIORef asyncer++ beforeAll theStart $ afterAll theStop $ specWith++withConn :: Temp.DB -> (Connection -> IO a) -> IO a+withConn db f = do+ let connStr = toConnectionString db+ E.bracket (either (throwIO . userError . show) pure =<< acquire connStr) release f++withSetup :: (Pool Connection -> IO ()) -> IO ()+withSetup f = either throwIO pure <=< withDbCache $ \dbCache -> do+ migratedConfig <- either throwIO pure =<<+ cacheAction+ (("~/.tmp-postgres/" <>) . BSC.unpack . Base64.encode . hash+ $ BSC.pack $ migrationQueryString "int4")+ (flip withConn $ flip migrate "int4")+ (verboseConfig <> cacheConfig dbCache)+ withConfig migratedConfig $ \db -> do+ f =<< createPool+ (either (throwIO . userError . show) pure =<< acquire (toConnectionString db))+ release+ 2+ 60+ 50++withConnection :: (Connection -> IO ()) -> Pool Connection -> IO ()+withConnection = flip withResource++runImplicitTransaction :: Pool Connection -> Session a -> IO a+runImplicitTransaction pool action = do+ let wrappedAction = do+ r <- action+ pure r+ withResource pool $ \conn ->+ either (throwIO . userError . show) pure =<< run wrappedAction conn++runReadCommitted :: Pool Connection -> Session a -> IO a+runReadCommitted = flip withReadCommitted++withReadCommitted :: Session a -> Pool Connection -> IO a+withReadCommitted action pool = do+ let wrappedAction = do+ sql "BEGIN"+ r <- action+ sql "ROLLBACK"+ pure r+ withResource pool $ \conn ->+ either (throwIO . userError . show) pure =<< run wrappedAction conn++newtype TooManyRetries = TooManyRetries Int64+ deriving (Show, Eq, Typeable)++instance Exception TooManyRetries++spec :: Spec+spec = describe "Hasql.Queue.Session" $ parallel $ do+ sequential $ aroundAll withSetup $ describe "basic" $ do+ it "is okay to migrate multiple times" $ withConnection $ \conn ->+ liftIO $ migrate conn "int4"++ it "empty locks nothing" $ \pool -> do+ runReadCommitted pool (withDequeue D.int4 8 1 return) >>= \x ->+ x `shouldBe` Nothing+ it "empty gives count 0" $ \pool ->+ runReadCommitted pool getCount `shouldReturn` 0++ it "dequeued paging works" $ \pool -> do+ (a, b) <- runReadCommitted pool $ do+ enqueue E.int4 [1,2,3,4]++ void $ dequeue D.int4 4++ (next, xs) <- dequeued D.int4 Nothing 2++ (_, ys) <- dequeued D.int4 (Just next) 2++ pure (xs, ys)++ a `shouldBe` [1,2]+ b `shouldBe` [3,4]++ it "failed paging works" $ \pool -> do+ runImplicitTransaction pool $ enqueue E.int4 [1,2,3,4]++ replicateM_ 8 $ E.handle (\(_ :: TooManyRetries) -> pure ()) $+ runImplicitTransaction pool $ do+ void $ withDequeue D.int4 1 1 $ const $+ throwM $ TooManyRetries 1++ (a, b) <- runImplicitTransaction pool $ do+ (next, xs) <- failed D.int4 Nothing 2+ (_, ys) <- failed D.int4 (Just next) 2++ pure (xs, ys)++ a `shouldBe` [1,2]+ b `shouldBe` [3,4]++ it "enqueue/withDequeue" $ \pool -> do+ (withDequeueResult, firstCount, secondCount) <- runReadCommitted pool $ do+ enqueueNotify "hey" E.int4 [1]+ firstCount <- getCount+ withDequeueResult <- withDequeue D.int4 8 1 (`shouldBe` [1])++ secondCount <- getCount+ pure (withDequeueResult, firstCount, secondCount)++ firstCount `shouldBe` 1+ secondCount `shouldBe` 0+ withDequeueResult `shouldBe` Just ()++ it "enqueue/withDequeue/retries" $ \pool -> do+ runImplicitTransaction pool $ enqueue E.int4 [1]++ e <- E.try $ runImplicitTransaction pool $ do+ theCount <- getCount++ void $ withDequeue D.int4 8 1 $ const $+ throwM $ TooManyRetries theCount++ (e :: Either TooManyRetries ()) `shouldBe` Left (TooManyRetries 1)++ runImplicitTransaction pool (dequeuePayload D.int4 1) >>= \[(Payload {..})] -> do+ pAttempts `shouldBe` 1+ pValue `shouldBe` 1++ runImplicitTransaction pool $ enqueue E.int4 [1]++ e1 <- E.try $ runImplicitTransaction pool $ do+ theCount <- getCount++ void $ withDequeue D.int4 8 1 $ const $+ throwM $ TooManyRetries theCount++ (e1 :: Either TooManyRetries ()) `shouldBe` Left (TooManyRetries 1)++ replicateM_ 6 $ E.handle (\(_ :: TooManyRetries) -> pure ()) $ runImplicitTransaction pool $ do+ void $ withDequeue D.int4 8 1 $ const $+ throwM $ TooManyRetries 1++ runImplicitTransaction pool (dequeuePayload D.int4 1) >>= \[(Payload {..})] -> do+ pAttempts `shouldBe` 7+ pValue `shouldBe` 1++ it "enqueue/withDequeue/timesout" $ \pool -> do+ e <- E.try $ runReadCommitted pool $ do+ enqueue E.int4 [1]+ firstCount <- getCount++ void $ withDequeue D.int4 0 1 $ const $+ throwM $ TooManyRetries firstCount++ (e :: Either TooManyRetries ())`shouldBe` Left (TooManyRetries 1)++ runReadCommitted pool getCount `shouldReturn` 0++ it "selects the oldest first" $ \pool -> do+ (firstCount, firstwithDequeueResult, secondwithDequeueResult, secondCount) <- runReadCommitted pool $ do+ enqueue E.int4 [1]+ liftIO $ threadDelay 100++ enqueue E.int4 [2]++ firstCount <- getCount++ firstwithDequeueResult <- withDequeue D.int4 8 1 (`shouldBe` [1])+ secondwithDequeueResult <- withDequeue D.int4 8 1 (`shouldBe` [2])++ secondCount <- getCount+ pure (firstCount, firstwithDequeueResult, secondwithDequeueResult, secondCount)++ firstCount `shouldBe` 2+ firstwithDequeueResult `shouldBe` Just ()++ secondCount `shouldBe` 0+ secondwithDequeueResult `shouldBe` Just ()
+ test/Main.hs view
@@ -0,0 +1,9 @@+import Test.Hspec+import Database.Hasql.Queue.SessionSpec as S+import Database.Hasql.Queue.IOSpec as I+++main :: IO ()+main = hspec $ do+ S.spec+ I.spec