powerqueue 0.1.0.0 → 0.2.0.0

raw patch · 2 files changed

+41/−7 lines, 2 filesdep +timespanPVP ok

version bump matches the API change (PVP)

Dependencies added: timespan

API changes (from Hackage documentation)

Files

• powerqueue.cabal +11/−1
• src/Data/PowerQueue.hs +30/−6

@@ -1,7 +1,16 @@ name:                powerqueue-version:             0.1.0.0+version:             0.2.0.0 synopsis:            A flexible job queue with exchangeable backends description:         A flexible job queue with exchangeable backends+                     .+                     Current Queue (Persistence) Backends:+                     .+                     * <https://hackage.haskell.org/package/powerqueue-levelmem LevelDB>+                     * <https://hackage.haskell.org/package/powerqueue-sqs AWS SQS>+                     .+                     Current Worker Backends:+                     .+                     * <https://hackage.haskell.org/package/powerqueue-distributed TCP-Network Distributed> homepage:            https://github.com/agrafix/powerqueue#readme license:             BSD3 license-file:        LICENSE@@ -19,6 +28,7 @@                 base >= 4.7 && < 5               , contravariant >= 1.4               , async >= 2.1+              , timespan >= 0.3   default-language:    Haskell2010  test-suite powerqueue-test

@@ -4,9 +4,9 @@ {-# LANGUAGE ScopedTypeVariables #-} module Data.PowerQueue     ( -- * Worker descriptions-      QueueWorker, newQueueWorker, JobResult(..)+      QueueWorker(..), newQueueWorker, JobResult(..)       -- * Queue control-    , Queue, newQueue, mapQueue, enqueueJob+    , Queue, newQueue, mapQueue, enqueueJob, getQueueBackend, getQueueWorker       -- * (persistent) queue backends     , QueueBackend(..), mapBackend, basicChanBackend       -- * execution strategies@@ -23,6 +23,7 @@ import Data.Functor.Contravariant import Data.Functor.Contravariant.Divisible import Data.IORef+import Data.Time.TimeSpan import qualified Control.Concurrent.Chan as C  -- | Result of the job@@ -46,6 +47,10 @@       -- ^ mark a single job as confirmed     , qb_rollback :: tx -> m ()       -- ^ mark a single job as failed+    , qb_reportProgress :: tx -> m ()+      -- ^ report progress on a job+    , qb_progressReportInterval :: !TimeSpan+      -- ^ how often should progress be reported     }  -- | A very basic in memory backend using only data structures from the base library.@@ -79,16 +84,20 @@                       case oldVal of                         Just j -> C.writeChan jobChannel j                         Nothing -> pure () -- should not really happen ...+           , qb_reportProgress = const $ pure ()+           , qb_progressReportInterval = hours 1            }  mapBackend :: (a -> b) -> (b -> a) -> QueueBackend a -> QueueBackend b-mapBackend f g (QueueBackend qlift qenq qdeq qconf qroll) =+mapBackend f g (QueueBackend qlift qenq qdeq qconf qroll rp intv) =     QueueBackend     { qb_lift = qlift     , qb_enqueue = qenq . g     , qb_dequeue = fmap (second f) qdeq     , qb_confirm = qconf     , qb_rollback = qroll+    , qb_reportProgress = rp+    , qb_progressReportInterval = intv     }  data QueueWorker j@@ -130,6 +139,12 @@     , q_backend :: !(QueueBackend j)     } +getQueueWorker :: Queue j -> QueueWorker j+getQueueWorker = q_worker++getQueueBackend :: Queue j -> QueueBackend j+getQueueBackend = q_backend+ mapQueue :: (a -> b) -> (b -> a) -> Queue a -> Queue b mapQueue f g q =     Queue@@ -158,11 +173,20 @@  workStepInternal :: QueueBackend j -> QueueWorker j -> IO () workStepInternal QueueBackend{..} QueueWorker{..} =-    let acquire = qb_lift qb_dequeue-        onError (txId, _) = qb_lift (qb_rollback txId)-        go (txId, job) =+    let acquire =+            do v <- qb_lift qb_dequeue+               pReport <-+                   async $ forever $+                   do sleepTS qb_progressReportInterval+                      qb_lift $ qb_reportProgress (fst v)+               pure (pReport, v)+        onError (pg, (txId, _)) =+            do cancel pg+               qb_lift (qb_rollback txId)+        go (pg, (txId, job)) =             do execRes <-                    try $ qw_execute job+               cancel pg                case execRes of                  Left (_ :: SomeException) ->                      qb_lift (qb_rollback txId)