faktory 1.0.1.6 → 1.0.2.0

raw patch · 12 files changed

+543/−127 lines, 12 filesdep +asyncdep +mtldep +semigroupsdep ~basePVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: async, mtl, semigroups, unordered-containers

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• CHANGELOG.md +7/−2
• README.lhs +3/−2
• faktory.cabal +18/−8
• library/Faktory/Client.hs +18/−9
• library/Faktory/Ent/Batch.hs +118/−0
• library/Faktory/Job.hs +46/−68
• library/Faktory/Job/Custom.hs +26/−0
• library/Faktory/JobOptions.hs +96/−0
• library/Faktory/Settings.hs +1/−0
• library/Faktory/Worker.hs +6/−2
• tests/Faktory/JobOptionsSpec.hs +100/−0
• tests/FaktorySpec.hs +104/−36

@@ -1,11 +1,16 @@-## [*Unreleased*](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.1.6...main)+## [*Unreleased*](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.2.0...main)  None +## [v1.0.2.0](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.1.6...v1.0.2.0)++- Partial `BATCH` support (Enterprise only)+- Support for `custom` field in Job payloads+- Lower-level `buildJob` and `commandByteString` functions+ ## [v1.0.1.6](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.1.5...v1.0.1.6)  - Relax dependencies upper bounds-  ## [v1.0.1.5](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.1.4...v1.0.1.5) 

