faktory 1.0.2.3 → 1.0.3.0

raw patch · 8 files changed

+346/−106 lines, 8 filesdep +errorsPVP ok

version bump matches the API change (PVP)

Dependencies added: errors

API changes (from Hackage documentation)

Files

• CHANGELOG.md +5/−1
• faktory.cabal +8/−2
• library/Faktory/Ent/Tracking.hs +95/−0
• library/Faktory/JobState.hs +53/−0
• tests/Faktory/Ent/BatchSpec.hs +76/−0
• tests/Faktory/Ent/TrackingSpec.hs +62/−0
• tests/Faktory/Test.hs +46/−0
• tests/FaktorySpec.hs +1/−103

@@ -1,6 +1,10 @@-## [*Unreleased*](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.2.3...main)+## [*Unreleased*](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.3.0...main)  None++## [v1.0.3.0](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.2.3...v1.0.3.0)++- Support for `TRACK` (Enterprise only)  ## [v1.0.2.3](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.2.2...v1.0.2.3) 

@@ -1,6 +1,6 @@ cabal-version:   1.18 name:            faktory-version:         1.0.2.3+version:         1.0.3.0 license:         MIT license-file:    LICENSE copyright:       2018 Freckle Education@@ -54,9 +54,11 @@         Faktory.Client         Faktory.Connection         Faktory.Ent.Batch+        Faktory.Ent.Tracking         Faktory.Job         Faktory.Job.Custom         Faktory.JobOptions+        Faktory.JobState         Faktory.Prelude         Faktory.Producer         Faktory.Protocol@@ -88,6 +90,7 @@         bytestring >=0.10.12.0,         connection >=0.3.1,         cryptonite >=0.27,+        errors >=2.3.0,         megaparsec >=9.0.1,         memory >=0.15.0,         mtl >=2.2.2,@@ -171,7 +174,10 @@     hs-source-dirs:     tests     other-modules:         Faktory.ConnectionSpec+        Faktory.Ent.BatchSpec+        Faktory.Ent.TrackingSpec         Faktory.JobOptionsSpec+        Faktory.Test         FaktorySpec         Paths_faktory @@ -206,7 +212,7 @@ test-suite readme     type:               exitcode-stdio-1.0     main-is:            README.lhs-    hs-source-dirs:     ./.+    hs-source-dirs:     ./     other-modules:      Paths_faktory     default-language:   Haskell2010     default-extensions:

@@ -0,0 +1,95 @@+-- | Support for the @TRACK@ command (Enterprise only)+--+-- <https://github.com/contribsys/faktory/wiki/Ent-Tracking>+--+module Faktory.Ent.Tracking+  ( CustomTrack(..)+  , trackPerform++  , JobDetails(..)+  , JobState(..)+  , trackGet+  , trackGetHush++  , SetJobDetails(..)+  , trackSet+  ) where++import Faktory.Prelude++import Control.Error.Util (hush)+import Data.Aeson+import Data.Aeson.Casing+import qualified Data.ByteString.Lazy.Char8 as BSL8+import Data.Maybe (fromMaybe)+import Data.Time (UTCTime)+import Faktory.Client (commandJSON, commandOK)+import Faktory.Job (JobId, JobOptions, custom, perform)+import Faktory.JobState (JobState(..))+import Faktory.Producer+import GHC.Generics (Generic)+import GHC.Stack (HasCallStack)++newtype CustomTrack = CustomTrack+  { track :: Int+  }+  deriving stock Generic+  deriving anyclass ToJSON++-- | 'perform', but adding @{ custom: { track: 1 } }@+--+-- Equivalent to:+--+-- @+-- 'perform' ('custom' $ 'CustomTrack' 1)+-- @+--+trackPerform+  :: (HasCallStack, ToJSON arg) => JobOptions -> Producer -> arg -> IO JobId+trackPerform options = perform (options <> custom (CustomTrack 1))++data JobDetails = JobDetails+  { jdJid :: JobId+  , jdPercent :: Maybe Int+  , jdDesc :: Maybe Text+  , jdState :: JobState+  , jdUpdatedAt :: Maybe UTCTime+  }+  deriving stock Generic++instance FromJSON JobDetails where+  parseJSON = genericParseJSON $ aesonPrefix snakeCase++unknownJobDetails :: JobId -> JobDetails+unknownJobDetails jid = JobDetails+  { jdJid = jid+  , jdPercent = Nothing+  , jdDesc = Nothing+  , jdState = JobStateUnknown+  , jdUpdatedAt = Nothing+  }++data SetJobDetails = SetJobDetails+  { sjdJid :: JobId+  , sjdPercent :: Maybe Int+  , sjdDesc :: Maybe Text+  , sjdReserveUntil :: Maybe UTCTime+  }+  deriving stock Generic++instance ToJSON SetJobDetails where+  toJSON = genericToJSON $ aesonPrefix snakeCase+  toEncoding = genericToEncoding $ aesonPrefix snakeCase++trackGet :: Producer -> JobId -> IO (Either String (Maybe JobDetails))+trackGet producer jid =+  commandJSON (producerClient producer) "TRACK GET" [BSL8.pack jid]++-- | 'trackGet' but mask any failures to 'JobStateUnknown'+trackGetHush :: Producer -> JobId -> IO JobDetails+trackGetHush producer jid =+  fromMaybe (unknownJobDetails jid) . join . hush <$> trackGet producer jid++trackSet :: HasCallStack => Producer -> SetJobDetails -> IO ()+trackSet producer details =+  commandOK (producerClient producer) "TRACK SET" [encode details]

