packages feed

hasql-queue 1.0.1.1 → 1.2.0.0

raw patch · 25 files changed

+1353/−812 lines, 25 filesnew-component:exe:hasql-queue-tmp-db

Files

CHANGELOG.md view
@@ -1,4 +1,19 @@ Changelog for hasql-queue+-1.2.0.0+  - Escape notification channel #100+  - Remove dequeued count from benchmarks #99+  - Document new api #98+  - Create teardown for migration #96+  - Extend API to use schemas #95+  - Add `delete` to the `AtLeastOnce` APIs #92+  - Change `failed` function to `failures` and return all `PayloadIds` #91+  - Reorg API #89+  - Use a compound index to make the partial index work #86+  - Specialize single element dequeue #84+  - Delete instead of changing the state to dequeued #82+  -++ - 1.0.1.1   - Fixed cabal meta data and copyright
benchmarks/Main.hs view
@@ -17,7 +17,8 @@ import qualified Hasql.Decoders as D import           Hasql.Statement import qualified Hasql.Queue.Internal as I-import qualified Hasql.Queue.Session as S+import qualified Hasql.Queue.Low.AtLeastOnce as IO+import qualified Hasql.Queue.High.ExactlyOnce as S import           Data.Int  -- TODO need to make sure the number of producers and consumers does not go over the number of connections@@ -27,14 +28,26 @@   let connStr = toConnectionString db   bracket (either (throwIO . userError . show) pure =<< acquire connStr) release f -withSetup :: (Pool Connection -> IO ()) -> IO ()-withSetup f = do+durableConfig :: Int -> Config+durableConfig microseconds = defaultConfig <> mempty+  { postgresConfigFile =+      [ ("wal_level", "replica")+      , ("archive_mode", "on")+      , ("max_wal_senders", "2")+      , ("fsync", "on")+      , ("synchronous_commit", "on")+      , ("commit_delay", show microseconds)+      ]+  }++withSetup :: Int -> Bool -> (Pool Connection -> IO ()) -> IO ()+withSetup microseconds durable 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+    let combinedConfig = (if durable then durableConfig microseconds else defaultConfig) <> cacheConfig dbCache     migratedConfig <- throwE $ cacheAction (("~/.tmp-postgres/" <>) . BSC.unpack . Base64.encode . hash         $ BSC.pack $ migrationQueryString "int4")         (flip withConn $ flip migrate "int4")@@ -51,7 +64,8 @@  main :: IO () main = do-  [producerCount, consumerCount, time, initialDequeueCount, initialEnqueueCount, batchCount] <- map read <$> getArgs+  [producerCount, consumerCount, time, initialEnqueueCount, enqueueBatchCount, dequeueBatchCount, notify, durable, microseconds]+    <- map read <$> getArgs   -- create a temporary database   enqueueCounter <- newIORef (0 :: Int)   dequeueCounter <- newIORef (0 :: Int)@@ -63,11 +77,16 @@         putStrLn $ "Enqueue Count: " <> show finalEnqueueCount         putStrLn $ "Dequeue Count: " <> show finalDequeueCount -  flip finally printCounters $ withSetup $ \pool -> do+  flip finally printCounters $ withSetup microseconds (1 == durable) $ \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+    let enqueueAction = if notify > 0+          then void $ withResource pool $ \conn -> IO.enqueue "channel" conn E.int4 (replicate enqueueBatchCount payload)+          else void $ withResource pool $ \conn -> I.runThrow (S.enqueue E.int4 (replicate enqueueBatchCount payload)) conn+        dequeueAction = if notify > 0+          then void $ withResource pool $ \conn ->+            IO.withDequeue "channel" conn D.int4 1 dequeueBatchCount (const $ pure ())+          else void $ withResource pool $ \conn -> fix $ \next ->+            I.runThrow (S.dequeue D.int4 dequeueBatchCount) conn >>= \case               [] -> next               _ -> pure () @@ -76,12 +95,6 @@           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"
+ hasql-queue-tmp-db/Main.hs view
@@ -0,0 +1,27 @@+import Database.Postgres.Temp+import Control.Concurrent+import Hasql.Queue.Migrate+import qualified Data.ByteString.Char8 as BSC+import qualified Data.ByteString.Base64.URL as Base64+import           Control.Exception+import           Control.Monad+import           Crypto.Hash.SHA1 (hash)+import           Hasql.Connection++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+++main :: IO ()+main = 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")+        (autoExplainConfig 1  <> cacheConfig dbCache)+  withConfig migratedConfig $ \db -> do+    print $ toConnectionString db+    forever $ threadDelay 100000000
hasql-queue.cabal view
@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 1e85f1ecfea2da607af0d7d470a2487f040dbe1020c661774ca3ebbaec5fcf34+-- hash: 0c6f854b16370d7971eb6b012e349140f35a0b01327d5144a2a21335b7def387  name:           hasql-queue-version:        1.0.1.1+version:        1.2.0.0 synopsis:       A PostgreSQL backed queue description:    A PostgreSQL backed queue. Please see README.md category:       Web@@ -30,10 +30,14 @@  library   exposed-modules:+      Hasql.Queue.High.AtLeastOnce+      Hasql.Queue.High.AtMostOnce+      Hasql.Queue.High.ExactlyOnce       Hasql.Queue.Internal-      Hasql.Queue.IO+      Hasql.Queue.Low.AtLeastOnce+      Hasql.Queue.Low.AtMostOnce+      Hasql.Queue.Low.ExactlyOnce       Hasql.Queue.Migrate-      Hasql.Queue.Session   other-modules:       Paths_hasql_queue   hs-source-dirs:@@ -64,7 +68,7 @@   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+  ghc-options: -Wall -Wno-unused-do-bind -Wno-unused-foralls -O2 -threaded -rtsopts -with-rtsopts=-N   build-depends:       aeson     , async@@ -88,12 +92,48 @@     , transformers   default-language: Haskell2010 +executable hasql-queue-tmp-db+  main-is: Main.hs+  other-modules:+      Paths_hasql_queue+  hs-source-dirs:+      hasql-queue-tmp-db+  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-queue+    , here+    , monad-control+    , postgresql-libpq+    , postgresql-libpq-notify >=0.2.0.0+    , random+    , 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+      Hasql.Queue.High.AtLeastOnceSpec+      Hasql.Queue.High.AtMostOnceSpec+      Hasql.Queue.High.ExactlyOnceSpec+      Hasql.Queue.Low.AtLeastOnceSpec+      Hasql.Queue.Low.AtMostOnceSpec+      Hasql.Queue.Low.ExactlyOnceSpec+      Hasql.Queue.MigrateSpec+      Hasql.Queue.TestUtils       Paths_hasql_queue   hs-source-dirs:       test
+ src/Hasql/Queue/High/AtLeastOnce.hs view
@@ -0,0 +1,99 @@+module Hasql.Queue.High.AtLeastOnce where+import qualified Hasql.Queue.High.ExactlyOnce as H+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++{-|Enqueue a list of payloads.+-}+enqueue :: Connection+        -- ^ Connection+        -> E.Value a+        -- ^ Payload encoder+        -> [a]+        -- ^ List of payloads to enqueue+        -> IO ()+enqueue conn encoder xs = I.runThrow (H.enqueue 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. '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 'failures'+to receive the list of failed payloads.++If the queue is empty 'withDequeue' return 'Nothing'. If there are+any entries 'withDequeue' will wrap the list in 'Just'.+-}+withDequeue :: Connection+            -- ^ Connection+            -> D.Value a+            -- ^ Payload decoder+            -> Int+            -- ^ Retry count+            -> Int+            -- ^ Element count+            -> ([a] -> IO b)+            -- ^ Continuation+            -> IO (Maybe b)+withDequeue = withDequeueWith @IOError+++{-|+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.+-}+failures :: 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)]+failures conn decoder mPayload count = I.runThrow (I.failures decoder mPayload count) conn+++{-|+Permantently remove a failed payload.+-}+delete :: Connection+       -> [I.PayloadId]+       -> IO ()+delete conn xs = I.runThrow (I.delete xs) conn++{-|+A more general configurable version of 'withDequeue'. Unlike 'withDequeue' one+can specify the exception that causes a retry. Additionally event+handlers can be specified to observe the internal behavior of the+retry loop.+-}+withDequeueWith :: forall e a b+                 . Exception e+                => Connection+                -- ^ Connection+                -> D.Value a+                -- ^ Payload decoder+                -> Int+                -- ^ Retry count+                -> Int+                -- ^ Element count+                -> ([a] -> IO b)+                -- ^ Continuation+                -> IO (Maybe b)+withDequeueWith conn decoder retryCount count f = (fix $ \restart i -> do+    try (flip I.runThrow conn $ I.withDequeue decoder retryCount count f) >>= \case+      Right x -> pure x+      Left (e :: e) ->+        if i < retryCount then+          restart $ i + 1+        else+          throwIO e+  ) 0
+ src/Hasql/Queue/High/AtMostOnce.hs view
@@ -0,0 +1,29 @@+module Hasql.Queue.High.AtMostOnce where+import qualified Hasql.Queue.High.ExactlyOnce as H+import qualified Hasql.Queue.Internal as I+import           Hasql.Connection+import qualified Hasql.Encoders as E+import qualified Hasql.Decoders as D++{-|Enqueue a payload.+-}+enqueue :: Connection+        -- ^ Connection+        -> E.Value a+        -- ^ Payload encoder+        -> [a]+        -- ^ List of payloads to enqueue+        -> IO ()+enqueue conn encoder xs = I.runThrow (H.enqueue encoder xs) conn++{-|+Dequeue a list of payloads.+-}+dequeue :: Connection+        -- ^ Connection+        -> D.Value a+        -- ^ Payload decoder+        -> Int+        -- ^ Element count+        -> IO [a]+dequeue conn theDecoder batchCount = I.runThrow (H.dequeue theDecoder batchCount) conn
+ src/Hasql/Queue/High/ExactlyOnce.hs view
@@ -0,0 +1,94 @@+{-|+A high throughput 'Session' based API for a PostgreSQL backed queue.+-}+module Hasql.Queue.High.ExactlyOnce+  ( enqueue+  , dequeue+  ) 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++++{-|Enqueue a payload.+-}+enqueue :: E.Value a+        -- ^ Payload encoder+        -> [a]+        -- ^ List of payloads to enqueue+        -> Session ()+enqueue theEncoder = \case+  []  -> pure ()+  [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+++{-|+Dequeue a list of payloads+-}+dequeue :: D.Value a+        -- ^ Payload decoder+        -> Int+        -- ^ Element count+        -> Session [a]+dequeue valueDecoder count+  | count <= 0 = pure []+  | otherwise = do+  let multipleQuery = [here|+        DELETE FROM payloads+        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|+        DELETE FROM payloads+        WHERE id =+          ( 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
− src/Hasql/Queue/IO.hs 
@@ -1,128 +0,0 @@-{-|-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
@@ -1,3 +1,7 @@+{-|+Internal module. Changes to this modules are not reflected in the+package version.+-} module Hasql.Queue.Internal where import qualified Hasql.Encoders as E import qualified Hasql.Decoders as D@@ -12,16 +16,19 @@ import           Hasql.Statement import           Data.ByteString (ByteString) import           Control.Exception-import           Control.Monad.IO.Class import           Data.Typeable import qualified Database.PostgreSQL.LibPQ as PQ+import           Data.Maybe+import           Control.Monad.IO.Class+import           Data.Text (Text)+import qualified Data.Text.Encoding as TE  -- | 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+data State = Enqueued | Failed   deriving (Show, Eq, Ord, Enum, Bounded)  state :: E.Params a -> D.Result b -> ByteString -> Statement a b@@ -31,8 +38,6 @@ 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@@ -40,7 +45,6 @@ stateEncoder :: E.Value State stateEncoder = E.enum $ \case   Enqueued -> "enqueued"-  Dequeued -> "dequeued"   Failed   -> "failed"  initialPayloadId :: PayloadId@@ -101,8 +105,7 @@ dequeuePayload :: D.Value a -> Int -> Session [Payload a] dequeuePayload valueDecoder count = do   let multipleQuery = [here|-        UPDATE payloads-        SET state='dequeued'+        DELETE FROM payloads         WHERE id in           ( SELECT p1.id             FROM payloads AS p1@@ -116,9 +119,8 @@       multipleEncoder = E.param $ E.nonNullable $ fromIntegral >$< E.int4        singleQuery = [here|-        UPDATE payloads-        SET state='dequeued'-        WHERE id in+        DELETE FROM payloads+        WHERE id =           ( SELECT p1.id             FROM payloads AS p1             WHERE p1.state='enqueued'@@ -178,24 +180,8 @@    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)@@ -233,17 +219,106 @@ 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)+data NoRows = NoRows+  deriving (Show, Eq, Typeable)++instance Exception NoRows++withNotifyWith :: WithNotifyHandlers+               -> Text+               -> Connection+               -> Session a+               -> IO a+withNotifyWith WithNotifyHandlers {..} channel conn action = bracket_+  (flip runThrow conn $ statement channel $ Statement "SELECT listen_on($1)" (E.param $ E.nonNullable E.text) D.noResult True)+  (flip runThrow conn $ statement channel $ Statement "SELECT unlisten_on($1)" (E.param $ E.nonNullable E.text) D.noResult True)   $ fix $ \restart -> do-    x <- runThrow action conn+    x <- try $ runThrow action conn     withNotifyHandlersAfterAction-    case theCast x of-      Nothing -> do+    case x of+      Left NoRows  -> do         -- TODO record the time here         withNotifyHandlersBeforeNotification-        notifyPayload channel conn+        notifyPayload (TE.encodeUtf8 channel) conn         restart-      Just xs -> pure xs+      Right xs -> pure xs++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++  statement (theState, defaultPayloadId, fromIntegral count) theStatement+{-|+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.+-}+failures :: D.Value a+         -- ^ Payload decoder+         -> Maybe PayloadId+         -- ^ Starting position of payloads. Pass 'Nothing' to+         --   start at the beginning+         -> Int+         -- ^ Count+         -> Session [(PayloadId, a)]+failures = listState Failed++-- 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+  -- TODO turn to a save point+  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"++delete :: [PayloadId] -> Session ()+delete xs = do+  let theQuery = [here|+        DELETE FROM payloads+        WHERE id = ANY($1)+        |]++      encoder = E.param+              $ E.nonNullable+              $ E.foldableArray+              $ E.nonNullable payloadIdEncoder++  statement xs $ Statement theQuery encoder D.noResult True
+ src/Hasql/Queue/Low/AtLeastOnce.hs view
@@ -0,0 +1,125 @@+{-|+A high throughput 'Session' based API for a PostgreSQL backed queue.+-}+module Hasql.Queue.Low.AtLeastOnce+  ( enqueue+  , withDequeue+  -- ** Listing API+  , I.PayloadId+  , failures+  , delete+  -- ** Advanced API+  , withDequeueWith+  , I.WithNotifyHandlers (..)+  ) where++import qualified Hasql.Queue.Low.ExactlyOnce as E+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.Text (Text)+import           Control.Monad.IO.Class++{-|Enqueue a list of payloads.+-}+enqueue :: Text+        -- ^ 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 (E.enqueue 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. '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 'failures'+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 :: Text+            -- ^ 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.+-}+failures :: 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)]+failures conn decoder mPayload count = I.runThrow (I.failures decoder mPayload count) conn++{-|+Permantently remove a failed payload.+-}+delete :: Connection+       -> [I.PayloadId]+       -> IO ()+delete conn xs = I.runThrow (I.delete xs) conn++{-|+A more general configurable version of 'withDequeue'. Unlike 'withDequeue' one+can specify the exception that causes a retry. Additionally event+handlers can be specified to observe the internal behavior of the+retry loop.+-}+withDequeueWith :: forall e a b+                 . Exception e+                => I.WithNotifyHandlers+                -- ^ Event handlers for events that occur as 'withDequeWith' loops+                -> Text+                -- ^ 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+    let action = I.withDequeue decoder retryCount count f >>= \case+          Nothing -> liftIO $ throwIO I.NoRows+          Just x  -> pure x++    try (I.withNotifyWith withNotifyHandlers channel conn action) >>= \case+      Right x -> pure x+      Left (e :: e) ->+        if i < retryCount then+          restart $ i + 1+        else+          throwIO e+  ) 0
+ src/Hasql/Queue/Low/AtMostOnce.hs view
@@ -0,0 +1,36 @@+module Hasql.Queue.Low.AtMostOnce where+import qualified Hasql.Queue.Low.ExactlyOnce as E+import           Hasql.Connection+import qualified Hasql.Encoders as E+import qualified Hasql.Decoders as D+import           Data.Text(Text)+import qualified Hasql.Queue.Internal as I+++{-|Enqueue a payload.+-}+enqueue :: Text+        -- ^ 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 (E.enqueue channel encoder xs) conn++{-|+Dequeue a list of payloads.+-}+dequeue :: Text+        -- ^ Notification channel name. Any valid PostgreSQL identifier+        -> Connection+        -- ^ Connection+        -> D.Value a+        -- ^ Payload decoder+        -> Int+        -- ^ Element count+        -> IO [a]+dequeue channel conn theDecoder batchCount =+  E.withDequeue channel conn theDecoder batchCount id
+ src/Hasql/Queue/Low/ExactlyOnce.hs view
@@ -0,0 +1,64 @@+module Hasql.Queue.Low.ExactlyOnce+  ( enqueue+  , withDequeue+  , withDequeueWith+  ) where+import qualified Hasql.Queue.High.ExactlyOnce as H+import Control.Exception+import qualified Hasql.Encoders as E+import qualified Hasql.Decoders as D+import           Hasql.Session+import           Hasql.Statement+import           Hasql.Connection+import qualified Hasql.Queue.Internal as I+import Control.Monad.IO.Class+import           Data.Text(Text)++{-|Enqueue a payload send a notification on the+specified channel.+-}+enqueue :: Text+        -- ^ Notification channel name. Any valid PostgreSQL identifier+        -> E.Value a+        -- ^ Payload encoder+        -> [a]+        -- ^ List of payloads to enqueue+        -> Session ()+enqueue channel theEncoder values = do+  H.enqueue theEncoder values+  statement channel $ Statement "SELECT notify_on($1)" (E.param $ E.nonNullable E.text) D.noResult True++dequeueOrRollbackAndThrow :: D.Value a -> Int -> Session [a]+dequeueOrRollbackAndThrow theDecoder dequeueCount = H.dequeue theDecoder dequeueCount >>= \case+  [] -> liftIO $ throwIO I.NoRows+  xs -> pure xs++withDequeue :: Text+            -- ^ Notification channel name. Any valid PostgreSQL identifier+            -> Connection+            -- ^ Connection+            -> D.Value a+            -- ^ Payload decoder+            -> Int+            -- ^ Batch count+            -> (Session [a] -> Session b)+            -- ^ Transaction runner+            -> IO b+withDequeue = withDequeueWith mempty++withDequeueWith :: I.WithNotifyHandlers+                -- ^ Event handlers for events that occur as 'withDequeWith' loops+                -> Text+                -- ^ Notification channel name. Any valid PostgreSQL identifier+                -> Connection+                -- ^ Connection+                -> D.Value a+                -- ^ Payload decoder+                -> Int+                -- ^ Batch count+                -> (Session [a] -> Session b)+                -- ^ Transaction runner+                -> IO b+withDequeueWith withNotifyHandlers channel conn theDecoder dequeueCount runner+  = I.withNotifyWith withNotifyHandlers channel conn+  $ runner (dequeueOrRollbackAndThrow theDecoder dequeueCount)
src/Hasql/Queue/Migrate.hs view
@@ -9,7 +9,7 @@ {-# OPTIONS_HADDOCK prune #-} {-# LANGUAGE QuasiQuotes, OverloadedStrings #-} module Hasql.Queue.Migrate where-import           Control.Monad+import qualified Hasql.Queue.Internal as I import           Data.String import           Data.String.Here.Interpolated import           Hasql.Connection@@ -24,10 +24,28 @@                      -- @jsonb@.                      -> String migrationQueryString valueType = [i|+  CREATE OR REPLACE FUNCTION notify_on(channel text) RETURNs VOID AS $$+    BEGIN+      EXECUTE (format(E'NOTIFY %I', channel));+    END;+  $$ LANGUAGE plpgsql;++  CREATE OR REPLACE FUNCTION listen_on(channel text) RETURNS VOID AS $$+    BEGIN+      EXECUTE (format(E'LISTEN %I', channel));+    END;+  $$ LANGUAGE plpgsql;++  CREATE OR REPLACE FUNCTION unlisten_on(channel text) RETURNS VOID AS $$+    BEGIN+      EXECUTE (format(E'UNLISTEN %I', channel));+    END;+  $$ LANGUAGE plpgsql;+     DO $$   BEGIN     IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'state_t') THEN-      CREATE TYPE state_t AS ENUM ('enqueued', 'dequeued', 'failed');+      CREATE TYPE state_t AS ENUM ('enqueued', 'failed');     END IF;   END$$; @@ -39,9 +57,9 @@   , 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)+  CREATE INDEX IF NOT EXISTS active_modified_at_idx ON payloads USING btree (modified_at, state)     WHERE (state = 'enqueued');  |]@@ -51,9 +69,28 @@   @  DO $$+  CREATE OR REPLACE FUNCTION notify_on(channel text) RETURNs VOID AS $$+    BEGIN+      EXECUTE (format(E'NOTIFY %I', channel));+    END;+  $$ LANGUAGE plpgsql;++  CREATE OR REPLACE FUNCTION listen_on(channel text) RETURNS VOID AS $$+    BEGIN+      EXECUTE (format(E'LISTEN %I', channel));+    END;+  $$ LANGUAGE plpgsql;++  CREATE OR REPLACE FUNCTION unlisten_on(channel text) RETURNS VOID AS $$+    BEGIN+      EXECUTE (format(E'UNLISTEN %I', channel));+    END;+  $$ LANGUAGE plpgsql;++    DO $$   BEGIN     IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'state_t') THEN-      CREATE TYPE state_t AS ENUM ('enqueued', 'dequeued', 'failed');+      CREATE TYPE state_t AS ENUM ('enqueued', 'failed');     END IF;   END$$; @@ -64,10 +101,10 @@   , 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);+  , value ${VALUE_TYPE} NOT NULL+  ); -  CREATE INDEX IF NOT EXISTS active_modified_at_idx ON payloads USING btree (modified_at)+  CREATE INDEX IF NOT EXISTS active_modified_at_idx ON payloads USING btree (modified_at, state)     WHERE (state = 'enqueued');  @ @@ -77,5 +114,21 @@         -> String         -- ^ The type of the @value@ column         -> IO ()-migrate conn valueType = void $-  run (sql $ fromString $ migrationQueryString valueType) conn+migrate conn valueType =+  I.runThrow (sql $ fromString $ migrationQueryString valueType) conn++{-|+Drop everything created by 'migrate'+-}+teardown :: Connection -> IO ()+teardown conn = do+  let theQuery = [i|+        DROP TABLE IF EXISTS payloads;+        DROP TYPE IF EXISTS state_t;+        DROP SEQUENCE IF EXISTS modified_index;+        DROP FUNCTION IF EXISTS notify_on;+        DROP FUNCTION IF EXISTS listen_on;+        DROP FUNCTION IF EXISTS unlisten_on;+      |]++  I.runThrow (sql theQuery) conn
− src/Hasql/Queue/Session.hs 
@@ -1,180 +0,0 @@-{-|-'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 
@@ -1,206 +0,0 @@-{-# 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 
@@ -1,226 +0,0 @@-{-# 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/Hasql/Queue/High/AtLeastOnceSpec.hs view
@@ -0,0 +1,92 @@+module Hasql.Queue.High.AtLeastOnceSpec where+import           Hasql.Queue.High.AtLeastOnce+import qualified Hasql.Encoders as E+import qualified Hasql.Decoders as D+import           Test.Hspec                     (Spec, describe, parallel, it)+import           Test.Hspec.Expectations.Lifted+import           Test.Hspec.Core.Spec (sequential)+import           Hasql.Queue.TestUtils+import qualified Hasql.Queue.Internal as I+import           Control.Exception as E+import           Hasql.Connection+import           Data.Typeable+import           Data.IORef+import           Control.Monad++data FailedwithDequeue = FailedwithDequeue+  deriving (Show, Eq, Typeable)++instance Exception FailedwithDequeue++getPayload :: Connection -> D.Value a -> I.PayloadId -> IO (Maybe (I.Payload a))+getPayload conn decoder payloadId = I.runThrow (I.getPayload decoder payloadId) conn++spec :: Spec+spec = describe "Hasql.Queue.High.AtLeastOnce" $ parallel $ do+  sequential $ aroundAll withSetup $ describe "enqueue/dequeue" $ do+    it "enqueue nothing gives nothing" $ withConnection $ \conn -> do+      enqueue conn E.int4 []+      withDequeue conn D.int4 1 1 pure `shouldReturn` Nothing++    it "enqueue 1 gives 1" $ withConnection $ \conn -> do+      enqueue conn E.int4 [1]+      withDequeue conn D.int4 1 1 pure `shouldReturn` Just [1]++    it "dequeue give nothing after enqueueing everything" $ withConnection $ \conn -> do+      withDequeue conn D.int4 1 1 pure `shouldReturn` Nothing++    it "dequeueing is in FIFO order" $ withConnection $ \conn -> do+      enqueue conn E.int4 [1]+      enqueue conn E.int4 [2]+      withDequeue conn D.int4 1 1 pure `shouldReturn` Just [1]+      withDequeue conn D.int4 1 1 pure `shouldReturn` Just [2]++    it "dequeueing a batch of elements works" $ withConnection $ \conn -> do+      enqueue conn E.int4 [1, 2, 3]+      withDequeue conn D.int4 1 2 pure `shouldReturn` Just [1, 2]++      withDequeue conn D.int4 1 2 pure `shouldReturn` Just [3]++    it "withDequeue fails if a non IOError is thrown" $ withConnection $ \conn -> do+      enqueue conn E.int4 [1]+      handle (\FailedwithDequeue -> pure Nothing) $+        withDequeue conn D.int4 2 1 $ \_ -> throwIO FailedwithDequeue++      failures conn D.int4 Nothing 1 `shouldReturn` []+      withDequeue conn D.int4 0 1 pure `shouldReturn` Just [1]++    it "withDequeue fails if throws occur and retry is zero" $ withConnection $ \conn -> do+      enqueue conn E.int4 [1]+      handle (\(_ :: IOError) -> pure Nothing) $+        withDequeue conn D.int4 0 1 $ \_ -> throwIO $ userError "hey"++      [(pId, x)] <- failures conn D.int4 Nothing 1+      x `shouldBe` 1+      delete conn [pId]++    it "withDequeue succeeds even if the first attempt fails" $ withConnection $ \conn -> do+      enqueue conn E.int4 [1]++      ref <- newIORef (0 :: Int)++      withDequeue conn D.int4 1 1 (\_ -> do+        count <- readIORef ref+        writeIORef ref $ count + 1+        when (count < 1) $ throwIO $ userError "hey"+        pure '!') `shouldReturn` Just '!'++      withDequeue conn D.int4 1 1 pure `shouldReturn` Nothing+      readIORef ref `shouldReturn` 2++    it "failures paging works" $ withConnection $ \conn -> do+      enqueue conn E.int4 [2]+      enqueue conn E.int4 [3]++      handle (\(_ :: IOError) -> pure Nothing) $+        withDequeue conn D.int4 0 1 $ \_ -> throwIO $ userError "fds"+      handle (\(_ :: IOError) -> pure Nothing) $+        withDequeue conn D.int4 0 1 $ \_ -> throwIO $ userError "fds"++      [(next, x)] <- failures conn D.int4 Nothing 1+      x `shouldBe` 2+      fmap (fmap snd) (failures conn D.int4 (Just next) 2) `shouldReturn` [3]
+ test/Hasql/Queue/High/AtMostOnceSpec.hs view
@@ -0,0 +1,34 @@+module Hasql.Queue.High.AtMostOnceSpec where+import           Hasql.Queue.High.AtMostOnce+import qualified Hasql.Encoders as E+import qualified Hasql.Decoders as D+import           Test.Hspec                     (Spec, describe, parallel, it)+import           Test.Hspec.Expectations.Lifted+import           Test.Hspec.Core.Spec (sequential)+import           Hasql.Queue.TestUtils++spec :: Spec+spec = describe "Hasql.Queue.High.AtMostOnce" $ parallel $ do+  sequential $ aroundAll withSetup $ describe "enqueue/dequeue" $ do+    it "enqueue nothing gives nothing" $ withConnection $ \conn -> do+      enqueue conn E.int4 []+      dequeue conn D.int4 1 `shouldReturn` []++    it "enqueue 1 gives 1" $ withConnection $ \conn -> do+      enqueue conn E.int4 [1]+      dequeue conn D.int4 1 `shouldReturn` [1]++    it "dequeue give nothing after enqueueing everything" $ withConnection $ \conn -> do+      dequeue conn D.int4 1 `shouldReturn` []++    it "dequeueing is in FIFO order" $ withConnection $ \conn -> do+      enqueue conn E.int4 [1]+      enqueue conn E.int4 [2]+      dequeue conn D.int4 1 `shouldReturn` [1]+      dequeue conn D.int4 1 `shouldReturn` [2]++    it "dequeueing a batch of elements works" $ withConnection $ \conn -> do+      enqueue conn E.int4 [1, 2, 3]+      dequeue conn D.int4 2 `shouldReturn` [1, 2]++      dequeue conn D.int4 2 `shouldReturn` [3]
+ test/Hasql/Queue/High/ExactlyOnceSpec.hs view
@@ -0,0 +1,47 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards   #-}+{-# LANGUAGE ScopedTypeVariables   #-}+module Hasql.Queue.High.ExactlyOnceSpec where+import           Control.Exception as E+import           Hasql.Queue.High.ExactlyOnce+import           Test.Hspec                     (Spec, describe, parallel, it)+import           Test.Hspec.Expectations.Lifted+import           Test.Hspec.Core.Spec (sequential)+import qualified Hasql.Encoders as E+import qualified Hasql.Decoders as D+import           Data.Typeable+import           Data.Int+import           Hasql.Queue.TestUtils++-- Fix this to be more of what I would expec++newtype TooManyRetries = TooManyRetries Int64+  deriving (Show, Eq, Typeable)++instance Exception TooManyRetries++spec :: Spec+spec = describe "Hasql.Queue.High.ExactlyOnce" $ parallel $ do+  sequential $ aroundAll withSetup $ describe "enqueue/dequeue" $ do+    it "enqueue nothing gives nothing" $ withReadCommitted $ do+      enqueue E.int4 []+      dequeue D.int4 1 `shouldReturn` []++    it "enqueue 1 gives 1" $ withReadCommitted $ do+      enqueue E.int4 [1]+      dequeue D.int4 1 `shouldReturn` [1]++    it "dequeue give nothing after enqueueing everything" $ withReadCommitted $ do+      dequeue D.int4 1 `shouldReturn` []++    it "dequeueing is in FIFO order" $ withReadCommitted $ do+      enqueue E.int4 [1]+      enqueue E.int4 [2]+      dequeue D.int4 1 `shouldReturn` [1]+      dequeue D.int4 1 `shouldReturn` [2]++    it "dequeueing a batch of elements works" $ withReadCommitted $ do+      enqueue E.int4 [1, 2, 3]+      dequeue D.int4 2 `shouldReturn` [1, 2]++      dequeue D.int4 2 `shouldReturn` [3]
+ test/Hasql/Queue/Low/AtLeastOnceSpec.hs view
@@ -0,0 +1,149 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards   #-}+{-# LANGUAGE ScopedTypeVariables   #-}+module Hasql.Queue.Low.AtLeastOnceSpec 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.Low.AtLeastOnce+import           Test.Hspec                     (Spec, describe, it)+import           Test.Hspec.Expectations.Lifted+import           Data.List.Split+import           Data.Text(Text)+import           Hasql.Connection+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 (..))+import           Hasql.Queue.TestUtils+import           System.Timeout++getCount :: Connection -> IO Int64+getCount = I.runThrow I.getCount++getPayload :: Connection -> D.Value a -> I.PayloadId -> IO (Maybe (I.Payload a))+getPayload conn decoder payloadId = I.runThrow (I.getPayload decoder payloadId) conn++channel :: Text+channel = "hey"++data FailedwithDequeue = FailedwithDequeue+  deriving (Show, Eq, Typeable)++instance Exception FailedwithDequeue++spec :: Spec+spec = describe "Hasql.Queue.Low.AtLeastOnce" $ aroundAll withSetup $ describe "enqueue/withDequeue" $ do+  it "enqueue nothing timesout" $ withConnection $ \conn -> do+    enqueue channel conn E.int4 []+    timeout 100000 (withDequeue channel conn D.int4 1 1 pure) `shouldReturn` Nothing++  it "enqueue 1 gives 1" $ withConnection $ \conn -> do+    enqueue channel conn E.int4 [1]+    withDequeue channel conn D.int4 1 1 pure `shouldReturn` [1]++  it "dequeue timesout after enqueueing everything" $ withConnection $ \conn -> do+    timeout 100000 (withDequeue channel conn D.int4 1 1 pure) `shouldReturn` Nothing++  it "dequeueing is in FIFO order" $ withConnection $ \conn -> do+    enqueue channel conn E.int4 [1]+    enqueue channel conn E.int4 [2]+    withDequeue channel conn D.int4 1 1 pure `shouldReturn` [1]+    withDequeue channel conn D.int4 1 1 pure `shouldReturn` [2]++  it "dequeueing a batch of elements works" $ withConnection $ \conn -> do+    enqueue channel conn E.int4 [1, 2, 3]+    withDequeue channel conn D.int4 1 2 pure `shouldReturn` [1, 2]++    withDequeue channel conn D.int4 1 1 pure `shouldReturn` [3]++  it "withDequeue blocks until something is enqueued: before" $ withConnection $ \conn -> do+    void $ enqueue channel conn E.int4 [1]+    res <- withDequeue channel conn D.int4 1 1 pure+    res `shouldBe` [1]++  it "withDequeue blocks until something is enqueued: during" $ withConnection $ \conn -> do+    afterActionMVar  <- newEmptyMVar+    beforeNotifyMVar <- newEmptyMVar++    let handlers = I.WithNotifyHandlers+          { withNotifyHandlersAfterAction        = putMVar afterActionMVar ()+          , withNotifyHandlersBeforeNotification = takeMVar beforeNotifyMVar+          }++    -- This is the definition of IO.dequeue+    resultThread <- async $ withDequeueWith @IOError 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" $ withConnection2 $ \(conn1, conn2) -> do+    thread <- async $ withDequeue channel conn1 D.int4 1 1 pure+    timeout 100000 (wait thread) `shouldReturn` Nothing++    enqueue channel conn2 E.int4 [1]++    wait thread `shouldReturn` [1]++  -- TODO redo just using failures+  it "withDequeue fails and sets the retries to +1" $ withConnection $ \conn -> do+    enqueue channel conn E.int4 [1]+    handle (\(_ :: IOError) -> pure ()) $ withDequeue channel conn D.int4 0 1 $ \_ -> throwIO $ userError "hey"+    xs <- failures conn D.int4 Nothing 1++    map snd xs `shouldBe` [1]++  it "withDequeue succeeds even if the first attempt fails" $ withConnection $ \conn -> do+    [payloadId] <- I.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` '!'++    getPayload conn D.int4 payloadId `shouldReturn` Nothing++  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] <- I.runThrow (I.enqueuePayload E.int4 [1]) conn+    Just actual <- getPayload conn D.int4 payloadId++    pValue actual `shouldBe` 1
+ test/Hasql/Queue/Low/AtMostOnceSpec.hs view
@@ -0,0 +1,48 @@+module Hasql.Queue.Low.AtMostOnceSpec where+import           Hasql.Queue.Low.AtMostOnce+import qualified Hasql.Encoders as E+import qualified Hasql.Decoders as D+import           Test.Hspec                     (Spec, describe, parallel, it)+import           Test.Hspec.Expectations.Lifted+import           Test.Hspec.Core.Spec (sequential)+import           Hasql.Queue.TestUtils+import           Data.Text(Text)+import           System.Timeout+import           Control.Concurrent.Async++channel :: Text+channel = "channel"++spec :: Spec+spec = describe "Hasql.Queue.Low.AtMostOnce" $ parallel $ do+  sequential $ aroundAll withSetup $ describe "enqueue/dequeue" $ do+    it "enqueue nothing timesout" $ withConnection $ \conn -> do+      enqueue channel conn E.int4 []+      timeout 100000 (dequeue channel conn D.int4 1) `shouldReturn` Nothing++    it "enqueue 1 gives 1" $ withConnection $ \conn -> do+      enqueue channel conn E.int4 [1]+      dequeue channel conn D.int4 1 `shouldReturn` [1]++    it "dequeue timesout after enqueueing everything" $ withConnection $ \conn -> do+      timeout 100000 (dequeue channel conn D.int4 1) `shouldReturn` Nothing++    it "dequeueing is in FIFO order" $ withConnection $ \conn -> do+      enqueue channel conn E.int4 [1]+      enqueue channel conn E.int4 [2]+      dequeue channel conn D.int4 1 `shouldReturn` [1]+      dequeue channel conn D.int4 1 `shouldReturn` [2]++    it "dequeueing a batch of elements works" $ withConnection $ \conn -> do+      enqueue channel conn E.int4 [1, 2, 3]+      dequeue channel conn D.int4 2 `shouldReturn` [1, 2]++      dequeue channel conn D.int4 2 `shouldReturn` [3]++    it "dequeueing blocks until something is enqueued" $ withConnection2 $ \(conn1, conn2) -> do+      thread <- async $ dequeue channel conn1 D.int4 1+      timeout 100000 (wait thread) `shouldReturn` Nothing++      enqueue channel conn2 E.int4 [1]++      wait thread `shouldReturn` [1]
+ test/Hasql/Queue/Low/ExactlyOnceSpec.hs view
@@ -0,0 +1,85 @@+module Hasql.Queue.Low.ExactlyOnceSpec where+import           Control.Exception as E+import           Hasql.Queue.Low.ExactlyOnce+import qualified Hasql.Queue.Internal as I+import           Test.Hspec                     (Spec, describe, it)+import           Test.Hspec.Expectations.Lifted+import qualified Hasql.Encoders as E+import qualified Hasql.Decoders as D+import           Data.Typeable+import           Data.Int+import           Hasql.Queue.TestUtils+import           System.Timeout+import           Control.Concurrent.Async+import           Hasql.Queue.Internal (runThrow)+import           Control.Concurrent+import           Control.Monad+import           Data.Text (Text)++-- Fix this to be more of what I would expec++newtype TooManyRetries = TooManyRetries Int64+  deriving (Show, Eq, Typeable)++instance Exception TooManyRetries++channel :: Text+channel = "channel"++spec :: Spec+spec = describe "Hasql.Queue.High.ExactlyOnce" $ do+  aroundAll withSetup $ describe "enqueue/withDequeue" $ do+    it "enqueue nothing timesout" $ withConnection $ \conn -> do+      runThrow (enqueue channel E.int4 []) conn+      timeout 100000 (withDequeue channel conn D.int4 1 id) `shouldReturn` Nothing++    it "enqueue 1 gives 1" $ withConnection $ \conn -> do+      runThrow (enqueue channel E.int4 [1]) conn+      withDequeue channel conn D.int4 1 id `shouldReturn` [1]++    it "dequeue timesout after enqueueing everything" $ withConnection $ \conn -> do+      timeout 100000 (withDequeue channel conn D.int4 1 id) `shouldReturn` Nothing++    it "dequeueing is in FIFO order" $ withConnection $ \conn -> do+      runThrow (enqueue channel E.int4 [1]) conn+      runThrow (enqueue channel E.int4 [2]) conn+      withDequeue channel conn D.int4 1 id `shouldReturn` [1]+      withDequeue channel conn D.int4 1 id `shouldReturn` [2]++    it "dequeueing a batch of elements works" $ withConnection $ \conn -> do+      runThrow (enqueue channel E.int4 [1, 2, 3]) conn+      withDequeue channel conn D.int4 1 id `shouldReturn` [1, 2]++      withDequeue channel conn D.int4 1 id `shouldReturn` [3]++    it "withDequeue blocks until something is enqueued: before" $ withConnection $ \conn -> do+      void $ runThrow (enqueue channel E.int4 [1]) conn+      res <- withDequeue channel conn D.int4 1 id+      res `shouldBe` [1]++    it "withDequeue blocks until something is enqueued: during" $ withConnection $ \conn -> do+      afterActionMVar  <- newEmptyMVar+      beforeNotifyMVar <- newEmptyMVar++      let handlers = I.WithNotifyHandlers+            { withNotifyHandlersAfterAction        = putMVar afterActionMVar ()+            , withNotifyHandlersBeforeNotification = takeMVar beforeNotifyMVar+            }++      -- This is the definition of IO.dequeue+      resultThread <- async $ withDequeueWith handlers channel conn D.int4 1 id+      takeMVar afterActionMVar++      void $ runThrow (enqueue "hey" E.int4 [1]) conn++      putMVar beforeNotifyMVar ()++      wait resultThread `shouldReturn` [1]++    it "withDequeue blocks until something is enqueued: after" $ withConnection2 $ \(conn1, conn2) -> do+      thread <- async $ withDequeue channel conn1 D.int4 1 id+      timeout 100000 (wait thread) `shouldReturn` Nothing++      runThrow (enqueue channel E.int4 [1]) conn2++      wait thread `shouldReturn` [1]
+ test/Hasql/Queue/MigrateSpec.hs view
@@ -0,0 +1,57 @@+module Hasql.Queue.MigrateSpec where+import Hasql.Queue.Migrate+import           Hasql.Queue.TestUtils+import           Test.Hspec                     (Spec, describe, parallel, it, shouldReturn)+import           Test.Hspec.Core.Spec (sequential)+import           Control.Monad.IO.Class+import qualified Hasql.Decoders as D+import           Hasql.Session+import           Hasql.Statement+import           Data.String.Here.Uninterpolated+import qualified Hasql.Queue.Internal as I+++spec :: Spec+spec = describe "Hasql.Queue.High.ExactlyOnce" $ parallel $ do+  sequential $ aroundAll withSetup $ describe "basic" $ do+    it "is okay to migrate multiple times" $ withConnection $ \conn ->+      liftIO $ migrate conn "int4"++    it "drops all items" $ withConnection $ \conn -> do+      teardown conn++      let theQuery = [here|+            SELECT EXISTS (+            SELECT FROM information_schema.tables+            WHERE  table_schema = 'public'+            AND    table_name   = 'payloads'+            );+          |]++          decoder = D.singleRow $ D.column $ D.nonNullable D.bool+      I.runThrow (statement () $ Statement theQuery mempty decoder True) conn `shouldReturn` False++      let theQuery' = [here|+             SELECT EXISTS (+             SELECT FROM pg_catalog.pg_class c+             JOIN   pg_catalog.pg_namespace n ON n.oid = c.relnamespace+             WHERE  n.nspname = 'public'+             AND    c.relname = 'modified_index'+          )+          |]++          decoder' = D.singleRow $ D.column $ D.nonNullable D.bool+      I.runThrow (statement () $ Statement theQuery' mempty decoder' True) conn `shouldReturn` False++      let theQuery'' = [here|+             SELECT EXISTS (+             SELECT 1+             FROM pg_type t+             JOIN pg_catalog.pg_namespace n ON n.oid = t.typnamespace+             WHERE t.typname = 'state_t'+             AND n.nspname = 'public'+          )+          |]++          decoder'' = D.singleRow $ D.column $ D.nonNullable D.bool+      I.runThrow (statement () $ Statement theQuery'' mempty decoder'' True) conn `shouldReturn` False
+ test/Hasql/Queue/TestUtils.hs view
@@ -0,0 +1,89 @@+module Hasql.Queue.TestUtils where+import qualified Data.ByteString.Char8 as BSC+import Control.Concurrent.Async+import           Database.Postgres.Temp as Temp+import           Test.Hspec+import           Control.Exception as E+import           Data.Pool+import           Hasql.Connection+import           Hasql.Session+import qualified Data.ByteString.Base64.URL as Base64+import           Control.Concurrent+import           Data.IORef+import           Data.Foldable+import           Control.Monad ((<=<))+import           Crypto.Hash.SHA1 (hash)+import           Hasql.Queue.Migrate++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++withConnection2 :: ((Connection, Connection) -> IO ()) -> Pool Connection -> IO ()+withConnection2 f pool = withResource pool $ \conn1 ->+   withResource pool $ \conn2 -> f (conn1, conn2)++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
test/Main.hs view
@@ -1,9 +1,19 @@ import           Test.Hspec-import           Database.Hasql.Queue.SessionSpec as S-import           Database.Hasql.Queue.IOSpec as I+import           Hasql.Queue.High.ExactlyOnceSpec as HE+import           Hasql.Queue.High.AtLeastOnceSpec as HL+import           Hasql.Queue.High.AtMostOnceSpec as HM+import           Hasql.Queue.Low.AtLeastOnceSpec as LL+import           Hasql.Queue.Low.AtMostOnceSpec as LM+import           Hasql.Queue.Low.AtMostOnceSpec as LE+import           Hasql.Queue.MigrateSpec as M   main :: IO () main = hspec $ do-  S.spec-  I.spec+  HE.spec+  HM.spec+  HL.spec+  LL.spec+  LM.spec+  LE.spec+  M.spec