faktory 1.0.3.1 → 1.1.0.0

raw patch · 13 files changed

+150/−27 lines, 13 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +14/−1
• README.lhs +3/−2
• examples/consumer/Main.hs +1/−1
• faktory.cabal +3/−1
• library/Faktory/Ent/Batch.hs +2/−1
• library/Faktory/Ent/Batch/Status.hs +48/−0
• library/Faktory/Ent/Tracking.hs +5/−0
• library/Faktory/Job.hs +1/−0
• library/Faktory/Job/Custom.hs +7/−0
• library/Faktory/Worker.hs +5/−4
• tests/Faktory/Ent/BatchSpec.hs +21/−0
• tests/Faktory/Ent/TrackingSpec.hs +5/−5
• tests/Faktory/Test.hs +35/−12

@@ -1,6 +1,19 @@-## [*Unreleased*](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.3.1...main)+## [*Unreleased*](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.0.0...main)  None++## [v1.1.0.0](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.3.1...v1.1.0.0)++- Pass value of type `Job arg` (not `arg`) to run-worker loops++  This will give consumer loops access to details like `jobJid` and+  `jobOptions`, so they can (for example) call `TRACK SET`.++  Call `jobArg` to get back what you were getting before this change.++- Support `BATCH STATUS`+- Add `tracked` Job Option+- Deprecate `trackPerform` (use `perform (options <> tracked)` instead)  ## [v1.0.3.1](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.3.0...v1.0.3.1) 

@@ -79,9 +79,10 @@ ### Worker  ```haskell-workerMain = runWorkerEnv $ \job ->+workerMain = runWorkerEnv $ \job -> do   -- Process your Job here-  putStrLn $ myJobMessage job+  putStrLn $ jobJid job+  putStrLn $ myJobMessage $ jobArg job    -- If any exception is thrown, the job will be marked as Failed in Faktory   -- and retried. Note: you will not otherwise hear about any such exceptions,

@@ -16,7 +16,7 @@ main = do   putStrLn "Starting consumer loop"   runWorkerEnv $ \job -> do-    let message = jobMessage job+    let message = jobMessage $ jobArg job      if message == "BOOM"       then throwString "Producer exception: BOOM"

@@ -1,6 +1,6 @@ cabal-version:   1.18 name:            faktory-version:         1.0.3.1+version:         1.1.0.0 license:         MIT license-file:    LICENSE copyright:       2018 Freckle Education@@ -54,6 +54,7 @@         Faktory.Client         Faktory.Connection         Faktory.Ent.Batch+        Faktory.Ent.Batch.Status         Faktory.Ent.Tracking         Faktory.Job         Faktory.Job.Custom@@ -203,6 +204,7 @@         base >=4.13 && <5,         faktory -any,         hspec >=2.7.8,+        mtl >=2.2.2,         time >=1.9.3      if impl(ghc >=8.10)

