faktory 1.0.1.2 → 1.0.1.3

raw patch · 14 files changed

+211/−114 lines, 14 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• CHANGELOG.md +5/−1
• LICENSE +7/−5
• README.lhs +26/−24
• examples/consumer/Main.hs +1/−3
• examples/producer/Main.hs +5/−8
• faktory.cabal +4/−3
• library/Faktory/Client.hs +1/−16
• library/Faktory/Connection.hs +10/−2
• library/Faktory/Job.hs +18/−9
• library/Faktory/Producer.hs +40/−0
• library/Faktory/Settings.hs +34/−10
• library/Faktory/Worker.hs +24/−8
• tests/Faktory/ConnectionSpec.hs +14/−0
• tests/FaktorySpec.hs +22/−25

@@ -1,6 +1,10 @@-## [*Unreleased*](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.1.1...master)+## [*Unreleased*](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.1.3...master)  None++## [v1.0.1.3](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.1.2...v1.0.1.3)++- Add support for queue namespacing  ## [v1.0.1.2](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.1.1...v1.0.1.2) 

@@ -1,5 +1,7 @@-Copyright (c) 2018 Freckle Education <engineering@freckle.com>+The MIT License (MIT) +Copyright (c) 2020 Renaissance Learning Inc+ 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@@ -7,13 +9,13 @@ 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 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.+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE+SOFTWARE.

@@ -49,17 +49,16 @@ ```haskell import Data.Aeson import Prelude-import Faktory.Client+import Faktory.Producer 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)+main = if True then pure () else (workerMain >> producerMain) workerMain :: IO ()-clientMain :: IO ()+producerMain :: IO () ``` --> @@ -79,44 +78,47 @@ ### Worker  ```haskell-workerMain = do-  settings <- envSettings--  runWorker settings $ \job ->-    -- Process your Job here-    putStrLn $ myJobMessage job+workerMain = runWorkerEnv $ \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.+  -- 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+### Producer +`Producer` wraps `Client` for push-only usage.+ ```haskell-clientMain = do-  settings <- envSettings-  client <- newClient settings Nothing -- N.B. A WorkerId is not necessary if-                                       -- only pushing Jobs.+producerMain = do+  producer <- newProducerEnv -  jobId <- perform mempty client $ MyJob "Hello world"+  jobId <- perform mempty producer $ MyJob "Hello world"    print jobId -  closeClient client+  closeProducer producer ```  ### 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`.+  to the Faktory server. Format is+  `tcp(+tls)://(:password@)host:port(/namespace)`. Defaults to+  `tcp://localhost:4719`. `namespace` is prependend to queue names on job+  submission and worker consumption.++When using `envWorkerSettings`, the following variables are also used:++- `FAKTORY_QUEUE`: the name of the queue to consume from. Default is "default".+- `FAKTORY_WORKER_ID`: the Id to use for this Worker. Default is to assign a+  random one.  ## Examples 

@@ -4,7 +4,6 @@  import Control.Exception.Safe import Data.Aeson-import Faktory.Settings import Faktory.Worker import GHC.Generics @@ -16,8 +15,7 @@ main :: IO () main = do   putStrLn "Starting consumer loop"-  settings <- envSettings-  runWorker settings $ \job -> do+  runWorkerEnv $ \job -> do     let message = jobMessage job      if message == "BOOM"

@@ -4,9 +4,8 @@  import Control.Exception.Safe import Data.Aeson-import Faktory.Client import Faktory.Job (perform)-import Faktory.Settings+import Faktory.Producer import GHC.Generics import System.Environment (getArgs) @@ -16,10 +15,8 @@   deriving anyclass ToJSON  main :: IO ()-main = do-  settings <- envSettings-  bracket (newClient settings Nothing) closeClient $ \client -> do-    args <- getArgs-    jobId <- perform mempty client Job { jobMessage = unwords args }+main = bracket newProducerEnv closeProducer $ \producer -> do+  args <- getArgs+  jobId <- perform mempty producer Job { jobMessage = unwords args } -    putStrLn $ "Pushed job: " <> show jobId+  putStrLn $ "Pushed job: " <> show jobId