@@ -0,0 +1,53 @@+module Faktory.JobState+  ( JobState(..)+  ) where++import Faktory.Prelude++import Control.Arrow ((&&&))+import Control.Error.Util (note)+import Data.Aeson+import Data.HashMap.Strict (HashMap)+import qualified Data.HashMap.Strict as HashMap+import qualified Data.Text as T++data JobState+  = JobStateUnknown+  | JobStateEnqueued+  | JobStateWorking+  | JobStateSuccess+  | JobStateFailed+  | JobStateDead+  deriving stock (Eq, Show, Bounded, Enum)++instance ToJSON JobState where+  toJSON = toJSON . jobStateToText+  toEncoding = toEncoding . jobStateToText++instance FromJSON JobState where+  parseJSON = withText "JobState" $ either fail pure . jobStateFromText++jobStateToText :: JobState -> Text+jobStateToText = \case+  JobStateUnknown -> "unknown"+  JobStateEnqueued -> "enqueued"+  JobStateWorking -> "working"+  JobStateSuccess -> "success"+  JobStateFailed -> "failed"+  JobStateDead -> "dead"++jobStateFromText :: Text -> Either String JobState+jobStateFromText x =+  note+      (unpack+      $ "Invalid JobState: "+      <> x+      <> ", must be one of "+      <> T.intercalate ", " (HashMap.keys jobStateMap)+      )+    $ HashMap.lookup x jobStateMap++jobStateMap :: HashMap Text JobState+jobStateMap =+  HashMap.fromList $ map (jobStateToText &&& id) [minBound .. maxBound]+{-# NOINLINE jobStateMap #-}

@@ -0,0 +1,76 @@+module Faktory.Ent.BatchSpec+  ( spec+  ) where++import Faktory.Test++import Control.Concurrent (threadDelay)+import Faktory.Ent.Batch++spec :: Spec+spec = do+  describe "runBatch" $ do+    it "runs a success job if all in-batch jobs succeed" $ do+      jobs <- workerTestCase $ \producer -> do+        c <- buildJob @Text mempty producer "c"+        void $ runBatch (success c) producer $ do+          void $ batchPerform @Text mempty producer "a"+          void $ batchPerform @Text mempty producer "b"+        -- Give a little time for Faktory to fire the callback+        liftIO $ threadDelay 500000++      jobs `shouldMatchList` ["a", "b", "c", "HALT"]++    it "does not run a success job if all jobs don't succeed" $ do+      jobs <- workerTestCase $ \producer -> do+        c <- buildJob @Text mempty producer "c"+        void $ runBatch (success c) producer $ do+          void $ batchPerform @Text mempty producer "BOOM"+          void $ batchPerform @Text mempty producer "b"+        liftIO $ threadDelay 500000++      jobs `shouldMatchList` ["BOOM", "b", "HALT"]++    it "runs a job on complete" $ do+      jobs <- workerTestCase $ \producer -> do+        c <- buildJob @Text mempty producer "c"+        void $ runBatch (complete c) producer $ do+          void $ batchPerform @Text mempty producer "a"+          void $ batchPerform @Text mempty producer "b"+        liftIO $ threadDelay 500000++      jobs `shouldMatchList` ["a", "b", "c", "HALT"]++    it "runs a job on complete, even if in-batch jobs fail" $ do+      jobs <- workerTestCase $ \producer -> do+        c <- buildJob @Text mempty producer "c"+        void $ runBatch (complete c) producer $ do+          void $ batchPerform @Text mempty producer "BOOM"+          void $ batchPerform @Text mempty producer "b"+        liftIO $ threadDelay 500000++      jobs `shouldMatchList` ["BOOM", "b", "c", "HALT"]++    it "combines duplicate options in last-wins fashion" $ do+      jobs <- workerTestCase $ \producer -> do+        c <- buildJob @Text mempty producer "c"+        d <- buildJob @Text mempty producer "d"+        let options = description "foo" <> success c <> success d+        void $ runBatch options producer $ do+          void $ batchPerform @Text mempty producer "a"+          void $ batchPerform @Text mempty producer "b"+        liftIO $ threadDelay 500000++      jobs `shouldMatchList` ["a", "b", "d", "HALT"]++    it "runs success and complete if all Jobs were successful" $ do+      jobs <- workerTestCase $ \producer -> do+        c <- buildJob @Text mempty producer "c"+        d <- buildJob @Text mempty producer "d"+        let options = description "foo" <> complete c <> success d+        void $ runBatch options producer $ do+          void $ batchPerform @Text mempty producer "a"+          void $ batchPerform @Text mempty producer "b"+        liftIO $ threadDelay 500000++      jobs `shouldMatchList` ["a", "b", "c", "d", "HALT"]

@@ -0,0 +1,62 @@+{-# OPTIONS_GHC -Wno-incomplete-uni-patterns #-}++module Faktory.Ent.TrackingSpec+  ( spec+  ) where++import Faktory.Test++import Control.Concurrent.MVar+import Faktory.Ent.Tracking++spec :: Spec+spec = do+  describe "trackPerform" $ do+    it "enqueues with TRACK so we can get details" $ do+      var <- newMVar []+      void $ workerTestCase $ \producer -> do+        a <- trackPerform @Text mempty producer "a"+        modifyMVar_ var $ pure . (<> [a])+        b <- trackPerform @Text mempty producer "BOOM"+        modifyMVar_ var $ pure . (<> [b])++      enqueuedJobIds <- readMVar var+      enqueuedJobIds `shouldSatisfy` (== 2) . length++      let [aJid, bJid] = enqueuedJobIds++      (aDetails, bDetails, cDetails) <-+        bracket newProducerEnv closeProducer $ \producer ->+          (,,)+            <$> trackGetHush producer aJid+            <*> trackGetHush producer bJid+            <*> trackGetHush producer "madeUp"++      jdState aDetails `shouldBe` JobStateSuccess+      jdState bDetails `shouldBe` JobStateFailed+      jdState cDetails `shouldBe` JobStateUnknown++  describe "trackSet" $ do+    it "updates Job Details" $ do+      var <- newMVar []+      void $ workerTestCase $ \producer -> do+        a <- trackPerform @Text mempty producer "a"+        modifyMVar_ var $ pure . (<> [a])++      enqueuedJobIds <- readMVar var+      enqueuedJobIds `shouldSatisfy` (== 1) . length++      let [aJid] = enqueuedJobIds++      aDetails <- bracket newProducerEnv closeProducer $ \producer -> do+        trackSet producer $ SetJobDetails+          { sjdJid = aJid+          , sjdPercent = Just 100+          , sjdDesc = Just "Updated"+          , sjdReserveUntil = Nothing+          }++        trackGetHush producer aJid++      jdPercent aDetails `shouldBe` Just 100+      jdDesc aDetails `shouldBe` Just "Updated"

@@ -0,0 +1,46 @@+module Faktory.Test+  ( module X+  , workerTestCase+  , workerTestCaseWith+  )+where++import Faktory.Prelude as X++import Control.Monad.IO.Class as X (MonadIO(..))+import Faktory.Job as X+import Faktory.Producer as X+import Test.Hspec as X++import Control.Concurrent.Async+import Control.Concurrent.MVar+import Faktory.Settings+import Faktory.Worker++workerTestCase :: HasCallStack => (Producer -> IO ()) -> IO [Text]+workerTestCase = workerTestCaseWith id++workerTestCaseWith+  :: HasCallStack+  => (WorkerSettings -> WorkerSettings)+  -> (Producer -> IO ())+  -> IO [Text]+workerTestCaseWith editSettings run = do+  bracket newProducerEnv closeProducer $ \producer -> do+    void $ flush producer++  settings <- envSettings+  workerSettings <- editSettings <$> envWorkerSettings++  processedJobs <- newMVar []+  a <- async $ runWorker settings workerSettings $ \job -> do+    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"++  void $ wait a+  readMVar processedJobs

@@ -2,19 +2,11 @@   ( spec   ) where -import Faktory.Prelude+import Faktory.Test  import Control.Concurrent (threadDelay)-import Control.Concurrent.Async-import Control.Concurrent.MVar-import Control.Monad.IO.Class (liftIO) import Data.Time (getCurrentTime)-import Faktory.Ent.Batch-import Faktory.Job-import Faktory.Producer import Faktory.Settings-import Faktory.Worker-import Test.Hspec  spec :: Spec spec = describe "Faktory" $ do@@ -59,97 +51,3 @@       threadDelay $ 2 * 1000000 + 250000      jobs `shouldMatchList` ["HALT"]--  context "BATCH" $ do-    it "runs a success job if all in-batch jobs succeed" $ do-      jobs <- workerTestCase $ \producer -> do-        c <- buildJob @Text mempty producer "c"-        void $ runBatch (success c) producer $ do-          void $ batchPerform @Text mempty producer "a"-          void $ batchPerform @Text mempty producer "b"-        -- Give a little time for Faktory to fire the callback-        liftIO $ threadDelay 500000--      jobs `shouldMatchList` ["a", "b", "c", "HALT"]--    it "does not run a success job if all jobs don't succeed" $ do-      jobs <- workerTestCase $ \producer -> do-        c <- buildJob @Text mempty producer "c"-        void $ runBatch (success c) producer $ do-          void $ batchPerform @Text mempty producer "BOOM"-          void $ batchPerform @Text mempty producer "b"-        liftIO $ threadDelay 500000--      jobs `shouldMatchList` ["BOOM", "b", "HALT"]--    it "runs a job on complete" $ do-      jobs <- workerTestCase $ \producer -> do-        c <- buildJob @Text mempty producer "c"-        void $ runBatch (complete c) producer $ do-          void $ batchPerform @Text mempty producer "a"-          void $ batchPerform @Text mempty producer "b"-        liftIO $ threadDelay 500000--      jobs `shouldMatchList` ["a", "b", "c", "HALT"]--    it "runs a job on complete, even if in-batch jobs fail" $ do-      jobs <- workerTestCase $ \producer -> do-        c <- buildJob @Text mempty producer "c"-        void $ runBatch (complete c) producer $ do-          void $ batchPerform @Text mempty producer "BOOM"-          void $ batchPerform @Text mempty producer "b"-        liftIO $ threadDelay 500000--      jobs `shouldMatchList` ["BOOM", "b", "c", "HALT"]--    it "combines duplicate options in last-wins fashion" $ do-      jobs <- workerTestCase $ \producer -> do-        c <- buildJob @Text mempty producer "c"-        d <- buildJob @Text mempty producer "d"-        let options = description "foo" <> success c <> success d-        void $ runBatch options producer $ do-          void $ batchPerform @Text mempty producer "a"-          void $ batchPerform @Text mempty producer "b"-        liftIO $ threadDelay 500000--      jobs `shouldMatchList` ["a", "b", "d", "HALT"]--    it "runs success and complete if all Jobs were successful" $ do-      jobs <- workerTestCase $ \producer -> do-        c <- buildJob @Text mempty producer "c"-        d <- buildJob @Text mempty producer "d"-        let options = description "foo" <> complete c <> success d-        void $ runBatch options producer $ do-          void $ batchPerform @Text mempty producer "a"-          void $ batchPerform @Text mempty producer "b"-        liftIO $ threadDelay 500000--      jobs `shouldMatchList` ["a", "b", "c", "d", "HALT"]--workerTestCase :: HasCallStack => (Producer -> IO ()) -> IO [Text]-workerTestCase = workerTestCaseWith id--workerTestCaseWith-  :: HasCallStack-  => (WorkerSettings -> WorkerSettings)-  -> (Producer -> IO ())-  -> IO [Text]-workerTestCaseWith editSettings run = do-  bracket newProducerEnv closeProducer $ \producer -> do-    void $ flush producer--  settings <- envSettings-  workerSettings <- editSettings <$> envWorkerSettings--  processedJobs <- newMVar []-  a <- async $ runWorker settings workerSettings $ \job -> do-    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"--  void $ wait a-  readMVar processedJobs