@@ -169,8 +169,9 @@ stack build --pedantic --test ``` -*NOTE*: `FactorySpec` requires a local Faktory server is running, and it will-flush all Jobs from this server as part of running the tests.+- `FactorySpec` requires a local Faktory server is running, and it will flush+  all Jobs from this server as part of running the tests.+- The tests for `BATCH` require testing against an Enterprise Faktory image  --- 

@@ -1,6 +1,6 @@ cabal-version:   1.18 name:            faktory-version:         1.0.1.6+version:         1.0.2.0 license:         MIT license-file:    LICENSE copyright:       2018 Freckle Education@@ -53,7 +53,10 @@     exposed-modules:         Faktory.Client         Faktory.Connection+        Faktory.Ent.Batch         Faktory.Job+        Faktory.Job.Custom+        Faktory.JobOptions         Faktory.Prelude         Faktory.Producer         Faktory.Protocol@@ -81,19 +84,22 @@     build-depends:         aeson >=1.5.5.1 && <1.6,         aeson-casing >=0.2.0.0 && <0.3,-        base >=4.11 && <5,+        base >=4.13 && <5,         bytestring >=0.10.12.0 && <0.12,         connection >=0.3.1 && <0.4,         cryptonite >=0.27 && <0.29,         megaparsec >=9.0.1 && <9.1,         memory >=0.15.0 && <0.16,+        mtl >=2.2.2 && <2.3,         network >=3.1.1.1 && <3.2,         random >=1.1 && <1.3,         safe-exceptions >=0.1.7.1 && <0.2,         scanner >=0.3.1 && <0.4,+        semigroups >=0.19.1 && <0.20,         text >1.2 && <1.4,         time >=1.9.3 && <1.12,-        unix >=2.7.2.2 && <2.8+        unix >=2.7.2.2 && <2.8,+        unordered-containers >=0.2.13.0 && <0.3      if impl(ghc >=8.10)         ghc-options:@@ -121,7 +127,7 @@      build-depends:         aeson >=1.5.5.1 && <1.6,-        base >=4.11 && <5,+        base >=4.13 && <5,         faktory -any,         safe-exceptions >=0.1.7.1 && <0.2 @@ -151,7 +157,7 @@      build-depends:         aeson >=1.5.5.1 && <1.6,-        base >=4.11 && <5,+        base >=4.13 && <5,         faktory -any,         safe-exceptions >=0.1.7.1 && <0.2 @@ -165,6 +171,7 @@     hs-source-dirs:     tests     other-modules:         Faktory.ConnectionSpec+        Faktory.JobOptionsSpec         FaktorySpec         Paths_faktory @@ -185,9 +192,12 @@         -Wno-monomorphism-restriction -Wno-missing-import-lists -rtsopts      build-depends:-        base >=4.11 && <5,+        aeson >=1.5.5.1 && <1.6,+        async >=2.2.2 && <2.3,+        base >=4.13 && <5,         faktory -any,-        hspec >=2.7.8 && <2.8+        hspec >=2.7.8 && <2.8,+        time >=1.9.3 && <1.12      if impl(ghc >=8.10)         ghc-options:@@ -217,7 +227,7 @@      build-depends:         aeson >=1.5.5.1 && <1.6,-        base >=4.11 && <5,+        base >=4.13 && <5,         faktory -any,         markdown-unlit >=0.5.1 && <0.6 

@@ -9,6 +9,7 @@   , command_   , commandOK   , commandJSON+  , commandByteString   ) where  import Faktory.Prelude@@ -120,15 +121,14 @@  -- | Send a command, read and discard the response command_ :: Client -> ByteString -> [ByteString] -> IO ()-command_ Client {..} cmd args = withMVar clientConnection $ \conn -> do-  sendUnsafe clientSettings conn cmd args-  void $ fromRightThrows =<< recvUnsafe clientSettings conn+command_ client cmd args = do+  response <- commandByteString client cmd args+  void $ fromRightThrows response  -- | Send a command, assert the response is @OK@ commandOK :: HasCallStack => Client -> ByteString -> [ByteString] -> IO ()-commandOK Client {..} cmd args = withMVar clientConnection $ \conn -> do-  sendUnsafe clientSettings conn cmd args-  response <- recvUnsafe clientSettings conn+commandOK client cmd args = do+  response <- commandByteString client cmd args   unless (response == Right (Just "OK"))     $ throwString     $ "Server not OK. Reply was: "@@ -141,10 +141,19 @@   -> ByteString   -> [ByteString]   -> IO (Either String (Maybe a))-commandJSON Client {..} cmd args = withMVar clientConnection $ \conn -> do-  sendUnsafe clientSettings conn cmd args-  emByteString <- recvUnsafe clientSettings conn+commandJSON client cmd args = do+  emByteString <- commandByteString client cmd args   either (pure . Left) (pure . traverse eitherDecode) emByteString++commandByteString+  :: Client+  -> ByteString+  -> [ByteString]+  -> IO (Either String (Maybe ByteString))+commandByteString Client {..} cmd args = withMVar clientConnection $ \conn ->+  do+    sendUnsafe clientSettings conn cmd args+    recvUnsafe clientSettings conn  -- | Send a command to the Server socket --

@@ -0,0 +1,118 @@+{-# LANGUAGE DerivingVia #-}++-- | Support for the @BATCH@ command (Enterprise only)+--+-- <https://github.com/contribsys/faktory/wiki/Ent-Batches>+--+-- Batches allow multiple Jobs to be enqueued as a group, with a description+-- (visible in the admin UI) and Jobs attached to run on completion of all Jobs+-- within the group (always, or only if all were successful).+--+-- Usage:+--+-- @+-- -- Build a Job to run at completion of the Batch. Arguments are the same as+-- -- you would pass to 'perform' the Job.+-- onComplete <- buildJob mempty producer myJob+--+-- 'runBatch' ('complete' onComplete <> 'description' "My Batch") producer $ do+--   -- Use 'batchPerform' instead of 'perform'+--   void $ 'batchPerform' mempty producer myBatchedJob1+--   void $ 'batchPerform' mempty producer myBatchedJob2+-- @+--+-- __/NOTE__: This module does not support batched Jobs dynamically adding more+-- Jobs to the Batch. PRs welcome.+--+module Faktory.Ent.Batch+    (+    -- * Options+      BatchOptions+    , description+    , complete+    , success++    -- * Running+    , runBatch+    , Batch+    , batchPerform+    )+where++import Faktory.Prelude++import Control.Monad.Reader+import Data.Aeson+import Data.Aeson.Casing+import Data.ByteString.Lazy as BSL+import Data.Semigroup (Last(..))+import Data.Semigroup.Generic+import Data.Text.Encoding (decodeUtf8, encodeUtf8)+import Faktory.Client+import Faktory.Job+import Faktory.Producer+import GHC.Generics+import GHC.Stack++newtype Batch a = Batch (ReaderT BatchId IO a)+  deriving newtype (Functor, Applicative, Monad, MonadIO, MonadReader BatchId)++newtype BatchId = BatchId Text+  deriving newtype ToJSON++data BatchOptions arg = BatchOptions+  { boDescription :: Maybe (Last Text)+  , boSuccess :: Maybe (Last (Job arg))+  , boComplete :: Maybe (Last (Job arg))+  }+  deriving stock Generic+  deriving (Semigroup, Monoid) via GenericSemigroupMonoid (BatchOptions arg)++instance ToJSON arg => ToJSON (BatchOptions arg) where+  toJSON = genericToJSON $ aesonPrefix snakeCase+  toEncoding = genericToEncoding $ aesonPrefix snakeCase++description :: Text -> BatchOptions arg+description d = mempty { boDescription = Just $ Last d }++complete :: Job arg -> BatchOptions arg+complete job = mempty { boComplete = Just $ Last job }++success :: Job arg -> BatchOptions arg+success job = mempty { boSuccess = Just $ Last job }++runBatch :: ToJSON arg => BatchOptions arg -> Producer -> Batch a -> IO a+runBatch options producer (Batch f) = do+  bid <- newBatch producer options+  result <- runReaderT f bid+  result <$ commitBatch producer bid++newtype CustomBatchId = CustomBatchId+  { bid :: BatchId+  }+  deriving stock Generic+  deriving anyclass ToJSON++batchPerform+  :: (HasCallStack, ToJSON arg) => JobOptions -> Producer -> arg -> Batch JobId+batchPerform options producer arg = do+  bid <- ask+  Batch $ lift $ perform (options <> custom (CustomBatchId bid)) producer arg++newBatch :: ToJSON arg => Producer -> BatchOptions arg -> IO BatchId+newBatch producer options = do+  result <- commandByteString+    (producerClient producer)+    "BATCH NEW"+    [encode options]+  case result of+    Left err -> batchNewError err+    Right Nothing -> batchNewError "No BatchId returned"+    Right (Just bs) -> pure $ BatchId $ decodeUtf8 $ BSL.toStrict bs+  where batchNewError err = throwString $ "BATCH NEW error: " <> err++commitBatch :: Producer -> BatchId -> IO ()+commitBatch producer (BatchId bid) = command_+  (producerClient producer)+  "BATCH COMMIT"+  [BSL.fromStrict $ encodeUtf8 bid]

@@ -9,6 +9,8 @@   , jobtype   , at   , in_+  , custom+  , buildJob   , newJob   , jobJid   , jobArg@@ -17,44 +19,27 @@ import Faktory.Prelude  import Data.Aeson-import Data.Aeson.Casing import Data.List.NonEmpty (NonEmpty) import qualified Data.List.NonEmpty as NE-import Data.Time+import Data.Time (UTCTime) import Faktory.Client (Client(..))+import Faktory.Connection (ConnectionInfo(..))+import Faktory.JobOptions import Faktory.Producer (Producer(..), pushJob)-import Faktory.Settings-import GHC.Generics+import Faktory.Settings (Namespace, Settings(..)) import GHC.Stack import System.Random  data Job arg = Job   { jobJid :: JobId-  , jobJobtype :: String+  , jobAt :: Maybe UTCTime+  -- ^ Will be set based on 'JobOptions' when enqueued   , jobArgs :: NonEmpty arg   -- ^ Faktory needs to serialize args as a list, but we like a single-argument   -- interface so that's what we expose. See @'jobArg'@.-  , jobRetry :: Maybe Int-  , jobQueue :: Maybe Queue-  , jobAt :: Maybe UTCTime+  , jobOptions :: JobOptions   }-  deriving stock Generic --- | Individual changes to a @'Job'@ to be 'perform'ed-data JobUpdate-  = SetRetry Int-  | SetQueue Queue-  | SetJobtype String-  | SetAt UTCTime-  | SetIn NominalDiffTime---- | Options for the execution of a job------ See @'perform'@ for more details.----newtype JobOptions = JobOptions [JobUpdate]-  deriving newtype (Semigroup, Monoid)- -- | Perform a Job with the given options -- -- @@@ -69,48 +54,26 @@ perform   :: (HasCallStack, ToJSON arg) => JobOptions -> Producer -> arg -> IO JobId perform options producer arg = do-  let-    namespace =-      connectionInfoNamespace-        $ settingsConnection-        $ clientSettings-        $ producerClient producer-  job <- applyOptions namespace options =<< newJob arg+  job <- buildJob options producer arg   jobJid job <$ pushJob producer job  applyOptions :: Namespace -> JobOptions -> Job arg -> IO (Job arg)-applyOptions namespace (JobOptions patches) = go patches- where-  go [] job = pure job-  go (set : sets) job = case set of-    SetRetry n -> go sets $ job { jobRetry = Just n }-    SetQueue q ->-      go sets $ job { jobQueue = Just $ namespaceQueue namespace q }-    SetJobtype jt -> go sets $ job { jobJobtype = jt }-    SetAt time -> go sets $ job { jobAt = Just time }-    SetIn diff -> do-      now <- getCurrentTime-      go sets $ job { jobAt = Just $ addUTCTime diff now }--retry :: Int -> JobOptions-retry n = JobOptions [SetRetry n]---- | Equivalent to @'retry' (-1)@: no retries, and move to Dead on failure-once :: JobOptions-once = retry (-1)--queue :: Queue -> JobOptions-queue q = JobOptions [SetQueue q]--jobtype :: String -> JobOptions-jobtype jt = JobOptions [SetJobtype jt]--at :: UTCTime -> JobOptions-at t = JobOptions [SetAt t]+applyOptions namespace options job = do+  scheduledAt <- getAtFromSchedule options+  let namespacedOptions = namespaceQueue namespace $ jobOptions job <> options+  pure $ job { jobAt = scheduledAt, jobOptions = namespacedOptions } -in_ :: NominalDiffTime -> JobOptions-in_ i = JobOptions [SetIn i]+-- | Construct a 'Job' and apply options and Producer settings+buildJob :: JobOptions -> Producer -> arg -> IO (Job arg)+buildJob options producer arg = applyOptions namespace options =<< newJob arg+ where+  namespace =+    connectionInfoNamespace+      $ settingsConnection+      $ clientSettings+      $ producerClient producer +-- | Construct a 'Job' with default 'JobOptions' newJob :: arg -> IO (Job arg) newJob arg = do   -- Ruby uses 12 random hex@@ -118,21 +81,36 @@    pure Job     { jobJid = jobId-    , jobJobtype = "Default"-    , jobArgs = pure arg-    , jobRetry = Nothing-    , jobQueue = Nothing     , jobAt = Nothing+    , jobArgs = pure arg+    , jobOptions = jobtype "Default"     }  jobArg :: Job arg -> arg jobArg Job {..} = NE.head jobArgs  instance ToJSON args => ToJSON (Job args) where-  toJSON = genericToJSON $ aesonPrefix snakeCase-  toEncoding = genericToEncoding $ aesonPrefix snakeCase+  toJSON = object . toPairs+  toEncoding = pairs . mconcat . toPairs +toPairs :: (KeyValue a, ToJSON arg) => Job arg -> [a]+toPairs Job {..} =+  [ "jid" .= jobJid+  , "at" .= jobAt+  , "args" .= jobArgs+  , "jobtype" .= joJobtype jobOptions+  , "retry" .= joRetry jobOptions+  , "queue" .= joQueue jobOptions+  , "custom" .= joCustom jobOptions+  ]++-- brittany-disable-next-binding+ instance FromJSON args => FromJSON (Job args) where-  parseJSON = genericParseJSON $ aesonPrefix snakeCase+  parseJSON = withObject "Job" $ \o -> Job+    <$> o .: "jid"+    <*> o .:? "at"+    <*> o .: "args"+    <*> parseJSON (Object o)  type JobId = String

@@ -0,0 +1,26 @@+-- | Wrapper for the schema-less 'Value' used for @custom@ in Job payloads+--+-- This type's 'Semigroup' will merge two Objects. It is right-biased, as we are+-- generally throughout this library, so called /last-wins/ semantics.+--+module Faktory.Job.Custom+    ( Custom+    , toCustom+    ) where++import Faktory.Prelude++import Data.Aeson+import qualified Data.HashMap.Strict as HashMap++newtype Custom = Custom Value+  deriving stock (Eq, Show)+  deriving newtype (FromJSON, ToJSON)++toCustom :: ToJSON a => a -> Custom+toCustom = Custom . toJSON++instance Semigroup Custom where+  (Custom (Object a)) <> (Custom (Object b)) =+    Custom $ Object $ HashMap.union b a+  _ <> b = b

@@ -0,0 +1,96 @@+{-# LANGUAGE DerivingVia #-}++module Faktory.JobOptions+  ( JobOptions(..)++  -- * Modifiers+  , retry+  , once+  , queue+  , jobtype+  , at+  , in_+  , custom++  -- * Enqueue-time modifiers+  , getAtFromSchedule+  , namespaceQueue+  ) where++import Faktory.Prelude++import Data.Aeson+import Data.Semigroup (Last(..))+import Data.Semigroup.Generic+import Data.Time+import Faktory.Job.Custom+import Faktory.Settings (Namespace, Queue)+import qualified Faktory.Settings as Settings+import GHC.Generics++-- | Options for the execution of a job+--+-- These can be constructed using '(<>)':+--+-- @+-- let options = 'retry' 1 <> 'jobtype' "MyJob"+-- @+--+-- To enqueue with defaults, use 'mempty'.+--+-- Options use 'Last' semantics, so (e.g.) @'retry' x <>@ will set retries to+-- @x@ only if not already set, and @<> 'retry' x@ will override any+-- already-present retries to @x@.+--+data JobOptions = JobOptions+  { joJobtype :: Maybe (Last String)+  , joRetry :: Maybe (Last Int)+  , joQueue :: Maybe (Last Queue)+  , joSchedule :: Maybe (Last (Either UTCTime NominalDiffTime))+  , joCustom :: Maybe Custom+  }+  deriving stock Generic+  deriving (Semigroup, Monoid) via GenericSemigroupMonoid JobOptions++-- brittany-disable-next-binding++instance FromJSON JobOptions where+  parseJSON = withObject "JobOptions" $ \o ->+    JobOptions+      <$> o .:? "jobtype"+      <*> o .:? "retry"+      <*> o .:? "queue"+      <*> (fmap (Last . Right) <$> o .:? "at")+      <*> o .:? "custom"++getAtFromSchedule :: JobOptions -> IO (Maybe UTCTime)+getAtFromSchedule options = for (getLast <$> joSchedule options) $ \case+  Left t -> pure t+  Right nd -> addUTCTime nd <$> getCurrentTime++namespaceQueue :: Namespace -> JobOptions -> JobOptions+namespaceQueue namespace options = case joQueue options of+  Nothing -> options+  Just (Last q) -> options <> queue (Settings.namespaceQueue namespace q)++retry :: Int -> JobOptions+retry n = mempty { joRetry = Just $ Last n }++-- | Equivalent to @'retry' (-1)@: no retries, and move to Dead on failure+once :: JobOptions+once = retry (-1)++queue :: Queue -> JobOptions+queue q = mempty { joQueue = Just $ Last q }++jobtype :: String -> JobOptions+jobtype jt = mempty { joJobtype = Just $ Last jt }++at :: UTCTime -> JobOptions+at t = mempty { joSchedule = Just $ Last $ Left t }++in_ :: NominalDiffTime -> JobOptions+in_ i = mempty { joSchedule = Just $ Last $ Right i }++custom :: ToJSON a => a -> JobOptions+custom v = mempty { joCustom = Just $ toCustom v }

@@ -73,6 +73,7 @@     }  newtype Queue = Queue Text+  deriving stock (Eq, Show)   deriving newtype (IsString, FromJSON, ToJSON)  namespaceQueue :: Namespace -> Queue -> Queue

@@ -58,7 +58,11 @@   toEncoding = genericToEncoding $ aesonPrefix snakeCase  runWorker-  :: FromJSON args => Settings -> WorkerSettings -> (args -> IO ()) -> IO ()+  :: (HasCallStack, FromJSON args)+  => Settings+  -> WorkerSettings+  -> (args -> IO ())+  -> IO () runWorker settings workerSettings f = do   workerId <- maybe randomWorkerId pure $ settingsId workerSettings   client <- newClient settings $ Just workerId@@ -75,7 +79,7 @@   runWorker settings workerSettings f  processorLoop-  :: FromJSON arg+  :: (HasCallStack, FromJSON arg)   => Client   -> Settings   -> WorkerSettings

@@ -0,0 +1,100 @@+module Faktory.JobOptionsSpec+  ( spec+  ) where++import Faktory.Prelude++import Data.Aeson+import Data.Semigroup (Last(..))+import Data.Time+  ( DiffTime+  , UTCTime(..)+  , addUTCTime+  , diffTimeToPicoseconds+  , getCurrentTime+  , secondsToDiffTime+  )+import Data.Time.Calendar (fromGregorian)+import Faktory.Job.Custom+import Faktory.JobOptions+import Faktory.Settings (Namespace(..), Queue(..))+import Test.Hspec++spec :: Spec+spec = do+  describe "JobOptions" $ do+    describe "(<>)" $ do+      it "uses Last semantics" $ do+        let options = retry 1 <> retry 2++        fmap getLast (joRetry options) `shouldBe` Just 2++      it "merges custom fields" $ do+        let+          custom1 = object ["a" .= True, "b" .= True, "c" .= True]+          custom2 = object ["a" .= False, "d" .= False]+          options = custom custom1 <> custom custom2++        joCustom options+          `shouldBe` Just+                       (toCustom+                       $ object+                           [ "a" .= False+                           , "b" .= True+                           , "c" .= True+                           , "d" .= False+                           ]+                       )++    describe "getAtFromSchedule" $ do+      it "sets at based on in" $ do+        now <- getCurrentTime+        let time = addUTCTime 30 now++        timeAt <- getAtFromSchedule $ in_ 30++        fmap truncateUTCTime timeAt `shouldBe` Just (truncateUTCTime time)++      it "respects the last option given" $ do+        now <- getCurrentTime+        let time = makeUTCTime 2020 1 1 0++        timeAt1 <- getAtFromSchedule $ in_ 5 <> at time+        timeAt2 <- getAtFromSchedule $ at time <> in_ 5++        timeAt1 `shouldBe` Just time+        fmap truncateUTCTime timeAt2+          `shouldBe` Just (truncateUTCTime $ addUTCTime 5 now)++    describe "namespaceQueue" $ do+      it "leaves no queue alone" $ do+        let options = namespaceQueue (Namespace "namespace.") mempty++        fmap getLast (joQueue options) `shouldBe` Nothing++      it "namespaces a queue when given" $ do+        let options = namespaceQueue (Namespace "namespace.") $ queue "queue"++        fmap getLast (joQueue options) `shouldBe` Just (Queue "namespace.queue")++makeUTCTime :: Integer -> Int -> Int -> DiffTime -> UTCTime+makeUTCTime y m d s =+  UTCTime { utctDay = fromGregorian y m d, utctDayTime = s }++truncateUTCTime :: UTCTime -> UTCTime+truncateUTCTime t = t+  { utctDayTime =+    secondsToDiffTime+    $ picosecondsToSeconds+    $ diffTimeToPicoseconds+    $ utctDayTime t+  }++-- sameSecondAs :: UTCTime -> UTCTime -> Bool+-- sameSecondAs a b =+--   (utctDay a == utctDay b)+--     &&+--     == picosecondsToSeconds (diffTimeToPicoseconds $ utctDayTime b)++picosecondsToSeconds :: Integer -> Integer+picosecondsToSeconds = round @Double . (/ 1000000000000) . fromIntegral

@@ -4,8 +4,11 @@  import Faktory.Prelude -import Control.Concurrent (forkIO, threadDelay)+import Control.Concurrent (threadDelay)+import Control.Concurrent.Async import Control.Concurrent.MVar+import Control.Monad.IO.Class (liftIO)+import Faktory.Ent.Batch import Faktory.Job import Faktory.Producer import Faktory.Settings@@ -15,42 +18,21 @@ spec :: Spec spec = describe "Faktory" $ do   it "can push and process jobs" $ do-    bracket newProducerEnv closeProducer $ \producer -> do-      void $ flush producer+    jobs <- workerTestCase $ \producer -> do       void $ perform @Text mempty producer "a"       void $ perform @Text mempty producer "b"-      void $ perform @Text mempty producer "HALT" -    processedJobs <- newMVar ([] :: [Text])-    runWorkerEnv $ \job -> do-      modifyMVar_ processedJobs $ pure . (job :)-      when (job == "HALT") $ throw WorkerHalt--    jobs <- readMVar processedJobs     jobs `shouldMatchList` ["a", "b", "HALT"]    it "can push jobs with optional attributes" $ do-    bracket newProducerEnv closeProducer $ \producer -> do-      void $ flush producer+    jobs <- workerTestCase $ \producer -> do       void $ perform @Text once producer "a"       void $ perform @Text (retry 0) producer "b"-      void $ perform @Text mempty producer "HALT" -    processedJobs <- newMVar ([] :: [Text])-    runWorkerEnv $ \job -> do-      modifyMVar_ processedJobs $ pure . (job :)-      when (job == "HALT") $ throw WorkerHalt--    jobs <- readMVar processedJobs     jobs `shouldMatchList` ["a", "b", "HALT"]    it "correctly handles fetch timeouts" $ do-    settings <- envSettings-    workerSettings' <- envWorkerSettings-    let workerSettings = workerSettings' { settingsIdleDelay = 0 }--    -- start a background thread that waits for longer than the fetch timeout,-    -- then stops the worker.+    -- Pause longer than the fetch timeout     --     -- https://github.com/contribsys/faktory/wiki/Worker-Lifecycle#fetching-jobs     --@@ -58,16 +40,102 @@     -- the Server and handles it correctly. Setting our own idle delay to 0     -- ensures that we'll pick up the following HALT message immediately.     ---    void $ forkIO $ bracket (newProducer settings) closeProducer $ \producer ->-      do-        void $ flush producer-        threadDelay $ 2 * 1000000 + 250000-        void $ perform @Text mempty producer "HALT"--    processedJobs <- newMVar ([] :: [Text])-    runWorker settings workerSettings $ \job -> do-      modifyMVar_ processedJobs $ pure . (job :)-      when (job == "HALT") $ throw WorkerHalt+    let editSettings ws = ws { settingsIdleDelay = 0 }+    jobs <- workerTestCaseWith editSettings $ \_ -> do+      threadDelay $ 2 * 1000000 + 250000 -    jobs <- readMVar processedJobs     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