faktory 1.1.2.7 → 1.1.3.0

raw patch · 5 files changed

+218/−2 lines, 5 filesdep +microlensdep +resource-pooldep +unliftioPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: microlens, resource-pool, unliftio

API changes (from Hackage documentation)

Files

• CHANGELOG.md +8/−1
• faktory.cabal +6/−1
• library/Data/Pool/Compat.hs +31/−0
• library/Faktory/Pool.hs +126/−0
• library/Faktory/Settings.hs +47/−0

@@ -1,4 +1,11 @@-## [_Unreleased_](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.2.7...main)+## [_Unreleased_](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.3.0...main)++## [v1.1.3.0](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.2.7...v1.1.3.0)++- Add `Faktory.Pool`++  This incurs the new dependencies, `unliftio`, `resource-pool`, and+  `microlens`.  ## [v1.1.2.7](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.2.6...v1.1.2.7) 

@@ -1,6 +1,6 @@ cabal-version:   1.18 name:            faktory-version:         1.1.2.7+version:         1.1.3.0 license:         MIT license-file:    LICENSE copyright:       2018 Freckle Education@@ -51,6 +51,7 @@  library     exposed-modules:+        Data.Pool.Compat         Faktory.Client         Faktory.Connection         Faktory.Ent.Batch@@ -61,6 +62,7 @@         Faktory.JobFailure         Faktory.JobOptions         Faktory.JobState+        Faktory.Pool         Faktory.Prelude         Faktory.Producer         Faktory.Protocol@@ -97,15 +99,18 @@         errors >=2.3.0,         megaparsec >=8.0.0,         memory >=0.15.0,+        microlens >=0.4.11.2,         mtl >=2.2.2,         network >=3.1.1.1,         random >=1.1,+        resource-pool >=0.2.3.2,         safe-exceptions >=0.1.7.1,         scanner >=0.3.1,         semigroups >=0.19.1,         text >=1.2.4.0,         time >=1.9.3,         unix >=2.7.2.2,+        unliftio >=0.2.13.1,         unordered-containers >=0.2.10.0      if impl(ghc >=8.10)

