faktory 1.1.2.4 → 1.1.2.5
raw patch · 29 files changed
+389/−345 lines, 29 filesdep +crypton-connectiondep −connectionPVP ok
version bump matches the API change (PVP)
Dependencies added: crypton-connection
Dependencies removed: connection
API changes (from Hackage documentation)
Files
- CHANGELOG.md +6/−1
- README.lhs +3/−0
- examples/consumer/Main.hs +3/−3
- examples/producer/Main.hs +4/−4
- faktory.cabal +2/−2
- library/Faktory/Client.hs +51/−49
- library/Faktory/Connection.hs +42/−28
- library/Faktory/Ent/Batch.hs +24/−23
- library/Faktory/Ent/Batch/Status.hs +10/−9
- library/Faktory/Ent/Tracking.hs +17/−20
- library/Faktory/Job.hs +33/−29
- library/Faktory/Job/Custom.hs +4/−5
- library/Faktory/JobFailure.hs +9/−8
- library/Faktory/JobOptions.hs +11/−12
- library/Faktory/JobState.hs +7/−7
- library/Faktory/Prelude.hs +2/−0
- library/Faktory/Producer.hs +1/−2
- library/Faktory/Protocol.hs +5/−5
- library/Faktory/Settings.hs +25/−23
- library/Faktory/Settings/Queue.hs +2/−2
- library/Faktory/Worker.hs +11/−9
- library/Network/Connection/Compat.hs +24/−25
- tests/Faktory/ConnectionSpec.hs +29/−36
- tests/Faktory/Ent/TrackingSpec.hs +7/−6
- tests/Faktory/JobOptionsSpec.hs +20/−19
- tests/Faktory/JobSpec.hs +32/−13
- tests/Faktory/Test.hs +3/−3
- tests/FaktorySpec.hs +1/−1
- tests/Spec.hs +1/−1
CHANGELOG.md view
@@ -1,4 +1,9 @@-## [_Unreleased_](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.2.4...main)+## [_Unreleased_](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.2.5...main)++## [v1.1.2.5](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.2.4...v1.1.2.5)++- Migrate to `crypton-connection`+- Remove CI for GHC 8.6 ## [v1.1.2.4](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.2.3...v1.1.2.4)
README.lhs view
@@ -1,5 +1,8 @@ # faktory\_worker\_haskell +[](https://hackage.haskell.org/package/faktory)+[](http://stackage.org/nightly/package/faktory)+[](http://stackage.org/lts/package/faktory) [](https://github.com/freckle/faktory_worker_haskell/actions/workflows/ci.yml) Haskell client and worker process for the Faktory background job server.
examples/consumer/Main.hs view
@@ -8,9 +8,9 @@ import GHC.Generics -- | Must match examples/producer-newtype Job = Job { jobMessage :: String }- deriving stock Generic- deriving anyclass FromJSON+newtype Job = Job {jobMessage :: String}+ deriving stock (Generic)+ deriving anyclass (FromJSON) main :: IO () main = do
examples/producer/Main.hs view
@@ -10,13 +10,13 @@ import System.Environment (getArgs) -- | Must match examples/consumer-newtype Job = Job { jobMessage :: String }- deriving stock Generic- deriving anyclass ToJSON+newtype Job = Job {jobMessage :: String}+ deriving stock (Generic)+ deriving anyclass (ToJSON) main :: IO () main = bracket newProducerEnv closeProducer $ \producer -> do args <- getArgs- jobId <- perform mempty producer Job { jobMessage = unwords args }+ jobId <- perform mempty producer Job {jobMessage = unwords args} putStrLn $ "Pushed job: " <> show jobId
faktory.cabal view
@@ -1,6 +1,6 @@ cabal-version: 1.18 name: faktory-version: 1.1.2.4+version: 1.1.2.5 license: MIT license-file: LICENSE copyright: 2018 Freckle Education@@ -92,7 +92,7 @@ aeson-casing, base >=4 && <5, bytestring,- connection,+ crypton-connection, cryptonite, errors, megaparsec,
library/Faktory/Client.hs view
@@ -1,11 +1,10 @@ module Faktory.Client- (- -- * Client operations- Client(..)+ ( -- * Client operations+ Client (..) , newClient , closeClient - -- * High-level Client API+ -- * High-level Client API , command_ , commandOK , commandJSON@@ -15,7 +14,7 @@ import Faktory.Prelude import Control.Concurrent.MVar-import Crypto.Hash (Digest, SHA256(..), hashWith)+import Crypto.Hash (Digest, SHA256 (..), hashWith) import Data.Aeson import Data.Bitraversable (bimapM) import Data.ByteArray (ByteArrayAccess)@@ -44,8 +43,8 @@ } instance FromJSON HiPayload where- parseJSON = withObject "HiPayload"- $ \o -> HiPayload <$> o .: "v" <*> o .:? "s" <*> o .:? "i"+ parseJSON = withObject "HiPayload" $+ \o -> HiPayload <$> o .: "v" <*> o .:? "s" <*> o .:? "i" data HelloPayload = HelloPayload { helloWorkerId :: Maybe WorkerId@@ -57,22 +56,25 @@ } instance ToJSON HelloPayload where- toJSON HelloPayload {..} = object- [ "wid" .= helloWorkerId- , "hostname" .= helloHostname- , "pid" .= helloProcessId- , "labels" .= helloLabels- , "v" .= helloVersion- , "pwdhash" .= helloPasswordHash- ]- toEncoding HelloPayload {..} = pairs $ mconcat- [ "wid" .= helloWorkerId- , "hostname" .= helloHostname- , "pid" .= helloProcessId- , "labels" .= helloLabels- , "v" .= helloVersion- , "pwdhash" .= helloPasswordHash- ]+ toJSON HelloPayload {..} =+ object+ [ "wid" .= helloWorkerId+ , "hostname" .= helloHostname+ , "pid" .= helloProcessId+ , "labels" .= helloLabels+ , "v" .= helloVersion+ , "pwdhash" .= helloPasswordHash+ ]+ toEncoding HelloPayload {..} =+ pairs $+ mconcat+ [ "wid" .= helloWorkerId+ , "hostname" .= helloHostname+ , "pid" .= helloProcessId+ , "labels" .= helloLabels+ , "v" .= helloVersion+ , "pwdhash" .= helloPasswordHash+ ] -- | Open a new @'Client'@ connection with the given @'Settings'@ newClient :: HasCallStack => Settings -> Maybe WorkerId -> IO Client@@ -82,21 +84,23 @@ greeting <- fromJustThrows "Unexpected end of HI message"- =<< fromRightThrows- =<< recvUnsafe settings conn+ =<< fromRightThrows+ =<< recvUnsafe settings conn stripped <-- fromJustThrows ("Missing HI prefix: " <> show greeting)- $ BSL8.stripPrefix "HI" greeting+ fromJustThrows ("Missing HI prefix: " <> show greeting) $+ BSL8.stripPrefix "HI" greeting HiPayload {..} <-- fromJustThrows ("Failed to parse HI payload: " <> show stripped)- $ decode stripped+ fromJustThrows ("Failed to parse HI payload: " <> show stripped) $+ decode stripped - when (hiVersion > expectedProtocolVersion) $ settingsLogError $ concat- [ "Server's protocol version "- , show hiVersion- , " higher than client's expected protocol version "- , show expectedProtocolVersion- ]+ when (hiVersion > expectedProtocolVersion) $+ settingsLogError $+ concat+ [ "Server's protocol version "+ , show hiVersion+ , " higher than client's expected protocol version "+ , show expectedProtocolVersion+ ] let mPassword = connectionInfoPassword settingsConnection@@ -104,14 +108,15 @@ helloPayload <- HelloPayload mWorkerId (show . fst $ connectionID conn)- <$> (toInteger <$> getProcessID)- <*> pure ["haskell"]- <*> pure expectedProtocolVersion- <*> pure mHashedPassword+ <$> (toInteger <$> getProcessID)+ <*> pure ["haskell"]+ <*> pure expectedProtocolVersion+ <*> pure mHashedPassword commandOK client "HELLO" [encode helloPayload] pure client- where fromJustThrows message = maybe (throwString message) pure+ where+ fromJustThrows message = maybe (throwString message) pure -- | Close a @'Client'@ closeClient :: Client -> IO ()@@ -129,10 +134,10 @@ commandOK :: HasCallStack => Client -> ByteString -> [ByteString] -> IO () commandOK client cmd args = do response <- commandByteString client cmd args- unless (response == Right (Just "OK"))- $ throwString- $ "Server not OK. Reply was: "- <> show response+ unless (response == Right (Just "OK")) $+ throwString $+ "Server not OK. Reply was: "+ <> show response -- | Send a command, parse the response as JSON commandJSON@@ -158,7 +163,6 @@ -- | Send a command to the Server socket -- -- Do not use outside of @'withMVar'@, this is not threadsafe.--- sendUnsafe :: Settings -> Connection -> ByteString -> [ByteString] -> IO () sendUnsafe Settings {..} conn cmd args = do let bs = BSL8.unwords (cmd : args)@@ -168,7 +172,6 @@ -- | Receive data from the Server socket -- -- Do not use outside of @'withMVar'@, this is not threadsafe.--- recvUnsafe :: Settings -> Connection -> IO (Either String (Maybe ByteString)) recvUnsafe Settings {..} conn = do emByteString <- readReply $ connectionGet conn 4096@@ -178,7 +181,6 @@ -- | Iteratively apply a function @n@ times -- -- This is like @iterate f s !! n@ but strict in @s@--- times :: Int -> (s -> s) -> s -> s times n f !s | n <= 0 = s@@ -193,13 +195,13 @@ . hash . T.encodeUtf8 $ T.pack password- <> nonce+ <> nonce where -- Note that we use hash at two different types above. -- -- 1. hash :: ByteString -> Digest SHA256 -- 2. hash :: Digest SHA256 -> Digest SHA256- hash :: (ByteArrayAccess b) => b -> Digest SHA256+ hash :: ByteArrayAccess b => b -> Digest SHA256 hash = hashWith SHA256 -- | Protocol version the client expects
library/Faktory/Connection.hs view
@@ -1,6 +1,6 @@ module Faktory.Connection- ( ConnectionInfo(..)- , Namespace(..)+ ( ConnectionInfo (..)+ , Namespace (..) , defaultConnectionInfo , envConnectionInfo , connect@@ -15,7 +15,15 @@ import Network.Socket (HostName, PortNumber) import System.Environment (lookupEnv) import Text.Megaparsec- (Parsec, anySingle, errorBundlePretty, manyTill, optional, parse, some, (<?>))+ ( Parsec+ , anySingle+ , errorBundlePretty+ , manyTill+ , optional+ , parse+ , some+ , (<?>)+ ) import Text.Megaparsec.Char (char, digitChar, string, upperChar) newtype Namespace = Namespace Text@@ -31,13 +39,14 @@ deriving stock (Eq, Show) defaultConnectionInfo :: ConnectionInfo-defaultConnectionInfo = ConnectionInfo- { connectionInfoTls = False- , connectionInfoPassword = Nothing- , connectionInfoHostName = "localhost"- , connectionInfoPort = 7419- , connectionInfoNamespace = Namespace ""- }+defaultConnectionInfo =+ ConnectionInfo+ { connectionInfoTls = False+ , connectionInfoPassword = Nothing+ , connectionInfoHostName = "localhost"+ , connectionInfoPort = 7419+ , connectionInfoNamespace = Namespace ""+ } -- | Parse a @'Connection'@ from environment variables --@@ -47,7 +56,6 @@ -- Supported format is @tcp(+tls):\/\/(:password@)host:port(/namespace)@. -- -- See <https://github.com/contribsys/faktory/wiki/Worker-Lifecycle#url-configuration>.--- envConnectionInfo :: IO ConnectionInfo envConnectionInfo = do providerString <- fromMaybe "FAKTORY_URL" <$> lookupEnv "FAKTORY_PROVIDER"@@ -61,29 +69,35 @@ where open = do ctx <- initConnectionContext- connectTo ctx $ ConnectionParams- { connectionHostname = connectionInfoHostName- , connectionPort = connectionInfoPort- , connectionUseSecure = if connectionInfoTls- then Just TLSSettingsSimple- { settingDisableCertificateValidation = False- , settingDisableSession = False- , settingUseServerName = False- }- else Nothing- , connectionUseSocks = Nothing- }+ connectTo ctx $+ ConnectionParams+ { connectionHostname = connectionInfoHostName+ , connectionPort = connectionInfoPort+ , connectionUseSecure =+ if connectionInfoTls+ then+ Just+ TLSSettingsSimple+ { settingDisableCertificateValidation = False+ , settingDisableSession = False+ , settingUseServerName = False+ }+ else Nothing+ , connectionUseSocks = Nothing+ } type Parser = Parsec Void String parseThrow :: Parser a -> String -> String -> IO a parseThrow parser name value = either err pure $ parse parser name value where- err ex = throwIO . userError $ unlines- [ ""- , "\"" <> value <> "\" is an invalid value for " <> name <> ":"- , errorBundlePretty ex- ]+ err ex =+ throwIO . userError $+ unlines+ [ ""+ , "\"" <> value <> "\" is an invalid value for " <> name <> ":"+ , errorBundlePretty ex+ ] parseProvider :: Parser String parseProvider =
library/Faktory/Ent/Batch.hs view
@@ -23,23 +23,21 @@ -- -- __/NOTE__: This module does not support batched Jobs dynamically adding more -- Jobs to the Batch. PRs welcome.--- module Faktory.Ent.Batch- (- -- * Options+ ( -- * Options BatchOptions , description , complete , success - -- * Running+ -- * Running , runBatch , Batch , batchPerform - -- * Low-level- , BatchId(..)- , CustomBatchId(..)+ -- * Low-level+ , BatchId (..)+ , CustomBatchId (..) , newBatch , commitBatch ) where@@ -50,7 +48,7 @@ import Data.Aeson import Data.Aeson.Casing import Data.ByteString.Lazy as BSL-import Data.Semigroup (Last(..))+import Data.Semigroup (Last (..)) import Data.Semigroup.Generic import Data.Text.Encoding (decodeUtf8, encodeUtf8) import Faktory.Client@@ -71,7 +69,7 @@ , boSuccess :: Maybe (Last (Job arg)) , boComplete :: Maybe (Last (Job arg)) }- deriving stock Generic+ deriving stock (Generic) deriving (Semigroup, Monoid) via GenericSemigroupMonoid (BatchOptions arg) instance ToJSON arg => ToJSON (BatchOptions arg) where@@ -79,13 +77,13 @@ toEncoding = genericToEncoding $ aesonPrefix snakeCase description :: Text -> BatchOptions arg-description d = mempty { boDescription = Just $ Last d }+description d = mempty {boDescription = Just $ Last d} complete :: Job arg -> BatchOptions arg-complete job = mempty { boComplete = Just $ Last job }+complete job = mempty {boComplete = Just $ Last job} success :: Job arg -> BatchOptions arg-success job = mempty { boSuccess = Just $ Last job }+success job = mempty {boSuccess = Just $ Last job} runBatch :: ToJSON arg => BatchOptions arg -> Producer -> Batch a -> IO a runBatch options producer (Batch f) = do@@ -96,8 +94,8 @@ newtype CustomBatchId = CustomBatchId { bid :: BatchId }- deriving stock Generic- deriving anyclass ToJSON+ deriving stock (Generic)+ deriving anyclass (ToJSON) batchPerform :: (HasCallStack, ToJSON arg) => JobOptions -> Producer -> arg -> Batch JobId@@ -107,18 +105,21 @@ newBatch :: ToJSON arg => Producer -> BatchOptions arg -> IO BatchId newBatch producer options = do- result <- commandByteString- (producerClient producer)- "BATCH NEW"- [encode options]+ 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+ 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]+commitBatch producer (BatchId bid) =+ command_+ (producerClient producer)+ "BATCH COMMIT"+ [BSL.fromStrict $ encodeUtf8 bid]
library/Faktory/Ent/Batch/Status.hs view
@@ -1,6 +1,6 @@ module Faktory.Ent.Batch.Status ( jobBatchId- , BatchStatus(..)+ , BatchStatus (..) , batchStatus ) where @@ -16,7 +16,7 @@ import Faktory.Ent.Batch import Faktory.Job (Job, jobOptions) import Faktory.Job.Custom-import Faktory.JobOptions (JobOptions(..))+import Faktory.JobOptions (JobOptions (..)) import Faktory.Producer import GHC.Generics @@ -28,13 +28,13 @@ , created_at :: UTCTime , description :: Text }- deriving stock Generic- deriving anyclass FromJSON+ deriving stock (Generic)+ deriving anyclass (FromJSON) newtype ReadCustomBatchId = ReadCustomBatchId { _bid :: BatchId }- deriving stock (Show,Eq,Generic)+ deriving stock (Show, Eq, Generic) instance FromJSON ReadCustomBatchId where -- Faktory seems to use the key '_bid' when enqueuing callback jobs and 'bid' for normal jobs...@@ -49,7 +49,8 @@ _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]+batchStatus producer (BatchId bid) =+ commandJSON+ (producerClient producer)+ "BATCH STATUS"+ [BSL.fromStrict $ encodeUtf8 bid]
library/Faktory/Ent/Tracking.hs view
@@ -1,18 +1,15 @@ -- | Support for the @TRACK@ command (Enterprise only) -- -- <https://github.com/contribsys/faktory/wiki/Ent-Tracking>--- module Faktory.Ent.Tracking- ( CustomTrack(..)+ ( CustomTrack (..) , tracked , trackPerform-- , JobDetails(..)- , JobState(..)+ , JobDetails (..)+ , JobState (..) , trackGet , trackGetHush-- , SetJobDetails(..)+ , SetJobDetails (..) , trackSet ) where @@ -26,7 +23,7 @@ import Data.Time (UTCTime) import Faktory.Client (commandJSON, commandOK) import Faktory.Job (JobId, JobOptions, custom, perform)-import Faktory.JobState (JobState(..))+import Faktory.JobState (JobState (..)) import Faktory.Producer import GHC.Generics (Generic) import GHC.Stack (HasCallStack)@@ -34,8 +31,8 @@ newtype CustomTrack = CustomTrack { track :: Int }- deriving stock Generic- deriving anyclass ToJSON+ deriving stock (Generic)+ deriving anyclass (ToJSON) tracked :: JobOptions tracked = custom (CustomTrack 1)@@ -47,7 +44,6 @@ -- @ -- 'perform' ('custom' $ 'CustomTrack' 1) -- @--- trackPerform :: (HasCallStack, ToJSON arg) => JobOptions -> Producer -> arg -> IO JobId trackPerform options = perform (options <> custom (CustomTrack 1))@@ -60,19 +56,20 @@ , jdState :: JobState , jdUpdatedAt :: Maybe UTCTime }- deriving stock Generic+ 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- }+unknownJobDetails jid =+ JobDetails+ { jdJid = jid+ , jdPercent = Nothing+ , jdDesc = Nothing+ , jdState = JobStateUnknown+ , jdUpdatedAt = Nothing+ } data SetJobDetails = SetJobDetails { sjdJid :: JobId@@ -80,7 +77,7 @@ , sjdDesc :: Maybe Text , sjdReserveUntil :: Maybe UTCTime }- deriving stock Generic+ deriving stock (Generic) instance ToJSON SetJobDetails where toJSON = genericToJSON $ aesonPrefix snakeCase
library/Faktory/Job.hs view
@@ -25,14 +25,14 @@ import Data.Aeson import Data.List.NonEmpty (NonEmpty) import qualified Data.List.NonEmpty as NE-import Data.Semigroup (Last(..))+import Data.Semigroup (Last (..)) import Data.Time (UTCTime)-import Faktory.Client (Client(..))-import Faktory.Connection (ConnectionInfo(..))+import Faktory.Client (Client (..))+import Faktory.Connection (ConnectionInfo (..)) import Faktory.JobFailure import Faktory.JobOptions-import Faktory.Producer (Producer(..), pushJob)-import Faktory.Settings (Namespace, Settings(..))+import Faktory.Producer (Producer (..), pushJob)+import Faktory.Settings (Namespace, Settings (..)) import GHC.Stack import System.Random @@ -57,7 +57,6 @@ -- 'perform' ('in_' 10 <> 'once') SomeJob -- 'perform' ('in_' 10 <> 'retry' 3) SomeJob -- @--- perform :: (HasCallStack, ToJSON arg) => JobOptions -> Producer -> arg -> IO JobId perform options producer arg = do@@ -68,21 +67,25 @@ applyOptions namespace options job = do scheduledAt <- getAtFromSchedule options let namespacedOptions = namespaceQueue namespace $ jobOptions job <> options- pure $ job { jobAt = scheduledAt, jobOptions = namespacedOptions }+ pure $ job {jobAt = scheduledAt, jobOptions = namespacedOptions} -- | Construct a 'Job' and apply options and Producer settings buildJob :: JobOptions -> Producer -> arg -> IO (Job arg)-buildJob options producer arg = applyOptions namespace (applyDefaults options)- =<< newJob arg+buildJob options producer arg =+ applyOptions namespace (applyDefaults options)+ =<< newJob arg where namespace =- connectionInfoNamespace- $ settingsConnection- $ clientSettings- $ producerClient producer+ connectionInfoNamespace $+ settingsConnection $+ clientSettings $+ producerClient producer applyDefaults =- mappend $ settingsDefaultJobOptions $ clientSettings $ producerClient- producer+ mappend $+ settingsDefaultJobOptions $+ clientSettings $+ producerClient+ producer -- | Construct a 'Job' with default 'JobOptions' newJob :: arg -> IO (Job arg)@@ -90,13 +93,14 @@ -- Ruby uses 12 random hex jobId <- take 12 . randomRs ('a', 'z') <$> newStdGen - pure Job- { jobJid = jobId- , jobAt = Nothing- , jobArgs = pure arg- , jobOptions = jobtype "Default"- , jobFailure = Nothing- }+ pure+ Job+ { jobJid = jobId+ , jobAt = Nothing+ , jobArgs = pure arg+ , jobOptions = jobtype "Default"+ , jobFailure = Nothing+ } jobArg :: Job arg -> arg jobArg Job {..} = NE.head jobArgs@@ -132,19 +136,19 @@ -- brittany-disable-next-binding instance FromJSON args => FromJSON (Job args) where- parseJSON = withObject "Job" $ \o -> Job- <$> o .: "jid"- <*> o .:? "at"- <*> o .: "args"- <*> parseJSON (Object o)- <*> o .:? "failure"+ parseJSON = withObject "Job" $ \o ->+ Job+ <$> o .: "jid"+ <*> o .:? "at"+ <*> o .: "args"+ <*> parseJSON (Object o)+ <*> o .:? "failure" type JobId = String -- | https://github.com/contribsys/faktory/wiki/Job-Errors#the-process -- -- > By default Faktory will retry a job 25 times--- faktoryDefaultRetry :: Int faktoryDefaultRetry = 25
library/Faktory/Job/Custom.hs view
@@ -2,12 +2,11 @@ -- -- 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- , fromCustom- ) where+ ( Custom+ , toCustom+ , fromCustom+ ) where import Faktory.Prelude
library/Faktory/JobFailure.hs view
@@ -1,5 +1,5 @@ module Faktory.JobFailure- ( JobFailure(..)+ ( JobFailure (..) ) where import Faktory.Prelude@@ -19,10 +19,11 @@ -- brittany-disable-next-binding instance FromJSON JobFailure where- parseJSON = withObject "Failure" $ \o -> JobFailure- <$> o .: "retry_count"- <*> o .: "failed_at"- <*> o .:? "next_at"- <*> o .:? "error_message"- <*> o .:? "error_type"- <*> o .:? "backtrace"+ parseJSON = withObject "Failure" $ \o ->+ JobFailure+ <$> o .: "retry_count"+ <*> o .: "failed_at"+ <*> o .:? "next_at"+ <*> o .:? "error_message"+ <*> o .:? "error_type"+ <*> o .:? "backtrace"
library/Faktory/JobOptions.hs view
@@ -1,9 +1,9 @@ {-# LANGUAGE DerivingVia #-} module Faktory.JobOptions- ( JobOptions(..)+ ( JobOptions (..) - -- * Modifiers+ -- * Modifiers , retry , once , reserveFor@@ -13,7 +13,7 @@ , in_ , custom - -- * Enqueue-time modifiers+ -- * Enqueue-time modifiers , getAtFromSchedule , namespaceQueue ) where@@ -21,7 +21,7 @@ import Faktory.Prelude import Data.Aeson-import Data.Semigroup (Last(..))+import Data.Semigroup (Last (..)) import Data.Semigroup.Generic import Data.Time import Faktory.Job.Custom@@ -43,7 +43,6 @@ -- 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)@@ -78,26 +77,26 @@ Just (Last q) -> options <> queue (Settings.namespaceQueue namespace q) reserveFor :: Natural -> JobOptions-reserveFor n = mempty { joReserveFor = Just $ Last n }+reserveFor n = mempty {joReserveFor = Just $ Last n} retry :: Int -> JobOptions-retry n = mempty { joRetry = Just $ Last n }+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 }+queue q = mempty {joQueue = Just $ Last q} jobtype :: String -> JobOptions-jobtype jt = mempty { joJobtype = Just $ Last jt }+jobtype jt = mempty {joJobtype = Just $ Last jt} at :: UTCTime -> JobOptions-at t = mempty { joSchedule = Just $ Last $ Left t }+at t = mempty {joSchedule = Just $ Last $ Left t} in_ :: NominalDiffTime -> JobOptions-in_ i = mempty { joSchedule = Just $ Last $ Right i }+in_ i = mempty {joSchedule = Just $ Last $ Right i} custom :: ToJSON a => a -> JobOptions-custom v = mempty { joCustom = Just $ toCustom v }+custom v = mempty {joCustom = Just $ toCustom v}
library/Faktory/JobState.hs view
@@ -1,5 +1,5 @@ module Faktory.JobState- ( JobState(..)+ ( JobState (..) ) where import Faktory.Prelude@@ -39,12 +39,12 @@ jobStateFromText :: Text -> Either String JobState jobStateFromText x = note- (unpack- $ "Invalid JobState: "- <> x- <> ", must be one of "- <> T.intercalate ", " (HashMap.keys jobStateMap)- )+ ( unpack $+ "Invalid JobState: "+ <> x+ <> ", must be one of "+ <> T.intercalate ", " (HashMap.keys jobStateMap)+ ) $ HashMap.lookup x jobStateMap jobStateMap :: HashMap Text JobState
library/Faktory/Prelude.hs view
@@ -13,6 +13,8 @@ import Data.Text as X (Text, pack, unpack) import Data.Traversable as X +{-# ANN module ("HLint: ignore Avoid restricted alias" :: String) #-}+ threadDelaySeconds :: Int -> IO () threadDelaySeconds n = threadDelay $ n * 1000000
library/Faktory/Producer.hs view
@@ -1,5 +1,5 @@ module Faktory.Producer- ( Producer(..)+ ( Producer (..) , newProducer , newProducerEnv , closeProducer@@ -35,6 +35,5 @@ -- | Clear all job data in the Faktory server -- -- Use with caution!--- flush :: HasCallStack => Producer -> IO () flush producer = commandOK (producerClient producer) "FLUSH" []
library/Faktory/Protocol.hs view
@@ -4,10 +4,9 @@ -- -- Faktory takes a lot of inspiration from Redis, so the connection and -- protocol-related code translated well with minor simplifications.--- module Faktory.Protocol ( readReply- , Reply(..)+ , Reply (..) , reply ) where @@ -55,9 +54,10 @@ {-# INLINE bulk #-} bulk :: Scanner Reply-bulk = Bulk <$> do- len <- integral- if len < 0 then return Nothing else Just <$> Scanner.take len <* eol+bulk =+ Bulk <$> do+ len <- integral+ if len < 0 then return Nothing else Just <$> Scanner.take len <* eol {-# INLINE integral #-} integral :: Integral i => Scanner i
library/Faktory/Settings.hs view
@@ -1,20 +1,20 @@ module Faktory.Settings- ( Settings(..)+ ( Settings (..) , defaultSettings , envSettings- , WorkerSettings(..)+ , WorkerSettings (..) , defaultWorkerSettings , envWorkerSettings- , Queue(..)+ , Queue (..) , namespaceQueue , queueArg , defaultQueue , WorkerId , randomWorkerId - -- * Re-exports- , ConnectionInfo(..)- , Namespace(..)+ -- * Re-exports+ , ConnectionInfo (..)+ , Namespace (..) ) where import Faktory.Prelude@@ -35,21 +35,21 @@ } defaultSettings :: Settings-defaultSettings = Settings- { settingsConnection = defaultConnectionInfo- , settingsLogDebug = \_msg -> pure ()- , settingsLogError = hPutStrLn stderr . ("[ERROR]: " <>)- , settingsDefaultJobOptions = mempty- }+defaultSettings =+ Settings+ { settingsConnection = defaultConnectionInfo+ , settingsLogDebug = \_msg -> pure ()+ , settingsLogError = hPutStrLn stderr . ("[ERROR]: " <>)+ , settingsDefaultJobOptions = mempty+ } -- | Defaults, but read @'Connection'@ from the environment -- -- See @'envConnection'@--- envSettings :: IO Settings envSettings = do connection <- envConnectionInfo- pure defaultSettings { settingsConnection = connection }+ pure defaultSettings {settingsConnection = connection} data WorkerSettings = WorkerSettings { settingsQueue :: Queue@@ -58,20 +58,22 @@ } defaultWorkerSettings :: WorkerSettings-defaultWorkerSettings = WorkerSettings- { settingsQueue = defaultQueue- , settingsId = Nothing- , settingsIdleDelay = 1- }+defaultWorkerSettings =+ WorkerSettings+ { settingsQueue = defaultQueue+ , settingsId = Nothing+ , settingsIdleDelay = 1+ } envWorkerSettings :: IO WorkerSettings envWorkerSettings = do mQueue <- lookupEnv "FAKTORY_QUEUE" mWorkerId <- lookupEnv "FAKTORY_WORKER_ID"- pure defaultWorkerSettings- { settingsQueue = maybe defaultQueue (Queue . pack) mQueue- , settingsId = WorkerId <$> mWorkerId- }+ pure+ defaultWorkerSettings+ { settingsQueue = maybe defaultQueue (Queue . pack) mQueue+ , settingsId = WorkerId <$> mWorkerId+ } newtype WorkerId = WorkerId String deriving newtype (FromJSON, ToJSON)
library/Faktory/Settings/Queue.hs view
@@ -1,9 +1,9 @@ module Faktory.Settings.Queue- ( Queue(..)+ ( Queue (..) , namespaceQueue , queueArg , defaultQueue- , Namespace(..)+ , Namespace (..) ) where import Faktory.Prelude
library/Faktory/Worker.hs view
@@ -2,9 +2,8 @@ -- -- Runs forever, @FETCH@-ing Jobs from the given Queue and handing each to your -- processing function.--- module Faktory.Worker- ( WorkerHalt(..)+ ( WorkerHalt (..) , runWorker , runWorkerEnv , jobArg@@ -27,12 +26,12 @@ -- | If processing functions @'throw'@ this, @'runWorker'@ will exit data WorkerHalt = WorkerHalt deriving stock (Eq, Show)- deriving anyclass Exception+ deriving anyclass (Exception) newtype BeatPayload = BeatPayload { _bpWid :: WorkerId }- deriving stock Generic+ deriving stock (Generic) instance ToJSON BeatPayload where toJSON = genericToJSON $ aesonPrefix snakeCase@@ -41,7 +40,7 @@ newtype AckPayload = AckPayload { _apJid :: JobId }- deriving stock Generic+ deriving stock (Generic) instance ToJSON AckPayload where toJSON = genericToJSON $ aesonPrefix snakeCase@@ -53,7 +52,7 @@ , _fpJid :: JobId , _fpBacktrace :: [String] }- deriving stock Generic+ deriving stock (Generic) instance ToJSON FailPayload where toJSON = genericToJSON $ aesonPrefix snakeCase@@ -96,8 +95,11 @@ Nothing -> settingsLogError settings "Job reservation period expired." Just () -> ackJob client job - emJob <- fetchJob client $ namespaceQueue namespace $ settingsQueue- workerSettings+ emJob <-+ fetchJob client $+ namespaceQueue namespace $+ settingsQueue+ workerSettings case emJob of Left err -> settingsLogError settings $ "Invalid Job: " <> err@@ -106,7 +108,7 @@ processAndAck job `catches` [ Handler $ \(ex :: WorkerHalt) -> throw ex , Handler $ \(ex :: SomeException) ->- failJob client job $ T.pack $ show ex+ failJob client job $ T.pack $ show ex ] -- | <https://github.com/contribsys/faktory/wiki/Worker-Lifecycle#heartbeat>
library/Network/Connection/Compat.hs view
@@ -1,7 +1,6 @@ -- | Copy of 'Network.Connection.connectTo' with 'KeepAlive' -- -- <https://hackage.haskell.org/package/connection-0.3.1/docs/src/Network.Connection.html#connectTo>--- module Network.Connection.Compat ( connectTo , module Network.Connection@@ -22,31 +21,31 @@ (resolve (connectionHostname cParams) (connectionPort cParams)) (close . fst) ( \(h, _) ->- connectFromSocket cg h cParams+ connectFromSocket cg h cParams ) resolve :: String -> PortNumber -> IO (Socket, SockAddr) resolve host port = do- let hints = defaultHints { addrFlags = [AI_ADDRCONFIG], addrSocketType = Stream }- addrs <- getAddrInfo (Just hints) (Just host) (Just $ show port)- firstSuccessful $ map tryToConnect addrs- where- tryToConnect addr =- bracketOnError- (socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr))- close- (\sock -> do- setSocketOption sock KeepAlive 1- S.connect sock (addrAddress addr)- return (sock, addrAddress addr)- )- firstSuccessful = go []- where- go :: [IOException] -> [IO a] -> IO a- go [] [] = throwIO $ HostNotResolved host- go l@(_:_) [] = throwIO $ HostCannotConnect host l- go acc (act:followingActs) = do- er <- try act- case er of- Left err -> go (err:acc) followingActs- Right r -> return r+ let hints = defaultHints {addrFlags = [AI_ADDRCONFIG], addrSocketType = Stream}+ addrs <- getAddrInfo (Just hints) (Just host) (Just $ show port)+ firstSuccessful $ map tryToConnect addrs+ where+ tryToConnect addr =+ bracketOnError+ (socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr))+ close+ ( \sock -> do+ setSocketOption sock KeepAlive 1+ S.connect sock (addrAddress addr)+ return (sock, addrAddress addr)+ )+ firstSuccessful = go []+ where+ go :: [IOException] -> [IO a] -> IO a+ go [] [] = throwIO $ HostNotResolved host+ go l@(_ : _) [] = throwIO $ HostCannotConnect host l+ go acc (act : followingActs) = do+ er <- try act+ case er of+ Left err -> go (err : acc) followingActs+ Right r -> return r
tests/Faktory/ConnectionSpec.hs view
@@ -20,9 +20,8 @@ connection `shouldBe` defaultConnectionInfo it "parses the provided URL" $ do- let- env =- [("FAKTORY_PROVIDER", Nothing), ("FAKTORY_URL", Just "tcp://foo:123")]+ let env =+ [("FAKTORY_PROVIDER", Nothing), ("FAKTORY_URL", Just "tcp://foo:123")] withEnvironment env $ do ConnectionInfo {..} <- envConnectionInfo@@ -33,11 +32,10 @@ connectionInfoNamespace `shouldBe` Namespace "" it "parses tls and password" $ do- let- env =- [ ("FAKTORY_PROVIDER", Nothing)- , ("FAKTORY_URL", Just "tcp+tls://:foo@bar:123")- ]+ let env =+ [ ("FAKTORY_PROVIDER", Nothing)+ , ("FAKTORY_URL", Just "tcp+tls://:foo@bar:123")+ ] withEnvironment env $ do ConnectionInfo {..} <- envConnectionInfo@@ -47,11 +45,10 @@ connectionInfoPort `shouldBe` 123 it "parses namespace" $ do- let- env =- [ ("FAKTORY_PROVIDER", Nothing)- , ("FAKTORY_URL", Just "tcp://localhost:7419/prefix")- ]+ let env =+ [ ("FAKTORY_PROVIDER", Nothing)+ , ("FAKTORY_URL", Just "tcp://localhost:7419/prefix")+ ] withEnvironment env $ do ConnectionInfo {..} <- envConnectionInfo@@ -60,11 +57,10 @@ connectionInfoNamespace `shouldBe` Namespace "prefix" it "follows _PROVIDER to find _URL" $ do- let- env =- [ ("FAKTORY_PROVIDER", Just "OTHER_URL")- , ("OTHER_URL", Just "tcp://foo:123")- ]+ let env =+ [ ("FAKTORY_PROVIDER", Just "OTHER_URL")+ , ("OTHER_URL", Just "tcp://foo:123")+ ] withEnvironment env $ do ConnectionInfo {..} <- envConnectionInfo@@ -72,21 +68,19 @@ connectionInfoPort `shouldBe` 123 it "throws nice errors for invalid PROVIDER" $ do- let- env =- [ ("FAKTORY_PROVIDER", Just "flippity-$flop")- , ("FAKTORY_URL", Nothing)- ]+ let env =+ [ ("FAKTORY_PROVIDER", Just "flippity-$flop")+ , ("FAKTORY_URL", Nothing)+ ] withEnvironment env envConnectionInfo `shouldThrowMessage` "expecting an environment variable name" it "throws nice errors for invalid _URL" $ do- let- env =- [ ("FAKTORY_PROVIDER", Nothing)- , ("FAKTORY_URL", Just "http://foo:123")- ]+ let env =+ [ ("FAKTORY_PROVIDER", Nothing)+ , ("FAKTORY_URL", Just "http://foo:123")+ ] withEnvironment env envConnectionInfo `shouldThrowMessage` "expecting tcp(+tls)://(:<password>@)<host>:<port>"@@ -94,9 +88,8 @@ it "throws nice errors for missing _PROVIDER" $ do pendingWith "This makes implementation more complicated" - let- env =- [("FAKTORY_PROVIDER", Just "MISSING_URL"), ("MISSING_URL", Nothing)]+ let env =+ [("FAKTORY_PROVIDER", Just "MISSING_URL"), ("MISSING_URL", Nothing)] withEnvironment env envConnectionInfo `shouldThrowMessage` "..." @@ -104,12 +97,12 @@ -- -- Values are @'Maybe'@ so that @'Nothing'@ can be used to unset values during -- override and/or restoration. Note: this is probably not thread-safe.--- withEnvironment :: [(String, Maybe String)] -> IO a -> IO a-withEnvironment env act = bracket- (traverse readAndReset env)- (traverse_ readAndReset)- (const act)+withEnvironment env act =+ bracket+ (traverse readAndReset env)+ (traverse_ readAndReset)+ (const act) where readAndReset :: (String, Maybe String) -> IO (String, Maybe String) readAndReset (variable, mNewValue) = do
tests/Faktory/Ent/TrackingSpec.hs view
@@ -49,12 +49,13 @@ let [aJid] = enqueuedJobIds aDetails <- bracket newProducerEnv closeProducer $ \producer -> do- trackSet producer $ SetJobDetails- { sjdJid = aJid- , sjdPercent = Just 100- , sjdDesc = Just "Updated"- , sjdReserveUntil = Nothing- }+ trackSet producer+ $ SetJobDetails+ { sjdJid = aJid+ , sjdPercent = Just 100+ , sjdDesc = Just "Updated"+ , sjdReserveUntil = Nothing+ } trackGetHush producer aJid
tests/Faktory/JobOptionsSpec.hs view
@@ -5,10 +5,10 @@ import Faktory.Prelude import Data.Aeson-import Data.Semigroup (Last(..))+import Data.Semigroup (Last (..)) import Data.Time ( DiffTime- , UTCTime(..)+ , UTCTime (..) , addUTCTime , diffTimeToPicoseconds , getCurrentTime@@ -17,7 +17,7 @@ import Data.Time.Calendar (fromGregorian) import Faktory.Job.Custom import Faktory.JobOptions-import Faktory.Settings (Namespace(..), Queue(..))+import Faktory.Settings (Namespace (..), Queue (..)) import Test.Hspec spec :: Spec@@ -37,14 +37,14 @@ joCustom options `shouldBe` Just- (toCustom- $ object- [ "a" .= False- , "b" .= True- , "c" .= True- , "d" .= False- ]- )+ ( toCustom $+ object+ [ "a" .= False+ , "b" .= True+ , "c" .= True+ , "d" .= False+ ]+ ) describe "getAtFromSchedule" $ do it "sets at based on in" $ do@@ -79,16 +79,17 @@ makeUTCTime :: Integer -> Int -> Int -> DiffTime -> UTCTime makeUTCTime y m d s =- UTCTime { utctDay = fromGregorian y m d, utctDayTime = s }+ UTCTime {utctDay = fromGregorian y m d, utctDayTime = s} truncateUTCTime :: UTCTime -> UTCTime-truncateUTCTime t = t- { utctDayTime =- secondsToDiffTime- $ picosecondsToSeconds- $ diffTimeToPicoseconds- $ utctDayTime t- }+truncateUTCTime t =+ t+ { utctDayTime =+ secondsToDiffTime $+ picosecondsToSeconds $+ diffTimeToPicoseconds $+ utctDayTime t+ } -- sameSecondAs :: UTCTime -> UTCTime -> Bool -- sameSecondAs a b =
tests/Faktory/JobSpec.hs view
@@ -19,7 +19,9 @@ -- https://github.com/contribsys/faktory/issues/374#issuecomment-902075572 describe "jobRetriesRemaining" $ do it "handles first consumed" $ do- job <- decodeJob [aesonQQ|+ job <-+ decodeJob+ [aesonQQ| { "jid": "abc" , "args": [""] , "retry": 2@@ -30,7 +32,9 @@ it "handles a first retry" $ do now <- getCurrentTime- job <- decodeJob [aesonQQ|+ job <-+ decodeJob+ [aesonQQ| { "jid": "abc" , "args": [""] , "retry": 2@@ -45,7 +49,9 @@ it "handles a final retry" $ do now <- getCurrentTime- job <- decodeJob [aesonQQ|+ job <-+ decodeJob+ [aesonQQ| { "jid": "abc" , "args": [""] , "retry": 2@@ -59,7 +65,9 @@ jobRetriesRemaining job `shouldBe` 0 it "uses Faktory's default" $ do- job <- decodeJob [aesonQQ|+ job <-+ decodeJob+ [aesonQQ| { "jid": "abc" , "args": [""] }@@ -69,13 +77,17 @@ it "handles retry -1" $ do now <- getCurrentTime- job1 <- decodeJob [aesonQQ|+ job1 <-+ decodeJob+ [aesonQQ| { "jid": "abc" , "args": [""] , "retry": -1 } |]- job2 <- decodeJob [aesonQQ|+ job2 <-+ decodeJob+ [aesonQQ| { "jid": "abc" , "args": [""] , "retry": -1@@ -91,13 +103,17 @@ it "handles retry 0" $ do now <- getCurrentTime- job1 <- decodeJob [aesonQQ|+ job1 <-+ decodeJob+ [aesonQQ| { "jid": "abc" , "args": [""] , "retry": 0 } |]- job2 <- decodeJob [aesonQQ|+ job2 <-+ decodeJob+ [aesonQQ| { "jid": "abc" , "args": [""] , "retry": 0@@ -113,7 +129,9 @@ it "handles nonsense" $ do now <- getCurrentTime- job <- decodeJob [aesonQQ|+ job <-+ decodeJob+ [aesonQQ| { "jid": "abc" , "args": [""] , "retry": 20@@ -127,7 +145,9 @@ jobRetriesRemaining job `shouldBe` 0 it "handles reserve_for" $ do- job <- decodeJob [aesonQQ|+ job <-+ decodeJob+ [aesonQQ| { "jid": "abc" , "args": [""] , "reserve_for": 3600@@ -144,19 +164,18 @@ jobOptions job `shouldBe` jobtype "Default" it "adds job option defaults from settings" $ do- let settings = defaultSettings { settingsDefaultJobOptions = retry 5 }+ let settings = defaultSettings {settingsDefaultJobOptions = retry 5} bracket (newProducer settings) closeProducer $ \producer -> do job <- buildJob @Text mempty producer "text" jobOptions job `shouldBe` (jobtype "Default" <> retry 5) it "doesn't prefers explicit job options over defaults in settings" $ do- let settings = defaultSettings { settingsDefaultJobOptions = retry 5 }+ let settings = defaultSettings {settingsDefaultJobOptions = retry 5} bracket (newProducer settings) closeProducer $ \producer -> do job <- buildJob @Text (retry 88) producer "text" jobOptions job `shouldBe` (jobtype "Default" <> retry 88)- decodeJob :: Value -> IO (Job Text) decodeJob v = case fromJSON v of
tests/Faktory/Test.hs view
@@ -5,7 +5,7 @@ , workerTestCase , workerTestCaseWith - -- * Lower-level+ -- * Lower-level , withProducer , withWorker , startWorker@@ -15,7 +15,7 @@ import Faktory.Prelude as X -import Control.Monad.IO.Class as X (MonadIO(..))+import Control.Monad.IO.Class as X (MonadIO (..)) import Faktory.Job as X import Faktory.Producer as X import Test.Hspec as X@@ -50,7 +50,7 @@ withWorker editSettings f = do a <- startWorker editSettings result <- f- (, result) <$> haltWorker a+ (,result) <$> haltWorker a startWorker :: HasCallStack => (WorkerSettings -> WorkerSettings) -> IO (Async [Job Text])
tests/FaktorySpec.hs view
@@ -46,7 +46,7 @@ -- the Server and handles it correctly. Setting our own idle delay to 0 -- ensures that we'll pick up the following HALT message immediately. --- let editSettings ws = ws { settingsIdleDelay = 0 }+ let editSettings ws = ws {settingsIdleDelay = 0} jobs <- workerTestCaseWith editSettings $ \_ -> do threadDelay $ 2 * 1000000 + 250000
tests/Spec.hs view
@@ -1,2 +1,2 @@-{-# OPTIONS_GHC -Wno-missing-export-lists #-} {-# OPTIONS_GHC -F -pgmF hspec-discover #-}+{-# OPTIONS_GHC -Wno-missing-export-lists #-}