@@ -1,13 +1,13 @@ cabal-version: 1.18 --- This file has been generated from package.yaml by hpack version 0.31.2.+-- This file has been generated from package.yaml by hpack version 0.33.0. -- -- see: https://github.com/sol/hpack ----- hash: 87c1de8469f9bffe9c0e5c342f2944a18b4725f0c07a15b3ff4bbcf88f056313+-- hash: e458b2e214999b21de0c526b70663c2995eb9f4f8fd87fd70757e1ac8bd087c7  name:           faktory-version:        1.0.1.2+version:        1.0.1.3 synopsis:       Faktory Worker for Haskell description:    Haskell client and worker process for the Faktory background job server.                 .@@ -60,6 +60,7 @@       Faktory.Connection       Faktory.Job       Faktory.Prelude+      Faktory.Producer       Faktory.Protocol       Faktory.Settings       Faktory.Worker

@@ -1,14 +1,10 @@ module Faktory.Client   (   -- * Client operations-    Client+    Client(..)   , newClient   , closeClient -  -- * High-level Job operations-  , pushJob-  , flush-   -- * High-level Client API   , command_   , commandOK@@ -121,17 +117,6 @@ 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 ()

@@ -1,5 +1,6 @@ module Faktory.Connection   ( ConnectionInfo(..)+  , Namespace(..)   , defaultConnectionInfo   , envConnectionInfo   , connect@@ -25,11 +26,15 @@   ) import Text.Megaparsec.Char (char, digitChar, string, upperChar) +newtype Namespace = Namespace Text+  deriving newtype (Eq, Show)+ data ConnectionInfo = ConnectionInfo   { connectionInfoTls :: Bool   , connectionInfoPassword :: Maybe String   , connectionInfoHostName :: HostName   , connectionInfoPort :: PortNumber+  , connectionInfoNamespace :: Namespace   }   deriving stock (Eq, Show) @@ -39,6 +44,7 @@   , connectionInfoPassword = Nothing   , connectionInfoHostName = "localhost"   , connectionInfoPort = 7419+  , connectionInfoNamespace = Namespace ""   }  -- | Parse a @'Connection'@ from environment variables@@ -46,7 +52,7 @@ -- > FAKTORY_PROVIDER=FAKTORY_URL -- > FAKTORY_URL=tcp://:my-password@localhost:7419 ----- Supported format is @tcp(+tls):\/\/(:password@)host:port@.+-- Supported format is @tcp(+tls):\/\/(:password@)host:port(/namespace)@. -- -- See <https://github.com/contribsys/faktory/wiki/Worker-Lifecycle#url-configuration>. --@@ -92,7 +98,7 @@   some (upperChar <|> char '_') <?> "an environment variable name"  parseConnection :: Parser ConnectionInfo-parseConnection = go <?> "tcp(+tls)://(:<password>@)<host>:<port>"+parseConnection = go <?> "tcp(+tls)://(:<password>@)<host>:<port>(/namespace)"  where   go =     ConnectionInfo@@ -100,3 +106,5 @@       <*> optional (char ':' *> manyTill anySingle (char '@'))       <*> manyTill anySingle (char ':')       <*> (read <$> some digitChar)+      <*> (toNamespace <$> optional (char '/' *> some anySingle))+  toNamespace = Namespace . maybe "" pack

@@ -21,8 +21,9 @@ 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 Faktory.Client (Client(..))+import Faktory.Producer (Producer(..), pushJob)+import Faktory.Settings import GHC.Generics import GHC.Stack import System.Random@@ -65,18 +66,26 @@ -- '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+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+  jobJid job <$ pushJob producer job -applyOptions :: JobOptions -> Job arg -> IO (Job arg)-applyOptions (JobOptions patches) = go patches+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 q }+    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