@@ -0,0 +1,31 @@+{-# LANGUAGE CPP #-}++module Data.Pool.Compat+  ( module Data.Pool+  , createPool+  ) where++import Prelude++import Data.Pool hiding (createPool)+#if MIN_VERSION_resource_pool(0,3,0)+#else+import Control.Concurrent (getNumCapabilities)+import qualified Data.Pool as Pool+#endif++createPool+  :: IO a+  -> (a -> IO ())+  -> Double+  -> Int+  -> IO (Pool a)+createPool create destroy timeout size = do+#if MIN_VERSION_resource_pool(0,3,0)+  newPool $ defaultPoolConfig create destroy timeout size+#else+  -- Re-implement instead of using the deprecated compatibility function, so+  -- that we can get a consistent numStripes and size behavior.+  numStripes <- getNumCapabilities+  Pool.createPool create destroy numStripes (realToFrac timeout) size+#endif

@@ -0,0 +1,126 @@+module Faktory.Pool+  ( FaktoryPool+  , HasFaktoryPool (..)++    -- * Pool Construction+  , Settings+  , PoolSettings+  , newFaktoryPool++    -- * Pool use+  , perform+  , buildJob++    -- * Direct access+  , withProducer+  , takeProducer++    -- * Re-exports+  , module Faktory.Job+  ) where++import Faktory.Prelude++import Control.Monad.IO.Class (MonadIO (..))+import Control.Monad.Reader (MonadReader, asks)+import Data.Aeson (ToJSON)+import Data.Pool.Compat (Pool)+import qualified Data.Pool.Compat as Pool+import Faktory.Job hiding (buildJob, perform)+import qualified Faktory.Job as Job+import Faktory.Producer+import Faktory.Settings (PoolSettings (..), Settings)+import GHC.Stack (HasCallStack)+import Lens.Micro (Lens', (^.))+import UnliftIO (MonadUnliftIO, withRunInIO)++-- |+--+-- @since 1.1.3.0+newtype FaktoryPool = FaktoryPool (Pool Producer)++-- |+--+-- @since 1.1.3.0+class HasFaktoryPool env where+  faktoryPoolL :: Lens' env FaktoryPool++instance HasFaktoryPool FaktoryPool where+  faktoryPoolL = id++-- | Build a 'FaktoryPool' with the given settings+--+-- See 'Settings', 'envSettings', 'PoolSettings', and 'envPoolSettings'.+--+-- @since 1.1.3.0+newFaktoryPool+  :: MonadIO m+  => Settings+  -> PoolSettings+  -> m FaktoryPool+newFaktoryPool settings PoolSettings {..} = do+  liftIO $+    FaktoryPool+      <$> Pool.createPool+        (newProducer settings)+        closeProducer+        (fromIntegral settingsTimeout)+        (fromIntegral settingsSize)++-- | 'Faktory.Job.perform' but using a 'Producer' from the pool+--+-- @since 1.1.3.0+perform+  :: ( MonadUnliftIO m+     , MonadReader env m+     , HasFaktoryPool env+     , ToJSON arg+     , HasCallStack+     )+  => JobOptions+  -> arg+  -> m JobId+perform options arg = do+  withProducer $ \producer -> do+    liftIO $ Job.perform options producer arg++-- | 'Faktory.Job.buildJob' but using a 'Producer' from the pool+--+-- @since 1.1.3.0+buildJob+  :: (MonadUnliftIO m, MonadReader env m, HasFaktoryPool env)+  => JobOptions+  -> arg+  -> m (Job arg)+buildJob options arg = do+  withProducer $ \producer -> do+    liftIO $ Job.buildJob options producer arg++-- | Acquire a 'Producer', use it, and return it to the pool+--+-- @since 1.1.3.0+withProducer+  :: (MonadUnliftIO m, MonadReader env m, HasFaktoryPool env)+  => (Producer -> m a)+  -> m a+withProducer f = do+  FaktoryPool p <- asks (^. faktoryPoolL)+  withRunInIO $ \runInIO -> do+    Pool.withResource p $ runInIO . f++-- | Get a 'Producer' from the pool along with an action to return it+--+-- You should prefer 'withProducer' if at all possible. With this function you+-- are responsible to ensure the return action is called (e.g. with 'finally').+--+-- This is only necessary if you are operating in a monad that doesn't have+-- 'MonadUnliftIO' (like 'ConduitT'), so you need to take and return a+-- 'Producer' separately (e.g. with 'bracketP').+--+-- @since 1.1.3.0+takeProducer+  :: (MonadIO m, MonadReader env m, HasFaktoryPool env) => m (Producer, m ())+takeProducer = do+  FaktoryPool p <- asks (^. faktoryPoolL)+  (producer, lp) <- liftIO $ Pool.takeResource p+  pure (producer, liftIO $ Pool.putResource lp producer)

@@ -2,6 +2,8 @@   ( Settings (..)   , defaultSettings   , envSettings++    -- * Worker   , WorkerSettings (..)   , defaultWorkerSettings   , envWorkerSettings@@ -12,6 +14,10 @@   , WorkerId   , randomWorkerId +    -- * Pool+  , PoolSettings (..)+  , envPoolSettings+     -- * Re-exports   , ConnectionInfo (..)   , Namespace (..)@@ -23,6 +29,7 @@ import Faktory.Connection import Faktory.JobOptions (JobOptions) import Faktory.Settings.Queue+import Numeric.Natural import System.Environment (lookupEnv) import System.IO (hPutStrLn, stderr) import System.Random@@ -82,3 +89,43 @@  randomWorkerId :: IO WorkerId randomWorkerId = WorkerId . take 8 . randomRs ('a', 'z') <$> newStdGen++-- |+--+-- @since 1.1.3.0+data PoolSettings = PoolSettings+  { settingsSize :: Natural+  -- ^ Maximum pool size+  --+  -- Default is @10@. Smallest acceptable value is @1@. Note that, due to the+  -- striping behavior of @resource-pool@, a configured size @N@ may result in+  -- @N - 1@ resources.+  , settingsTimeout :: Natural+  -- ^ How long before destroying a resource, in seconds+  --+  -- Default is @600@.+  }++-- |+--+-- @since 1.1.3.0+defaultPoolSettings :: PoolSettings+defaultPoolSettings =+  PoolSettings+    { settingsSize = 10+    , settingsTimeout = 600+    }++-- | Read 'PoolSettings' from the environment+--+-- - @FAKTORY_POOL_SIZE@+-- - @FAKTORY_POOL_TIMEOUT@+--+-- @since 1.1.3.0+envPoolSettings :: IO PoolSettings+envPoolSettings =+  PoolSettings+    <$> (maybe settingsSize read <$> lookupEnv "FAKTORY_POOL_SIZE")+    <*> (maybe settingsTimeout read <$> lookupEnv "FAKTORY_POOL_TIMEOUT")+ where+  PoolSettings {..} = defaultPoolSettings