faktory (empty) → 1.0.0.0
raw patch · 17 files changed
+1439/−0 lines, 17 filesdep +aesondep +aeson-casingdep +base
Dependencies added: aeson, aeson-casing, base, bytestring, connection, cryptonite, faktory, hspec, markdown-unlit, megaparsec, memory, network, random, safe-exceptions, scanner, text, time, unix
Files
- CHANGELOG.md +7/−0
- LICENSE +19/−0
- README.lhs +175/−0
- README.md +175/−0
- examples/consumer/Main.hs +25/−0
- examples/producer/Main.hs +26/−0
- faktory.cabal +152/−0
- library/Faktory/Client.hs +218/−0
- library/Faktory/Connection.hs +92/−0
- library/Faktory/Job.hs +124/−0
- library/Faktory/Prelude.hs +23/−0
- library/Faktory/Protocol.hs +78/−0
- library/Faktory/Settings.hs +67/−0
- library/Faktory/Worker.hs +102/−0
- tests/Faktory/ConnectionSpec.hs +108/−0
- tests/FaktorySpec.hs +46/−0
- tests/Spec.hs +2/−0
+ CHANGELOG.md view
@@ -0,0 +1,7 @@+## [*Unreleased*](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.0.0...master)++None++## [v1.0.0.0](https://github.com/frontrowed/faktory_worker_haskell/tree/v1.0.0.0)++Initial release.
+ LICENSE view
@@ -0,0 +1,19 @@+Copyright (c) 2018 Freckle Education <engineering@freckle.com>++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in+all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN+THE SOFTWARE.
+ README.lhs view
@@ -0,0 +1,175 @@+# faktory\_worker\_haskell++[](https://circleci.com/gh/frontrowed/faktory_worker_haskell)++Haskell client and worker process for the Faktory background job server.++Architecture overview from [Ruby client README](https://github.com/contribsys/faktory_worker_ruby#readme):++```+ +--------------------++ | |+ | Faktory |+ | Server |+ +---------->>>>| +>>>>--------++ | | | |+ | | | |+ | +--------------------+ |++-----------------+ +-------------------++| | | |+| Client | | Worker |+| pushes | | pulls |+| jobs | | jobs |+| | | |+| | | |++-----------------+ +-------------------++```++- Client - an API any process can use to push jobs to the Faktory server.+- Worker - a process that pulls jobs from Faktory and executes them.+- Server - the Faktory daemon which stores background jobs in queues to be+ processed by Workers.++This package contains only the client and worker parts. The server part is+[here](https://github.com/contribsys/faktory/)++## Installation++TODO.++## Documentation++See the [wiki](//github.com/contribsys/faktory_worker_ruby/wiki) for more+details.++## Usage++<!--+```haskell+import Data.Aeson+import Prelude+import Faktory.Client+import Faktory.Job+import Faktory.Settings+import Faktory.Worker+import GHC.Generics++{- Don't actually run anything -}+main :: IO ()+main = if True then pure () else (workerMain >> clientMain)+workerMain :: IO ()+clientMain :: IO ()+```+-->++### Job++Any value can be a "Job" that is pushed and pulled to and from Faktory via its+`ToJSON` and `FromJSON` instances:++```haskell+newtype MyJob = MyJob+ { myJobMessage :: String+ }+ deriving (Generic)++instance ToJSON MyJob+instance FromJSON MyJob+```++### Worker++```haskell+workerMain = do+ settings <- envSettings++ runWorker settings $ \job ->+ -- Process your Job here+ putStrLn $ myJobMessage 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,+ -- unless you catch-and-rethrow them yourself.+```++### Client++```haskell+clientMain = do+ settings <- envSettings+ client <- newClient settings Nothing -- N.B. A WorkerId is not necessary if+ -- only pushing Jobs.++ jobId <- perform mempty client $ MyJob "Hello world"++ print jobId++ closeClient client+```++### Configuration++When using `envSettings`, the following variables will be used:++- `FAKTORY_QUEUE`: the name of the queue to consume from. This is Worker-only,+ for `perform`, a non-default Queue should be given by the `queue` option+- `FAKTORY_PROVIDER`: the name of another environment variable where the+ connection string can be found. Defaults to `FAKTORY_URL`.+- `FAKTORY_URL` (or whatever you named in `FAKTORY_PROVIDER`): connection string+ to the Faktory server. Format is `tcp(+tls)://(:password@)host:port`. Defaults+ to `tcp://localhost:4719`.++## Examples++See the [examples](./examples). To run them:++1. Run a local Faktory server++ ```console+ docker run --rm \+ --publish 7419:7419 \+ --publish 7420:7420 \+ contribsys/faktory+ ```++1. Run the consumer example++ ```console+ % stack exec faktory-example-consumer+ Starting consumer loop+ ```++ (Assumes you've built the project.)++1. Submit a Job through the producer example++ ```console+ % stack exec faktory-example-producer hello world+ Pushed job: "ljcjlbexbgun"+ ```++ *NOTE*: if you submit "BOOM" as a Job, the processing loop will raise an+ exception, so you can see how a Failed Job looks in Faktory.++1. See that your Job was processed back in the consumer++ ```console+ % stack exec faktory-example-consumer+ Starting consumer loop+ hello world+ ```++## Development & Tests++```console+stack build --dependencies-only --test --no-run-tests+stack build --pedantic --test --no-run-tests+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.++---++[CHANGELOG](./CHANGELOG.md) | [LICENSE](./LICENSE)
+ README.md view
@@ -0,0 +1,175 @@+# faktory\_worker\_haskell++[](https://circleci.com/gh/frontrowed/faktory_worker_haskell)++Haskell client and worker process for the Faktory background job server.++Architecture overview from [Ruby client README](https://github.com/contribsys/faktory_worker_ruby#readme):++```+ +--------------------++ | |+ | Faktory |+ | Server |+ +---------->>>>| +>>>>--------++ | | | |+ | | | |+ | +--------------------+ |++-----------------+ +-------------------++| | | |+| Client | | Worker |+| pushes | | pulls |+| jobs | | jobs |+| | | |+| | | |++-----------------+ +-------------------++```++- Client - an API any process can use to push jobs to the Faktory server.+- Worker - a process that pulls jobs from Faktory and executes them.+- Server - the Faktory daemon which stores background jobs in queues to be+ processed by Workers.++This package contains only the client and worker parts. The server part is+[here](https://github.com/contribsys/faktory/)++## Installation++TODO.++## Documentation++See the [wiki](//github.com/contribsys/faktory_worker_ruby/wiki) for more+details.++## Usage++<!--+```haskell+import Data.Aeson+import Prelude+import Faktory.Client+import Faktory.Job+import Faktory.Settings+import Faktory.Worker+import GHC.Generics++{- Don't actually run anything -}+main :: IO ()+main = if True then pure () else (workerMain >> clientMain)+workerMain :: IO ()+clientMain :: IO ()+```+-->++### Job++Any value can be a "Job" that is pushed and pulled to and from Faktory via its+`ToJSON` and `FromJSON` instances:++```haskell+newtype MyJob = MyJob+ { myJobMessage :: String+ }+ deriving (Generic)++instance ToJSON MyJob+instance FromJSON MyJob+```++### Worker++```haskell+workerMain = do+ settings <- envSettings++ runWorker settings $ \job ->+ -- Process your Job here+ putStrLn $ myJobMessage 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,+ -- unless you catch-and-rethrow them yourself.+```++### Client++```haskell+clientMain = do+ settings <- envSettings+ client <- newClient settings Nothing -- N.B. A WorkerId is not necessary if+ -- only pushing Jobs.++ jobId <- perform mempty client $ MyJob "Hello world"++ print jobId++ closeClient client+```++### Configuration++When using `envSettings`, the following variables will be used:++- `FAKTORY_QUEUE`: the name of the queue to consume from. This is Worker-only,+ for `perform`, a non-default Queue should be given by the `queue` option+- `FAKTORY_PROVIDER`: the name of another environment variable where the+ connection string can be found. Defaults to `FAKTORY_URL`.+- `FAKTORY_URL` (or whatever you named in `FAKTORY_PROVIDER`): connection string+ to the Faktory server. Format is `tcp(+tls)://(:password@)host:port`. Defaults+ to `tcp://localhost:4719`.++## Examples++See the [examples](./examples). To run them:++1. Run a local Faktory server++ ```console+ docker run --rm \+ --publish 7419:7419 \+ --publish 7420:7420 \+ contribsys/faktory+ ```++1. Run the consumer example++ ```console+ % stack exec faktory-example-consumer+ Starting consumer loop+ ```++ (Assumes you've built the project.)++1. Submit a Job through the producer example++ ```console+ % stack exec faktory-example-producer hello world+ Pushed job: "ljcjlbexbgun"+ ```++ *NOTE*: if you submit "BOOM" as a Job, the processing loop will raise an+ exception, so you can see how a Failed Job looks in Faktory.++1. See that your Job was processed back in the consumer++ ```console+ % stack exec faktory-example-consumer+ Starting consumer loop+ hello world+ ```++## Development & Tests++```console+stack build --dependencies-only --test --no-run-tests+stack build --pedantic --test --no-run-tests+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.++---++[CHANGELOG](./CHANGELOG.md) | [LICENSE](./LICENSE)
+ examples/consumer/Main.hs view
@@ -0,0 +1,25 @@+module Main (main) where++import Prelude++import Control.Exception.Safe+import Data.Aeson+import Faktory.Settings+import Faktory.Worker+import GHC.Generics++-- | Must match examples/producer+newtype Job = Job { jobMessage :: String }+ deriving Generic+instance FromJSON Job++main :: IO ()+main = do+ putStrLn "Starting consumer loop"+ settings <- envSettings+ runWorker settings $ \job -> do+ let message = jobMessage job++ if message == "BOOM"+ then throwString "Producer exception: BOOM"+ else putStrLn message
+ examples/producer/Main.hs view
@@ -0,0 +1,26 @@+module Main (main) where++import Prelude++import Control.Exception.Safe+import Data.Aeson+import Data.Semigroup ((<>))+import Faktory.Client+import Faktory.Job (perform)+import Faktory.Settings+import GHC.Generics+import System.Environment (getArgs)++-- | Must match examples/consumer+newtype Job = Job { jobMessage :: String }+ deriving Generic+instance ToJSON Job++main :: IO ()+main = do+ settings <- envSettings+ bracket (newClient settings Nothing) closeClient $ \client -> do+ args <- getArgs+ jobId <- perform mempty client Job {jobMessage = unwords args}++ putStrLn $ "Pushed job: " <> show jobId
+ faktory.cabal view
@@ -0,0 +1,152 @@+cabal-version: 1.18++-- This file has been generated from package.yaml by hpack version 0.31.1.+--+-- see: https://github.com/sol/hpack+--+-- hash: 5b58d182a42e13721957b2b5e0234d010b79698c8ed5197f94a5a14d5114964c++name: faktory+version: 1.0.0.0+synopsis: Faktory Worker for Haskell+description: Haskell client and worker process for the Faktory background job server.+ .+ == Architecture overview+ .+ @+ | +--------------------++ | | |+ | | Faktory |+ | | Server |+ | +---------->>>>| +>>>>--------++ | | | | |+ | | | | |+ | | +--------------------+ |+ | +-----------------+ +-------------------++ | | | | |+ | | Client | | Worker |+ | | pushes | | pulls |+ | | jobs | | jobs |+ | | | | |+ | | | | |+ | +-----------------+ +-------------------++ @+ .+ * `Client` - an API any process can use to push jobs to the Faktory server.+ * `Worker` - a process that pulls jobs from Faktory and executes them.+ * `Server` - the Faktory daemon which stores background jobs in queues to be processed by Workers.+ .+ This package contains only the `Client` and `Worker`.+category: Network+homepage: https://github.com/frontrowed/faktory_worker_haskell#readme+bug-reports: https://github.com/frontrowed/faktory_worker_haskell/issues+author: Freckle Engineering+maintainer: engineering@freckle.com+copyright: 2018 Freckle Education+license: MIT+license-file: LICENSE+tested-with: GHC==8.4.3+build-type: Simple+extra-doc-files:+ CHANGELOG.md+ README.md++source-repository head+ type: git+ location: https://github.com/frontrowed/faktory_worker_haskell++library+ exposed-modules:+ Faktory.Client+ Faktory.Connection+ Faktory.Job+ Faktory.Prelude+ Faktory.Protocol+ Faktory.Settings+ Faktory.Worker+ other-modules:+ Paths_faktory+ hs-source-dirs:+ library+ default-extensions: BangPatterns DeriveAnyClass DeriveFoldable DeriveFunctor DeriveGeneric DeriveLift DeriveTraversable DerivingStrategies FlexibleContexts FlexibleInstances GADTs GeneralizedNewtypeDeriving LambdaCase MultiParamTypeClasses NoImplicitPrelude NoMonomorphismRestriction OverloadedStrings RankNTypes RecordWildCards ScopedTypeVariables StandaloneDeriving TypeApplications TypeFamilies+ ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-monomorphism-restriction -Wno-missing-import-lists+ build-depends:+ aeson >=1.3 && <2+ , aeson-casing >=0.1 && <1+ , base >=4.11 && <5+ , bytestring >=0.1 && <1+ , connection >=0.2 && <1+ , cryptonite >=0.2 && <1+ , megaparsec >=6.5 && <7+ , memory >=0.1 && <1+ , network >=2.6 && <3+ , random >=1.1 && <2+ , safe-exceptions >=0.1 && <1+ , scanner >=0.2 && <1+ , text >=1.2 && <2+ , time >=1.8 && <2+ , unix >=2.7 && <3+ default-language: Haskell2010++executable faktory-example-consumer+ main-is: Main.hs+ other-modules:+ Paths_faktory+ hs-source-dirs:+ examples/consumer+ default-extensions: BangPatterns DeriveAnyClass DeriveFoldable DeriveFunctor DeriveGeneric DeriveLift DeriveTraversable DerivingStrategies FlexibleContexts FlexibleInstances GADTs GeneralizedNewtypeDeriving LambdaCase MultiParamTypeClasses NoImplicitPrelude NoMonomorphismRestriction OverloadedStrings RankNTypes RecordWildCards ScopedTypeVariables StandaloneDeriving TypeApplications TypeFamilies+ ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-monomorphism-restriction -Wno-missing-import-lists+ build-depends:+ aeson+ , base >=4.11 && <5+ , faktory+ , safe-exceptions+ default-language: Haskell2010++executable faktory-example-producer+ main-is: Main.hs+ other-modules:+ Paths_faktory+ hs-source-dirs:+ examples/producer+ default-extensions: BangPatterns DeriveAnyClass DeriveFoldable DeriveFunctor DeriveGeneric DeriveLift DeriveTraversable DerivingStrategies FlexibleContexts FlexibleInstances GADTs GeneralizedNewtypeDeriving LambdaCase MultiParamTypeClasses NoImplicitPrelude NoMonomorphismRestriction OverloadedStrings RankNTypes RecordWildCards ScopedTypeVariables StandaloneDeriving TypeApplications TypeFamilies+ ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-monomorphism-restriction -Wno-missing-import-lists+ build-depends:+ aeson+ , base >=4.11 && <5+ , faktory+ , safe-exceptions+ default-language: Haskell2010++test-suite hspec+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ other-modules:+ Faktory.ConnectionSpec+ FaktorySpec+ Paths_faktory+ hs-source-dirs:+ tests+ default-extensions: BangPatterns DeriveAnyClass DeriveFoldable DeriveFunctor DeriveGeneric DeriveLift DeriveTraversable DerivingStrategies FlexibleContexts FlexibleInstances GADTs GeneralizedNewtypeDeriving LambdaCase MultiParamTypeClasses NoImplicitPrelude NoMonomorphismRestriction OverloadedStrings RankNTypes RecordWildCards ScopedTypeVariables StandaloneDeriving TypeApplications TypeFamilies+ ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-monomorphism-restriction -Wno-missing-import-lists -rtsopts+ build-depends:+ base >=4.11 && <5+ , faktory+ , hspec+ default-language: Haskell2010++test-suite readme+ type: exitcode-stdio-1.0+ main-is: README.lhs+ other-modules:+ Paths_faktory+ hs-source-dirs:+ ./.+ default-extensions: BangPatterns DeriveAnyClass DeriveFoldable DeriveFunctor DeriveGeneric DeriveLift DeriveTraversable DerivingStrategies FlexibleContexts FlexibleInstances GADTs GeneralizedNewtypeDeriving LambdaCase MultiParamTypeClasses NoImplicitPrelude NoMonomorphismRestriction OverloadedStrings RankNTypes RecordWildCards ScopedTypeVariables StandaloneDeriving TypeApplications TypeFamilies+ ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-monomorphism-restriction -Wno-missing-import-lists -pgmL markdown-unlit+ build-depends:+ aeson+ , base+ , faktory+ , markdown-unlit+ default-language: Haskell2010
+ library/Faktory/Client.hs view
@@ -0,0 +1,218 @@+module Faktory.Client+ (+ -- * Client operations+ Client+ , newClient+ , closeClient++ -- * High-level Job operations+ , pushJob+ , flush++ -- * High-level Client API+ , command_+ , commandOK+ , commandJSON+ ) where++import Faktory.Prelude++import Control.Concurrent.MVar+import Crypto.Hash (Digest, SHA256(..), hashWith)+import Data.Aeson+import Data.ByteArray (ByteArrayAccess)+import Data.ByteString.Lazy (ByteString, fromStrict)+import qualified Data.ByteString.Lazy.Char8 as BSL8+import Data.Text (Text)+import qualified Data.Text as T+import qualified Data.Text.Encoding as T+import Faktory.Connection (connect)+import Faktory.Protocol+import Faktory.Settings+import GHC.Stack+import Network.Connection+import Network.Socket (HostName)+import System.Posix.Process (getProcessID)++data Client = Client+ { clientConnection :: MVar Connection+ , clientSettings :: Settings+ }++-- | <https://github.com/contribsys/faktory/wiki/Worker-Lifecycle#initial-handshake>+data HiPayload = HiPayload+ { hiVersion :: Int+ , hiNonce :: Maybe Text+ , hiIterations :: Maybe Int+ }++instance FromJSON HiPayload where+ parseJSON = withObject "HiPayload" $ \o ->+ HiPayload+ <$> o .: "v"+ <*> o .:? "s"+ <*> o .:? "i"++data HelloPayload = HelloPayload+ { helloWorkerId :: Maybe WorkerId+ , helloHostname :: HostName+ , helloProcessId :: Integer -- TODO: Orphan ToJSON ProcessID+ , helloLabels :: [Text]+ , helloVersion :: Int+ , helloPasswordHash :: Maybe Text+ }++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+ ]++-- | Open a new @'Client'@ connection with the given @'Settings'@+newClient :: HasCallStack => Settings -> Maybe WorkerId -> IO Client+newClient settings@Settings {..} mWorkerId =+ bracketOnError (connect settingsConnection) connectionClose $ \conn -> do+ client <- Client <$> newMVar conn <*> pure settings++ greeting <-+ fromJustThrows "Unexpected end of HI message" =<< recvUnsafe settings conn+ stripped <-+ fromJustThrows ("Missing HI prefix: " <> show greeting)+ $ BSL8.stripPrefix "HI" greeting+ HiPayload {..} <-+ 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+ ]++ let+ mPassword = connectionInfoPassword settingsConnection+ mHashedPassword = hashPassword <$> hiNonce <*> hiIterations <*> mPassword++ helloPayload <-+ HelloPayload mWorkerId (show . fst $ connectionID conn)+ <$> (toInteger <$> getProcessID)+ <*> pure ["haskell"]+ <*> pure expectedProtocolVersion+ <*> pure mHashedPassword++ commandOK client "HELLO" [encode helloPayload]+ pure client+ where fromJustThrows message = maybe (throwString message) pure++-- | Close a @'Client'@+closeClient :: Client -> IO ()+closeClient Client {..} = withMVar clientConnection $ \conn -> do+ sendUnsafe clientSettings conn "END" []+ connectionClose conn++-- | Push a Job to the Server+pushJob :: (HasCallStack, ToJSON a) => Client -> a -> IO ()+pushJob client job = commandOK client "PUSH" [encode job]++-- | Clear all job data in the Faktory server+--+-- Use with caution!+--+flush :: HasCallStack => Client -> IO ()+flush client = commandOK client "FLUSH" []++-- | 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 $ recvUnsafe clientSettings conn++-- | 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+ unless (response == Just "OK") $ throwString "Server not OK"++-- | Send a command, parse the response as JSON+commandJSON+ :: FromJSON a+ => Client+ -> ByteString+ -> [ByteString]+ -> IO (Either String (Maybe a))+commandJSON Client {..} cmd args = withMVar clientConnection $ \conn -> do+ sendUnsafe clientSettings conn cmd args+ mByteString <- recvUnsafe clientSettings conn+ pure $ traverse eitherDecode mByteString++-- | 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)+ settingsLogDebug $ "> " <> show bs+ void . connectionPut conn . BSL8.toStrict $ bs <> "\n"++-- | Receive data from the Server socket+--+-- Do not use outside of @'withMVar'@, this is not threadsafe.+--+recvUnsafe :: Settings -> Connection -> IO (Maybe ByteString)+recvUnsafe Settings {..} conn = do+ eByteString <- readReply $ connectionGet conn 4096+ settingsLogDebug $ "< " <> show eByteString++ case eByteString of+ Left err -> do+ settingsLogError err+ pure Nothing+ Right mByteString -> pure $ fromStrict <$> mByteString++-- | 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+ | otherwise = times (n - 1) f (f s)++-- | Hash password using provided @nonce@ for @n@ iterations+hashPassword :: Text -> Int -> String -> Text+hashPassword nonce n password =+ T.pack+ . show+ . times (n - 1) hash+ . hash+ . T.encodeUtf8+ $ T.pack password+ <> 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 = hashWith SHA256++-- | Protocol version the client expects+expectedProtocolVersion :: Int+expectedProtocolVersion = 2
+ library/Faktory/Connection.hs view
@@ -0,0 +1,92 @@+module Faktory.Connection+ ( ConnectionInfo(..)+ , defaultConnectionInfo+ , envConnectionInfo+ , connect+ ) where++import Faktory.Prelude++import Data.Maybe (fromMaybe)+import Data.Void+import Network.Connection+import Network.Socket (HostName, PortNumber)+import System.Environment (lookupEnv)+import Text.Megaparsec+import Text.Megaparsec.Char++data ConnectionInfo = ConnectionInfo+ { connectionInfoTls :: Bool+ , connectionInfoPassword :: Maybe String+ , connectionInfoHostName :: HostName+ , connectionInfoPort :: PortNumber+ }+ deriving (Eq, Show)++defaultConnectionInfo :: ConnectionInfo+defaultConnectionInfo = ConnectionInfo+ { connectionInfoTls = False+ , connectionInfoPassword = Nothing+ , connectionInfoHostName = "localhost"+ , connectionInfoPort = 7419+ }++-- | Parse a @'Connection'@ from environment variables+--+-- > FAKTORY_PROVIDER=FAKTORY_URL+-- > FAKTORY_URL=tcp://:my-password@localhost:7419+--+-- Supported format is @tcp(+tls):\/\/(:password@)host:port@.+--+-- See <https://github.com/contribsys/faktory/wiki/Worker-Lifecycle#url-configuration>.+--+envConnectionInfo :: IO ConnectionInfo+envConnectionInfo = do+ providerString <- fromMaybe "FAKTORY_URL" <$> lookupEnv "FAKTORY_PROVIDER"+ provider <- parseThrow parseProvider "FAKTORY_PROVIDER" providerString+ connectionString <- fromMaybe "tcp://localhost:7419" <$> lookupEnv provider+ parseThrow parseConnection provider connectionString++-- | Connect to the given @'Connection'@ as a @'Socket'@+connect :: ConnectionInfo -> IO Connection+connect ConnectionInfo {..} = bracketOnError open connectionClose pure+ 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+ }++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 <> ":"+ , parseErrorPretty ex+ ]++parseProvider :: Parser String+parseProvider =+ some (upperChar <|> char '_') <?> "an environment variable name"++parseConnection :: Parser ConnectionInfo+parseConnection = go <?> "tcp(+tls)://(:<password>@)<host>:<port>"+ where+ go =+ ConnectionInfo+ <$> (False <$ string "tcp://" <|> True <$ string "tcp+tls://")+ <*> optional (char ':' *> manyTill anyChar (char '@'))+ <*> manyTill anyChar (char ':')+ <*> (read <$> some digitChar)
+ library/Faktory/Job.hs view
@@ -0,0 +1,124 @@+module Faktory.Job+ ( Job+ , JobId+ , perform+ , retry+ , once+ , queue+ , jobtype+ , at+ , in_+ , newJob+ , jobJid+ , jobArg+ ) where++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 Faktory.Client (Client, pushJob)+import Faktory.Settings (Queue)+import GHC.Generics+import GHC.Stack+import System.Random++data Job arg = Job+ { jobJid :: JobId+ , jobJobtype :: String+ , 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+ }+ deriving Generic++-- | Individual changes to a @'Job'@ to be 'perform'ed+data JobUpdate+ = SetRetry Int+ | SetQueue Queue+ | SetJobtype String+ | SetAt UTCTime+ | SetIn NominalDiffTime++newtype JobOptions = JobOptions [JobUpdate]+ deriving newtype (Semigroup, Monoid)++-- | Perform a Job with the given options+--+-- @+-- 'perform' 'mempty' SomeJob+-- 'perform' ('queue' "SomeQueue") SomeJob+-- 'perform' 'once' SomeJob+-- 'perform' ('at' someTime <> 'once') SomeJob+-- 'perform' ('in_' 10 <> 'once') SomeJob+-- 'perform' ('in_' 10 <> 'retry' 3) SomeJob+-- @+--+perform :: (HasCallStack, ToJSON arg) => JobOptions -> Client -> arg -> IO JobId+perform options client arg = do+ job <- applyOptions options =<< newJob arg+ jobJid job <$ pushJob client job++applyOptions :: JobOptions -> Job arg -> IO (Job arg)+applyOptions (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 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]++in_ :: NominalDiffTime -> JobOptions+in_ i = JobOptions [SetIn i]++newJob :: arg -> IO (Job arg)+newJob arg = do+ -- Ruby uses 12 random hex+ jobId <- take 12 . randomRs ('a', 'z') <$> newStdGen++ pure Job+ { jobJid = jobId+ , jobJobtype = "Default"+ , jobArgs = pure arg+ , jobRetry = Nothing+ , jobQueue = Nothing+ , jobAt = Nothing+ }++jobArg :: Job arg -> arg+jobArg Job {..} = NE.head jobArgs++instance ToJSON args => ToJSON (Job args) where+ toJSON = genericToJSON $ aesonPrefix snakeCase+ toEncoding = genericToEncoding $ aesonPrefix snakeCase++instance FromJSON args => FromJSON (Job args) where+ parseJSON = genericParseJSON $ aesonPrefix snakeCase++type JobId = String
+ library/Faktory/Prelude.hs view
@@ -0,0 +1,23 @@+module Faktory.Prelude+ ( module X+ , module Faktory.Prelude+ )+where++import Prelude as X++import Control.Concurrent (ThreadId, forkIO, myThreadId, threadDelay)+import Control.Exception.Safe as X+import Control.Monad as X+import Data.Foldable as X+import Data.Semigroup as X ((<>))+import Data.Text as X (Text, pack, unpack)+import Data.Traversable as X++threadDelaySeconds :: Int -> IO ()+threadDelaySeconds n = threadDelay $ n * 1000000++forkIOWithThrowToParent :: IO () -> IO ThreadId+forkIOWithThrowToParent action = do+ parent <- myThreadId+ forkIO $ action `X.catchAny` \err -> throwTo parent err
+ library/Faktory/Protocol.hs view
@@ -0,0 +1,78 @@+-- | Modified version of @"Database.Redis.Protocol"@+--+-- <https://github.com/informatikr/hedis/blob/master/src/Database/Redis/Protocol.hs>+--+-- 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+ ) where++import Prelude hiding (error)++import Data.ByteString (ByteString)+import qualified Data.ByteString.Char8 as BS8+import qualified Data.Text.Encoding as T+import qualified Data.Text.Read as T+import Scanner (Scanner)+import qualified Scanner++data Reply+ = SingleLine ByteString+ | Error ByteString+ | Bulk (Maybe ByteString)++readReply :: IO ByteString -> IO (Either String (Maybe ByteString))+readReply getMore = fromScanResult <$> Scanner.scanWith getMore reply BS8.empty+ where+ fromScanResult = \case+ Scanner.Fail _ msg -> Left msg+ Scanner.More _ -> Left "Unexpected More"+ Scanner.Done _ (SingleLine bs) -> Right $ Just bs+ Scanner.Done _ (Error bs) -> Left $ BS8.unpack bs+ Scanner.Done _ (Bulk mByteString) -> Right mByteString++{-# INLINE reply #-}+reply :: Scanner Reply+reply = do+ c <- Scanner.anyChar8+ case c of+ '+' -> string+ '-' -> error+ '$' -> bulk+ _ -> fail "Unknown reply type"++{-# INLINE string #-}+string :: Scanner Reply+string = SingleLine <$> line++{-# INLINE error #-}+error :: Scanner Reply+error = Error <$> line++{-# INLINE bulk #-}+bulk :: Scanner Reply+bulk = Bulk <$> do+ len <- integral+ if len < 0 then return Nothing else Just <$> Scanner.take len <* eol++{-# INLINE integral #-}+integral :: Integral i => Scanner i+integral = do+ str <- line+ case T.signed T.decimal (T.decodeUtf8 str) of+ Left err -> fail (show err)+ Right (l, _) -> return l++{-# INLINE line #-}+line :: Scanner ByteString+line = Scanner.takeWhileChar8 (/= '\r') <* eol++{-# INLINE eol #-}+eol :: Scanner ()+eol = do+ Scanner.char8 '\r'+ Scanner.char8 '\n'
+ library/Faktory/Settings.hs view
@@ -0,0 +1,67 @@+module Faktory.Settings+ ( Settings(..)+ , ConnectionInfo(..)+ , defaultSettings+ , envSettings+ , Queue(..)+ , queueArg+ , defaultQueue+ , WorkerId+ , randomWorkerId+ ) where++import Faktory.Prelude++import Data.Aeson+import Data.ByteString.Lazy (ByteString, fromStrict)+import Data.String+import Data.Text.Encoding (encodeUtf8)+import Faktory.Connection+import System.Environment (lookupEnv)+import System.IO (hPutStrLn, stderr)+import System.Random++data Settings = Settings+ { settingsQueue :: Queue+ , settingsConnection :: ConnectionInfo+ , settingsLogDebug :: String -> IO ()+ , settingsLogError :: String -> IO ()+ , settingsWorkerIdleDelay :: Int+ }++defaultSettings :: Settings+defaultSettings = Settings+ { settingsQueue = defaultQueue+ , settingsConnection = defaultConnectionInfo+ , settingsLogDebug = \_msg -> pure ()+ , settingsLogError = hPutStrLn stderr . ("[ERROR]: " <>)+ , settingsWorkerIdleDelay = 1+ }++-- | Defaults, but read @'Connection'@ from the environment+--+-- See @'envConnection'@+--+envSettings :: IO Settings+envSettings = do+ mQueue <- lookupEnv "FAKTORY_QUEUE"+ connection <- envConnectionInfo+ pure defaultSettings+ { settingsQueue = maybe defaultQueue (Queue . pack) mQueue+ , settingsConnection = connection+ }++newtype Queue = Queue Text+ deriving newtype (IsString, FromJSON, ToJSON)++queueArg :: Queue -> ByteString+queueArg (Queue q) = fromStrict $ encodeUtf8 q++defaultQueue :: Queue+defaultQueue = "default"++newtype WorkerId = WorkerId String+ deriving newtype (FromJSON, ToJSON)++randomWorkerId :: IO WorkerId+randomWorkerId = WorkerId . take 8 . randomRs ('a', 'z') <$> newStdGen
+ library/Faktory/Worker.hs view
@@ -0,0 +1,102 @@+-- | High-level interface for a Worker+--+-- Runs forever, @FETCH@-ing Jobs from the given Queue and handing each to your+-- processing function.+--+module Faktory.Worker+ ( WorkerHalt(..)+ , runWorker+ )+where++import Faktory.Prelude++import Control.Concurrent (killThread)+import Data.Aeson+import Data.Aeson.Casing+import qualified Data.Text as T+import Faktory.Client+import Faktory.Job (Job, JobId, jobArg, jobJid)+import Faktory.Settings+import GHC.Generics+import GHC.Stack++-- | If processing functions @'throw'@ this, @'runWorker'@ will exit+data WorkerHalt = WorkerHalt+ deriving (Eq, Show, Exception)++newtype BeatPayload = BeatPayload+ { _bpWid :: WorkerId+ }+ deriving Generic++instance ToJSON BeatPayload where+ toJSON = genericToJSON $ aesonPrefix snakeCase+ toEncoding = genericToEncoding $ aesonPrefix snakeCase++newtype AckPayload = AckPayload+ { _apJid :: JobId+ }+ deriving Generic++instance ToJSON AckPayload where+ toJSON = genericToJSON $ aesonPrefix snakeCase+ toEncoding = genericToEncoding $ aesonPrefix snakeCase++data FailPayload = FailPayload+ { _fpMessage :: Text+ , _fpErrtype :: String+ , _fpJid :: JobId+ , _fpBacktrace :: [String]+ }+ deriving Generic++instance ToJSON FailPayload where+ toJSON = genericToJSON $ aesonPrefix snakeCase+ toEncoding = genericToEncoding $ aesonPrefix snakeCase++runWorker :: FromJSON args => Settings -> (args -> IO ()) -> IO ()+runWorker settings f = do+ workerId <- randomWorkerId+ client <- newClient settings $ Just workerId+ beatThreadId <- forkIOWithThrowToParent $ forever $ heartBeat client workerId++ forever (processorLoop client settings f)+ `catch` (\(_ex :: WorkerHalt) -> pure ())+ `finally` (killThread beatThreadId >> closeClient client)++processorLoop :: FromJSON arg => Client -> Settings -> (arg -> IO ()) -> IO ()+processorLoop client settings f = do+ let+ processAndAck job = do+ f $ jobArg job+ ackJob client job++ emJob <- fetchJob client $ settingsQueue settings++ case emJob of+ Left err -> settingsLogError settings $ "Invalid Job: " <> err+ Right Nothing -> threadDelaySeconds $ settingsWorkerIdleDelay settings+ Right (Just job) ->+ processAndAck job+ `catches` [ Handler $ \(ex :: WorkerHalt) -> throw ex+ , Handler $ \(ex :: SomeException) ->+ failJob client job $ T.pack $ show ex+ ]++-- | <https://github.com/contribsys/faktory/wiki/Worker-Lifecycle#heartbeat>+heartBeat :: Client -> WorkerId -> IO ()+heartBeat client workerId = do+ threadDelaySeconds 25+ command_ client "BEAT" [encode $ BeatPayload workerId]++fetchJob+ :: FromJSON args => Client -> Queue -> IO (Either String (Maybe (Job args)))+fetchJob client queue = commandJSON client "FETCH" [queueArg queue]++ackJob :: HasCallStack => Client -> Job args -> IO ()+ackJob client job = commandOK client "ACK" [encode $ AckPayload $ jobJid job]++failJob :: HasCallStack => Client -> Job args -> Text -> IO ()+failJob client job message =+ commandOK client "FAIL" [encode $ FailPayload message "" (jobJid job) []]
+ tests/Faktory/ConnectionSpec.hs view
@@ -0,0 +1,108 @@+module Faktory.ConnectionSpec+ ( spec+ ) where++import Faktory.Prelude++import Data.List (isInfixOf)+import Faktory.Connection+import System.Environment+import Test.Hspec++spec :: Spec+spec = do+ describe "envConnectionInfo" $ do+ it "returns default settings in an empty environment" $ do+ let env = [("FAKTORY_PROVIDER", Nothing), ("FAKTORY_URL", Nothing)]++ withEnvironment env $ do+ connection <- envConnectionInfo+ connection `shouldBe` defaultConnectionInfo++ it "parses the provided URL" $ do+ let+ env =+ [("FAKTORY_PROVIDER", Nothing), ("FAKTORY_URL", Just "tcp://foo:123")]++ withEnvironment env $ do+ ConnectionInfo {..} <- envConnectionInfo+ connectionInfoTls `shouldBe` False+ connectionInfoPassword `shouldBe` Nothing+ connectionInfoHostName `shouldBe` "foo"+ connectionInfoPort `shouldBe` 123++ it "parses tls and password" $ do+ let+ env =+ [ ("FAKTORY_PROVIDER", Nothing)+ , ("FAKTORY_URL", Just "tcp+tls://:foo@bar:123")+ ]++ withEnvironment env $ do+ ConnectionInfo {..} <- envConnectionInfo+ connectionInfoTls `shouldBe` True+ connectionInfoPassword `shouldBe` Just "foo"+ connectionInfoHostName `shouldBe` "bar"+ connectionInfoPort `shouldBe` 123++ it "follows _PROVIDER to find _URL" $ do+ let+ env =+ [ ("FAKTORY_PROVIDER", Just "OTHER_URL")+ , ("OTHER_URL", Just "tcp://foo:123")+ ]++ withEnvironment env $ do+ ConnectionInfo {..} <- envConnectionInfo+ connectionInfoHostName `shouldBe` "foo"+ connectionInfoPort `shouldBe` 123++ it "throws nice errors for invalid PROVIDER" $ do+ 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")+ ]++ withEnvironment env envConnectionInfo+ `shouldThrowMessage` "expecting tcp(+tls)://(:<password>@)<host>:<port>"++ it "throws nice errors for missing _PROVIDER" $ do+ pendingWith "This makes implementation more complicated"++ let+ env =+ [("FAKTORY_PROVIDER", Just "MISSING_URL"), ("MISSING_URL", Nothing)]++ withEnvironment env envConnectionInfo `shouldThrowMessage` "..."++-- | Override ENV with the given values, and restore them after+--+-- 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)+ where+ readAndReset :: (String, Maybe String) -> IO (String, Maybe String)+ readAndReset (variable, mNewValue) = do+ mOriginalValue <- lookupEnv variable+ maybe (unsetEnv variable) (setEnv variable) mNewValue+ pure (variable, mOriginalValue)++shouldThrowMessage :: IO a -> String -> Expectation+shouldThrowMessage act msg =+ act `shouldThrow` \ex -> msg `isInfixOf` show @SomeException ex
+ tests/FaktorySpec.hs view
@@ -0,0 +1,46 @@+module FaktorySpec+ ( spec+ ) where++import Faktory.Prelude++import Control.Concurrent.MVar+import Faktory.Client+import Faktory.Job+import Faktory.Settings+import Faktory.Worker+import Test.Hspec++spec :: Spec+spec = describe "Faktory" $ do+ it "can push and process jobs" $ do+ settings <- envSettings+ bracket (newClient settings Nothing) closeClient $ \client -> do+ void $ flush client+ void $ perform @Text mempty client "a"+ void $ perform @Text mempty client "b"+ void $ perform @Text mempty client "HALT"++ processedJobs <- newMVar ([] :: [Text])+ runWorker settings $ \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+ settings <- envSettings+ bracket (newClient settings Nothing) closeClient $ \client -> do+ void $ flush client+ void $ perform @Text once client "a"+ void $ perform @Text (retry 0) client "b"+ void $ perform @Text mempty client "HALT"++ processedJobs <- newMVar ([] :: [Text])+ runWorker settings $ \job -> do+ modifyMVar_ processedJobs $ pure . (job :)+ when (job == "HALT") $ throw WorkerHalt++ jobs <- readMVar processedJobs+ jobs `shouldMatchList` ["a", "b", "HALT"]
+ tests/Spec.hs view
@@ -0,0 +1,2 @@+{-# OPTIONS_GHC -Wno-missing-export-lists #-}+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}