@@ -0,0 +1,40 @@+module Faktory.Producer+  ( Producer(..)+  , newProducer+  , newProducerEnv+  , closeProducer+  , pushJob+  , flush+  )+where++import Faktory.Prelude++import Data.Aeson+import Faktory.Client+import Faktory.Settings+import GHC.Stack++newtype Producer = Producer+  { producerClient :: Client+  }++newProducer :: Settings -> IO Producer+newProducer settings = Producer <$> newClient settings Nothing++newProducerEnv :: IO Producer+newProducerEnv = newProducer =<< envSettings++closeProducer :: Producer -> IO ()+closeProducer = closeClient . producerClient++-- | Push a Job to the Server+pushJob :: (HasCallStack, ToJSON a) => Producer -> a -> IO ()+pushJob producer job = commandOK (producerClient producer) "PUSH" [encode job]++-- | Clear all job data in the Faktory server+--+-- Use with caution!+--+flush :: HasCallStack => Producer -> IO ()+flush producer = commandOK (producerClient producer) "FLUSH" []

@@ -1,13 +1,20 @@ module Faktory.Settings   ( Settings(..)-  , ConnectionInfo(..)   , defaultSettings   , envSettings+  , WorkerSettings(..)+  , defaultWorkerSettings+  , envWorkerSettings   , Queue(..)+  , namespaceQueue   , queueArg   , defaultQueue   , WorkerId   , randomWorkerId++  -- * Re-exports+  , ConnectionInfo(..)+  , Namespace(..)   ) where  import Faktory.Prelude@@ -22,20 +29,16 @@ import System.Random  data Settings = Settings-  { settingsQueue :: Queue-  , settingsConnection :: ConnectionInfo+  { settingsConnection :: ConnectionInfo   , settingsLogDebug :: String -> IO ()   , settingsLogError :: String -> IO ()-  , settingsWorkerIdleDelay :: Int   }  defaultSettings :: Settings defaultSettings = Settings-  { settingsQueue = defaultQueue-  , settingsConnection = defaultConnectionInfo+  { settingsConnection = defaultConnectionInfo   , settingsLogDebug = \_msg -> pure ()   , settingsLogError = hPutStrLn stderr . ("[ERROR]: " <>)-  , settingsWorkerIdleDelay = 1   }  -- | Defaults, but read @'Connection'@ from the environment@@ -44,15 +47,36 @@ -- envSettings :: IO Settings envSettings = do-  mQueue <- lookupEnv "FAKTORY_QUEUE"   connection <- envConnectionInfo-  pure defaultSettings+  pure defaultSettings { settingsConnection = connection }++data WorkerSettings = WorkerSettings+  { settingsQueue :: Queue+  , settingsId :: Maybe WorkerId+  , settingsIdleDelay :: Int+  }++defaultWorkerSettings :: WorkerSettings+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-    , settingsConnection = connection+    , settingsId = WorkerId <$> mWorkerId     }  newtype Queue = Queue Text   deriving newtype (IsString, FromJSON, ToJSON)++namespaceQueue :: Namespace -> Queue -> Queue+namespaceQueue (Namespace n) (Queue q) = Queue $ mappend n q  queueArg :: Queue -> ByteString queueArg (Queue q) = fromStrict $ encodeUtf8 q

@@ -6,6 +6,7 @@ module Faktory.Worker   ( WorkerHalt(..)   , runWorker+  , runWorkerEnv   ) where @@ -56,28 +57,43 @@   toJSON = genericToJSON $ aesonPrefix snakeCase   toEncoding = genericToEncoding $ aesonPrefix snakeCase -runWorker :: FromJSON args => Settings -> (args -> IO ()) -> IO ()-runWorker settings f = do-  workerId <- randomWorkerId+runWorker+  :: FromJSON args => Settings -> WorkerSettings -> (args -> IO ()) -> IO ()+runWorker settings workerSettings f = do+  workerId <- maybe randomWorkerId pure $ settingsId workerSettings   client <- newClient settings $ Just workerId   beatThreadId <- forkIOWithThrowToParent $ forever $ heartBeat client workerId -  forever (processorLoop client settings f)+  forever (processorLoop client settings workerSettings f)     `catch` (\(_ex :: WorkerHalt) -> pure ())     `finally` (killThread beatThreadId >> closeClient client) -processorLoop :: FromJSON arg => Client -> Settings -> (arg -> IO ()) -> IO ()-processorLoop client settings f = do+runWorkerEnv :: FromJSON args => (args -> IO ()) -> IO ()+runWorkerEnv f = do+  settings <- envSettings+  workerSettings <- envWorkerSettings+  runWorker settings workerSettings f++processorLoop+  :: FromJSON arg+  => Client+  -> Settings+  -> WorkerSettings+  -> (arg -> IO ())+  -> IO ()+processorLoop client settings workerSettings f = do   let+    namespace = connectionInfoNamespace $ settingsConnection settings     processAndAck job = do       f $ jobArg job       ackJob client job -  emJob <- fetchJob client $ settingsQueue settings+  emJob <- fetchJob client $ namespaceQueue namespace $ settingsQueue+    workerSettings    case emJob of     Left err -> settingsLogError settings $ "Invalid Job: " <> err-    Right Nothing -> threadDelaySeconds $ settingsWorkerIdleDelay settings+    Right Nothing -> threadDelaySeconds $ settingsIdleDelay workerSettings     Right (Just job) ->       processAndAck job         `catches` [ Handler $ \(ex :: WorkerHalt) -> throw ex

@@ -30,6 +30,7 @@         connectionInfoPassword `shouldBe` Nothing         connectionInfoHostName `shouldBe` "foo"         connectionInfoPort `shouldBe` 123+        connectionInfoNamespace `shouldBe` Namespace ""      it "parses tls and password" $ do       let@@ -44,6 +45,19 @@         connectionInfoPassword `shouldBe` Just "foo"         connectionInfoHostName `shouldBe` "bar"         connectionInfoPort `shouldBe` 123++    it "parses namespace" $ do+      let+        env =+          [ ("FAKTORY_PROVIDER", Nothing)+          , ("FAKTORY_URL", Just "tcp://localhost:7419/prefix")+          ]++      withEnvironment env $ do+        ConnectionInfo {..} <- envConnectionInfo+        connectionInfoHostName `shouldBe` "localhost"+        connectionInfoPort `shouldBe` 7419+        connectionInfoNamespace `shouldBe` Namespace "prefix"      it "follows _PROVIDER to find _URL" $ do       let

@@ -6,8 +6,8 @@  import Control.Concurrent (forkIO, threadDelay) import Control.Concurrent.MVar-import Faktory.Client import Faktory.Job+import Faktory.Producer import Faktory.Settings import Faktory.Worker import Test.Hspec@@ -15,15 +15,14 @@ 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"+    bracket newProducerEnv closeProducer $ \producer -> do+      void $ flush producer+      void $ perform @Text mempty producer "a"+      void $ perform @Text mempty producer "b"+      void $ perform @Text mempty producer "HALT"      processedJobs <- newMVar ([] :: [Text])-    runWorker settings $ \job -> do+    runWorkerEnv $ \job -> do       modifyMVar_ processedJobs $ pure . (job :)       when (job == "HALT") $ throw WorkerHalt @@ -31,15 +30,14 @@     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"+    bracket newProducerEnv closeProducer $ \producer -> do+      void $ flush producer+      void $ perform @Text once producer "a"+      void $ perform @Text (retry 0) producer "b"+      void $ perform @Text mempty producer "HALT"      processedJobs <- newMVar ([] :: [Text])-    runWorker settings $ \job -> do+    runWorkerEnv $ \job -> do       modifyMVar_ processedJobs $ pure . (job :)       when (job == "HALT") $ throw WorkerHalt @@ -47,8 +45,9 @@     jobs `shouldMatchList` ["a", "b", "HALT"]    it "correctly handles fetch timeouts" $ do-    settings' <- envSettings-    let settings = settings' { settingsWorkerIdleDelay = 0 }+    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.@@ -59,16 +58,14 @@     -- 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 (newClient settings Nothing) closeClient-      $ \client -> do-          void $ flush client-          threadDelay $ 2 * 1000000 + 250000-          void $ perform @Text mempty client "HALT"+    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 $ \job -> do+    runWorker settings workerSettings $ \job -> do       modifyMVar_ processedJobs $ pure . (job :)       when (job == "HALT") $ throw WorkerHalt