diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,6 @@ +# consumers-2.2.0.1 (2019-05-22) +* Support hpqtypes-1.7.0.0 and hpqtypes-extras-1.9.0.0. + # consumers-2.2.0.0 (2019-02-19) * When jobs are relased from a dead consumer, reset their finished_at column * Start the consumer only when ccMaxRunningJobs >= 1 diff --git a/consumers.cabal b/consumers.cabal --- a/consumers.cabal +++ b/consumers.cabal @@ -1,5 +1,5 @@ name: consumers -version: 2.2.0.0 +version: 2.2.0.1 synopsis: Concurrent PostgreSQL data consumers description: Library for setting up concurrent consumers of data @@ -31,20 +31,20 @@ Database.PostgreSQL.Consumers.Components, Database.PostgreSQL.Consumers.Utils - build-depends: base >4 && <5, - containers, - exceptions, - extra, - hpqtypes >= 1.6 && < 1.7, - lifted-base, - lifted-threads, - log-base >= 0.7, - monad-control, - monad-time, - mtl, - stm, - time, - transformers-base + build-depends: base >= 4.9 && < 4.13 + , containers >= 0.5 && < 0.7 + , exceptions >= 0.10 && < 0.11 + , extra >= 1.6 && < 1.7 + , hpqtypes >= 1.7 && < 1.8 + , lifted-base >= 0.2 && < 0.3 + , lifted-threads >= 1.0 && < 1.1 + , log-base >= 0.7 && < 0.9 + , monad-control >= 1.0 && < 1.1 + , monad-time >= 0.3 && < 0.4 + , mtl >= 2.2 && < 2.3 + , stm >= 2.4 && < 2.6 + , time >= 1.6 && < 1.9 + , transformers-base >= 0.4 && < 0.5 hs-source-dirs: src @@ -74,7 +74,7 @@ build-depends: base, consumers, hpqtypes, - hpqtypes-extras, + hpqtypes-extras >= 1.9 && < 2.0, log-base, text, text-show, diff --git a/example/Example.hs b/example/Example.hs --- a/example/Example.hs +++ b/example/Example.hs @@ -36,13 +36,14 @@ [] -> defaultConnString (cs:_) -> cs - let connSettings = def { csConnInfo = T.pack connString } + let connSettings = defaultConnectionSettings + { csConnInfo = T.pack connString } ConnectionSource connSource = simpleSource connSettings -- Monad stack initialisation. withSimpleStdOutLogger $ \logger -> runLogT "consumers-example" logger $ - runDBT connSource {- transactionSettings -} def $ do + runDBT connSource defaultTransactionSettings $ do -- Initialise. createTables @@ -72,14 +73,17 @@ createTables :: AppM () createTables = do - migrateDatabase {- options -} def {- extensions -} [] {- domains -} [] + migrateDatabase defaultExtrasOptions + {- extensions -} [] {- composites -} [] {- domains -} [] tables migrations - checkDatabase {- options -} def {- domains -} [] tables + checkDatabase defaultExtrasOptions + {- composites -} [] {- domains -} [] + tables dropTables :: AppM () dropTables = do - migrateDatabase {- options -} def {- extensions -} [] {- domains -} [] - {- tables -} [] + migrateDatabase defaultExtrasOptions + {- extensions -} [] {- composites -} [] {- domains -} [] {- tables -} [] [ dropTableMigration jobsTable , dropTableMigration consumersTable ] diff --git a/src/Database/PostgreSQL/Consumers/Components.hs b/src/Database/PostgreSQL/Consumers/Components.hs --- a/src/Database/PostgreSQL/Consumers/Components.hs +++ b/src/Database/PostgreSQL/Consumers/Components.hs @@ -72,7 +72,8 @@ runningJobsInfo <- liftBase $ newTVarIO M.empty runningJobs <- liftBase $ newTVarIO 0 - skipLockedTest :: Either DBException () <- try . runDBT cs def $ + skipLockedTest :: Either DBException () <- + try . runDBT cs defaultTransactionSettings $ runSQL_ "SELECT TRUE FOR UPDATE SKIP LOCKED" let useSkipLocked = either (const False) (const True) skipLockedTest @@ -139,7 +140,7 @@ signalDispatcher = do liftBase $ tryPutMVar semaphore () - noTs = def { + noTs = defaultTransactionSettings { tsAutoTransaction = False } @@ -395,7 +396,7 @@ ---------------------------------------- ts :: TransactionSettings -ts = def { +ts = defaultTransactionSettings { -- PostgreSQL doesn't seem to handle very high amount of -- concurrent transactions that modify multiple rows in -- the same table well (see updateJobs) and sometimes (very diff --git a/src/Database/PostgreSQL/Consumers/Config.hs b/src/Database/PostgreSQL/Consumers/Config.hs --- a/src/Database/PostgreSQL/Consumers/Config.hs +++ b/src/Database/PostgreSQL/Consumers/Config.hs @@ -99,7 +99,9 @@ , ccNotificationTimeout :: !Int -- | Maximum amount of jobs that can be processed in parallel. , ccMaxRunningJobs :: !Int --- | Function that processes a job. +-- | Function that processes a job. It's recommended to process each +-- job in a separate DB transaction, otherwise you'll have to remember +-- to commit your changes to the database manually. , ccProcessJob :: !(job -> m Result) -- | Action taken if a job processing function throws an exception. -- Note that if this action throws an exception, the consumer goes diff --git a/src/Database/PostgreSQL/Consumers/Consumer.hs b/src/Database/PostgreSQL/Consumers/Consumer.hs --- a/src/Database/PostgreSQL/Consumers/Consumer.hs +++ b/src/Database/PostgreSQL/Consumers/Consumer.hs @@ -49,7 +49,7 @@ ] fetchOne runIdentity where - ts = def { + ts = defaultTransactionSettings { tsAutoTransaction = False } @@ -74,7 +74,7 @@ , " AND name =" unRawSQL ccJobsTable ] where - ts = def { + ts = defaultTransactionSettings { tsRestartPredicate = Just . RestartPredicate $ \e _ -> qeErrorCode e == DeadlockDetected } diff --git a/test/Test.hs b/test/Test.hs --- a/test/Test.hs +++ b/test/Test.hs @@ -62,7 +62,7 @@ runTestEnv :: ConnectionSourceM (LogT IO) -> Logger -> TestEnv a -> IO a runTestEnv connSource logger m = (runLogT "consumers-test" logger) - . (runDBT connSource {- transactionSettings -} def) + . (runDBT connSource defaultTransactionSettings) . (\m' -> fst <$> (runStateT m' $ TestEnvSt $ UTCTime (ModifiedJulianDay 0) 0)) . unTestEnv $ m @@ -77,7 +77,8 @@ [] -> defaultConnString (cs:_) -> cs - let connSettings = def { csConnInfo = T.pack connString } + let connSettings = defaultConnectionSettings + { csConnInfo = T.pack connString } ConnectionSource connSource = simpleSource connSettings withSimpleStdOutLogger $ \logger -> @@ -120,14 +121,17 @@ createTables :: TestEnv () createTables = do - migrateDatabase {- options -} def {- extensions -} [] {- domains -} [] + migrateDatabase defaultExtrasOptions + {- extensions -} [] {- composites -} [] {- domains -} [] tables migrations - checkDatabase {- options -} def {- domains -} [] tables + checkDatabase defaultExtrasOptions + {- composites -} [] {- domains -} [] + tables dropTables :: TestEnv () dropTables = do - migrateDatabase {- options -} def {- extensions -} [] {- domains -} [] - {- tables -} [] + migrateDatabase defaultExtrasOptions + {- extensions -} [] {- composites -} [] {- domains -} [] {- tables -} [] [ dropTableMigration jobsTable , dropTableMigration consumersTable ]