@@ -38,6 +38,7 @@   , batchPerform    -- * Low-level+  , BatchId(..)   , CustomBatchId(..)   , newBatch   , commitBatch@@ -62,7 +63,7 @@   deriving newtype (Functor, Applicative, Monad, MonadIO, MonadReader BatchId)  newtype BatchId = BatchId Text-  deriving newtype ToJSON+  deriving newtype (FromJSON, ToJSON)  data BatchOptions arg = BatchOptions   { boDescription :: Maybe (Last Text)

@@ -0,0 +1,48 @@+module Faktory.Ent.Batch.Status+  ( jobBatchId+  , BatchStatus(..)+  , batchStatus+  ) where++import Faktory.Prelude++import Control.Error.Util (hush)+import Data.Aeson+import Data.ByteString.Lazy as BSL+import Data.Text.Encoding (encodeUtf8)+import Data.Time (UTCTime)+import Faktory.Client+import Faktory.Ent.Batch+import Faktory.Job (Job, jobOptions)+import Faktory.Job.Custom+import Faktory.JobOptions (JobOptions(..))+import Faktory.Producer+import GHC.Generics++data BatchStatus = BatchStatus+  { bid :: BatchId+  , total :: Int+  , pending :: Int+  , failed :: Int+  , created_at :: UTCTime+  , description :: Text+  }+  deriving stock Generic+  deriving anyclass FromJSON++newtype ReadCustomBatchId = ReadCustomBatchId+  { _bid :: BatchId+  }+  deriving stock Generic+  deriving anyclass FromJSON++jobBatchId :: Job arg -> Maybe BatchId+jobBatchId job = do+  custom <- joCustom $ jobOptions job+  _bid <$> hush (fromCustom custom)++batchStatus :: Producer -> BatchId -> IO (Either String (Maybe BatchStatus))+batchStatus producer (BatchId bid) = commandJSON+  (producerClient producer)+  "BATCH STATUS"+  [BSL.fromStrict $ encodeUtf8 bid]

@@ -4,6 +4,7 @@ -- module Faktory.Ent.Tracking   ( CustomTrack(..)+  , tracked   , trackPerform    , JobDetails(..)@@ -36,6 +37,9 @@   deriving stock Generic   deriving anyclass ToJSON +tracked :: JobOptions+tracked = custom (CustomTrack 1)+ -- | 'perform', but adding @{ custom: { track: 1 } }@ -- -- Equivalent to:@@ -47,6 +51,7 @@ trackPerform   :: (HasCallStack, ToJSON arg) => JobOptions -> Producer -> arg -> IO JobId trackPerform options = perform (options <> custom (CustomTrack 1))+{-# DEPRECATED trackPerform "Use ‘perform (options <> tracked)’ instead" #-}  data JobDetails = JobDetails   { jdJid :: JobId

@@ -14,6 +14,7 @@   , newJob   , jobJid   , jobArg+  , jobOptions   ) where  import Faktory.Prelude

@@ -6,6 +6,7 @@ module Faktory.Job.Custom     ( Custom     , toCustom+    , fromCustom     ) where  import Faktory.Prelude@@ -19,6 +20,12 @@  toCustom :: ToJSON a => a -> Custom toCustom = Custom . toJSON++-- | Read a 'Custom' value to a type using 'FromJSON'+fromCustom :: FromJSON a => Custom -> Either String a+fromCustom (Custom v) = case fromJSON v of+  Error e -> Left e+  Success a -> Right a  instance Semigroup Custom where   (Custom (Object a)) <> (Custom (Object b)) =

@@ -7,6 +7,7 @@   ( WorkerHalt(..)   , runWorker   , runWorkerEnv+  , jobArg   ) where @@ -61,7 +62,7 @@   :: (HasCallStack, FromJSON args)   => Settings   -> WorkerSettings-  -> (args -> IO ())+  -> (Job args -> IO ())   -> IO () runWorker settings workerSettings f = do   workerId <- maybe randomWorkerId pure $ settingsId workerSettings@@ -72,7 +73,7 @@     `catch` (\(_ex :: WorkerHalt) -> pure ())     `finally` (killThread beatThreadId >> closeClient client) -runWorkerEnv :: FromJSON args => (args -> IO ()) -> IO ()+runWorkerEnv :: FromJSON args => (Job args -> IO ()) -> IO () runWorkerEnv f = do   settings <- envSettings   workerSettings <- envWorkerSettings@@ -83,13 +84,13 @@   => Client   -> Settings   -> WorkerSettings-  -> (arg -> IO ())+  -> (Job arg -> IO ())   -> IO () processorLoop client settings workerSettings f = do   let     namespace = connectionInfoNamespace $ settingsConnection settings     processAndAck job = do-      f $ jobArg job+      f job       ackJob client job    emJob <- fetchJob client $ namespaceQueue namespace $ settingsQueue

@@ -5,7 +5,9 @@ import Faktory.Test  import Control.Concurrent (threadDelay)+import Control.Monad.Reader import Faktory.Ent.Batch+import qualified Faktory.Ent.Batch.Status as BatchStatus  spec :: Spec spec = do@@ -74,3 +76,22 @@         liftIO $ threadDelay 500000        jobs `shouldMatchList` ["a", "b", "c", "d", "HALT"]++    it "supports BATCH STATUS" $ do+      batchId <- withWorker id $ withProducer $ \producer -> do+        c <- buildJob @Text mempty producer "c"+        d <- buildJob @Text mempty producer "d"+        let options = description "foo" <> complete c <> success d+        batchId <- runBatch options producer $ do+          void $ batchPerform @Text mempty producer "a"+          void $ batchPerform @Text mempty producer "b"+          ask+        batchId <$ liftIO (threadDelay 500000)++      emStatus <- bracket newProducerEnv closeProducer+        $ \producer -> BatchStatus.batchStatus producer batchId++      fmap (fmap BatchStatus.description) emStatus `shouldBe` Right (Just "foo")+      fmap (fmap BatchStatus.total) emStatus `shouldBe` Right (Just 2)+      fmap (fmap BatchStatus.pending) emStatus `shouldBe` Right (Just 0)+      fmap (fmap BatchStatus.failed) emStatus `shouldBe` Right (Just 0)

@@ -11,13 +11,13 @@  spec :: Spec spec = do-  describe "trackPerform" $ do-    it "enqueues with TRACK so we can get details" $ do+  describe "tracked" $ do+    it "adds custom option so we can use TRACK GET later" $ do       var <- newMVar []       void $ workerTestCase $ \producer -> do-        a <- trackPerform @Text mempty producer "a"+        a <- perform @Text tracked producer "a"         modifyMVar_ var $ pure . (<> [a])-        b <- trackPerform @Text mempty producer "BOOM"+        b <- perform @Text tracked producer "BOOM"         modifyMVar_ var $ pure . (<> [b])        enqueuedJobIds <- readMVar var@@ -40,7 +40,7 @@     it "updates Job Details" $ do       var <- newMVar []       void $ workerTestCase $ \producer -> do-        a <- trackPerform @Text mempty producer "a"+        a <- perform @Text tracked producer "a"         modifyMVar_ var $ pure . (<> [a])        enqueuedJobIds <- readMVar var

@@ -2,6 +2,12 @@   ( module X   , workerTestCase   , workerTestCaseWith++  -- * Lower-level+  , withProducer+  , withWorker+  , startWorker+  , haltWorker   ) where @@ -26,21 +32,38 @@   -> (Producer -> IO ())   -> IO [Text] workerTestCaseWith editSettings run = do-  bracket newProducerEnv closeProducer $ \producer -> do-    void $ flush producer+  a <- startWorker editSettings+  withProducer run+  haltWorker a +withProducer :: (Producer -> IO a) -> IO a+withProducer f = bracket newProducerEnv closeProducer f++withWorker+  :: HasCallStack => (WorkerSettings -> WorkerSettings) -> IO a -> IO a+withWorker editSettings f = do+  a <- startWorker editSettings+  result <- f+  result <$ haltWorker a++startWorker+  :: HasCallStack => (WorkerSettings -> WorkerSettings) -> IO (Async [Text])+startWorker editSettings = do+  withProducer $ void . flush   settings <- envSettings   workerSettings <- editSettings <$> envWorkerSettings+  async $ do+    processedJobs <- newMVar [] -  processedJobs <- newMVar []-  a <- async $ runWorker settings workerSettings $ \job -> do-    modifyMVar_ processedJobs $ pure . (job :)-    when (job == "BOOM") $ throw $ userError "BOOM"-    when (job == "HALT") $ throw WorkerHalt+    runWorker settings workerSettings $ \faktoryJob -> do+      let job = jobArg faktoryJob+      modifyMVar_ processedJobs $ pure . (job :)+      when (job == "BOOM") $ throw $ userError "BOOM"+      when (job == "HALT") $ throw WorkerHalt -  bracket newProducerEnv closeProducer $ \producer -> do-    run producer-    void $ perform @Text mempty producer "HALT"+    readMVar processedJobs -  void $ wait a-  readMVar processedJobs+haltWorker :: Async a -> IO a+haltWorker a = do+  withProducer $ \producer -> void $ perform @Text mempty producer "HALT"+  wait a