consumers 2.1.0.0 → 2.1.1.0
raw patch · 5 files changed
+41/−12 lines, 5 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- CHANGELOG.md +4/−0
- consumers.cabal +6/−4
- example/Example.hs +27/−4
- src/Database/PostgreSQL/Consumers/Consumer.hs +1/−1
- test/Test.hs +3/−3
CHANGELOG.md view
@@ -1,3 +1,7 @@+# consumers-2.1.1.0 (2018-03-18)++* GHC 8.4.1 support.+ # consumers-2.1.0.0 (2017-09-18) * Added a `MonadTime` constraint to `runConsumer`. The `currentTime`
consumers.cabal view
@@ -1,5 +1,5 @@ name: consumers-version: 2.1.0.0+version: 2.1.1.0 synopsis: Concurrent PostgreSQL data consumers description: Library for setting up concurrent consumers of data@@ -12,12 +12,14 @@ extra-source-files: CHANGELOG.md, README.md author: Scrive AB maintainer: Andrzej Rybczak <andrzej@rybczak.net>,- Jonathan Jouty <jonathan@scrive.com>+ Jonathan Jouty <jonathan@scrive.com>,+ Mikhail Glushenkov <mikhail@scrive.com> copyright: Scrive AB category: Concurrency, Database build-type: Simple cabal-version: >=1.10-tested-with: GHC == 7.10.3, GHC == 8.0.2, GHC == 8.2.1+tested-with: GHC == 7.10.3, GHC == 8.0.2, GHC == 8.2.2,+ GHC == 8.4.1 Source-repository head Type: git@@ -34,7 +36,7 @@ containers, exceptions, extra,- hpqtypes >=1.5,+ hpqtypes >= 1.5, lifted-base, lifted-threads, log-base >= 0.7,
example/Example.hs view
@@ -24,6 +24,8 @@ import qualified Data.Text as T +-- | Main application monad. See the 'log-base' and the 'hpqtypes'+-- packages for documentation on 'DBT' and 'LogT'. type AppM a = DBT (LogT IO) a main :: IO ()@@ -36,19 +38,29 @@ let connSettings = def { csConnInfo = T.pack connString } ConnectionSource connSource = simpleSource connSettings + -- Monad stack initialisation. withSimpleStdOutLogger $ \logger -> runLogT "consumers-example" logger $ runDBT connSource {- transactionSettings -} def $ do++ -- Initialise. createTables++ -- Create a consumer, put ten jobs into its queue, and wait+ -- for it to finish. 'runConsumer' returns a finaliser that is+ -- invoked by 'finalize' after the 'putJob' loop. finalize (localDomain "process" $- runConsumer consumerConfig connSource) $ do+ runConsumer consumerConfig connSource) $ forM_ [(0::Int)..10] $ \_ -> do putJob liftIO $ threadDelay (1 * 1000000) -- 1 sec++ -- Clean up. dropTables where + -- How to connect to DB. defaultConnString = "postgresql://postgres@localhost/travis_ci_test" @@ -59,17 +71,19 @@ createTables :: AppM () createTables = do- migrateDatabase {- options -} [] {- extensions -} [] {- domains -} []+ migrateDatabase {- options -} def {- extensions -} [] {- domains -} [] tables migrations- checkDatabase {- domains -} [] tables+ checkDatabase {- options -} def {- domains -} [] tables dropTables :: AppM () dropTables = do- migrateDatabase {- options -} [] {- extensions -} [] {- domains -} []+ migrateDatabase {- options -} def {- extensions -} [] {- domains -} [] {- tables -} [] [ dropTableMigration jobsTable , dropTableMigration consumersTable ] + -- Configuration of a consumer. See+ -- 'Database.PostgreSQL.Consumers.Config.ConsumerConfig'. consumerConfig = ConsumerConfig { ccJobsTable = "consumers_example_jobs" , ccConsumersTable = "consumers_example_consumers"@@ -83,6 +97,7 @@ , ccOnException = handleException } + -- Add a job to the consumer's queue. putJob :: AppM () putJob = localDomain "put" $ do logInfo_ "putJob"@@ -91,11 +106,15 @@ <> "VALUES (NOW(), NULL, NULL, 0, 'hello')" commit + -- Invoked when a job is ready to be processed. processJob :: (Int64, T.Text) -> AppM Result processJob (_idx, msg) = do logInfo_ msg return (Ok Remove) + -- Invoked when 'processJob' throws an exception. Can handle+ -- failure in different ways, such as: remove the job from the+ -- queue, mark it as processed, or schedule it for rerun. handleException :: SomeException -> (Int64, T.Text) -> AppM Action handleException exc (idx, _msg) = do logAttention_ $@@ -103,6 +122,8 @@ return . RerunAfter $ imicroseconds 500000 +-- | Table where jobs are stored. See+-- 'Database.PostgreSQL.Consumers.Config.ConsumerConfig'. jobsTable :: Table jobsTable = tblTable@@ -132,6 +153,8 @@ ] } +-- | Table where registered consumers are stored. See+-- 'Database.PostgreSQL.Consumers.Config.ConsumerConfig'. consumersTable :: Table consumersTable = tblTable
src/Database/PostgreSQL/Consumers/Consumer.hs view
@@ -28,7 +28,7 @@ instance Show ConsumerID where showsPrec p (ConsumerID n) = showsPrec p n- + -- | Register consumer in the consumers table, -- so that it can reserve jobs using acquired ID. registerConsumer
test/Test.hs view
@@ -119,13 +119,13 @@ createTables :: TestEnv () createTables = do- migrateDatabase {- options -} [] {- extensions -} [] {- domains -} []+ migrateDatabase {- options -} def {- extensions -} [] {- domains -} [] tables migrations- checkDatabase {- domains -} [] tables+ checkDatabase {- options -} def {- domains -} [] tables dropTables :: TestEnv () dropTables = do- migrateDatabase {- options -} [] {- extensions -} [] {- domains -} []+ migrateDatabase {- options -} def {- extensions -} [] {- domains -} [] {- tables -} [] [ dropTableMigration jobsTable , dropTableMigration consumersTable ]