conduit-throttle 0.2.0.1 → 0.3.0.0

raw patch · 7 files changed

+222/−82 lines, 7 filesdep +monad-controlPVP ok

version bump matches the API change (PVP)

Dependencies added: monad-control

API changes (from Hackage documentation)

Files

• conduit-throttle.cabal +10/−3
• src/Data/Conduit/Throttle.hs +4/−71
• src/Data/Conduit/Throttle/Internal.hs +85/−0
• src/Data/Conduit/Throttle/MBC.hs +65/−0
• tests/Data/Conduit/Throttle/MBC/Test.hs +37/−0
• tests/Data/Conduit/Throttle/Test.hs +3/−8
• tests/Tests.hs +18/−0

@@ -3,7 +3,7 @@ -- see: https://github.com/sol/hpack  name:                conduit-throttle-version:             0.2.0.1+version:             0.3.0.0 synopsis:            Throttle Conduit Producers description:         This packages is based on the throttle-io-stream package and provides functionality for throttling Conduit producers according to a provided configuration. homepage:            https://github.com/mtesseract/conduit-throttle#readme@@ -30,7 +30,9 @@   ghc-options: -Wall   exposed-modules:       Data.Conduit.Throttle+      Data.Conduit.Throttle.MBC   other-modules:+      Data.Conduit.Throttle.Internal       Paths_conduit_throttle   build-depends:       base >=4.7 && <5@@ -45,13 +47,14 @@     , unliftio-core     , throttle-io-stream     , resourcet+    , monad-control   default-language: Haskell2010  test-suite conduit-test   type: exitcode-stdio-1.0-  main-is: Test.hs+  main-is: Tests.hs   hs-source-dirs:-      tests/Data/Conduit/Throttle+      tests   default-extensions: OverloadedStrings   ghc-options: -Wall -Wall -fno-warn-type-defaults   build-depends:@@ -67,6 +70,7 @@     , unliftio-core     , throttle-io-stream     , resourcet+    , monad-control     , base >=4.7 && <5     , HUnit     , test-framework@@ -74,4 +78,7 @@     , conduit-throttle     , throttle-io-stream     , stm-conduit+  other-modules:+      Data.Conduit.Throttle.MBC.Test+      Data.Conduit.Throttle.Test   default-language: Haskell2010

@@ -1,6 +1,5 @@ {-# LANGUAGE LambdaCase        #-} {-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RankNTypes        #-} {-# LANGUAGE RecordWildCards   #-}  module Data.Conduit.Throttle@@ -19,83 +18,17 @@ import           Control.Concurrent.STM.TBMQueue import qualified Control.Concurrent.Throttle     as Throttle import           Control.Monad.Trans.Resource-import           Data.Function+import           Data.Conduit.Throttle.Internal import           UnliftIO -data Conf a = Conf-  { _measure       :: Throttle.Measure a-  , _interval      :: Double-  , _maxThroughput :: Double-  , _bufferSize    :: Int-  , _emaAlpha      :: Double-  }--newConf :: Conf a-newConf = defaultConf---- | Default 'ThrottleConf'.-defaultConf :: Conf a-defaultConf = Conf-  { _measure       = const 1-  , _interval      = 1000-  , _maxThroughput = 100-  , _bufferSize    = 1024-  , _emaAlpha      = defaultEmaAlpha }---- | Set measure function in configuration.-setMeasure :: Throttle.Measure a-           -> Conf a-           -> Conf a-setMeasure measure conf = conf { _measure = measure }---- | Set interval in configuration.-setInterval :: Double-            -> Conf a-            -> Conf a-setInterval interval conf = conf { _interval = interval }---- | Set max throughput in configuration.-setMaxThroughput :: Double-                 -> Conf a-                 -> Conf a-setMaxThroughput throughput conf =-  conf { _maxThroughput = throughput }---- | Set buffer size in configuration.-setBufferSize :: Int-              -> Conf a-              -> Conf a-setBufferSize n conf = conf { _bufferSize = n }---- | Set exponential weight factor used for computing current item--- size.-setEmaAlpha :: Double-            -> Conf a-            -> Conf a-setEmaAlpha alpha conf = conf { _emaAlpha = alpha }---- | Default exponential weight factor for computing current item--- size.-defaultEmaAlpha :: Double-defaultEmaAlpha = 0.5--throttleConfPrepare :: Conf a -> Throttle.ThrottleConf a-throttleConfPrepare Conf { .. } = Throttle.newThrottleConf-  & Throttle.throttleConfThrottleProducer-  & Throttle.throttleConfSetMeasure _measure-  & Throttle.throttleConfSetInterval _interval-  & Throttle.throttleConfSetMaxThroughput _maxThroughput-  & Throttle.throttleConfSetBufferSize _bufferSize-  & Throttle.throttleConfSetEmaAlpha _emaAlpha- -- | Given a 'ThrottleConf' and a 'Producer', create and return a new -- 'Producer', which yields the same stream of values like the -- provided producer but throttled according to the provided -- throttling configuration. throttleProducer :: (MonadUnliftIO m, MonadResource m)-                 => Conf a-                 -> Producer m a-                 -> Producer m a+                 => Conf o+                 -> ConduitM () o m ()+                 -> ConduitM () o m () throttleProducer conf producer = do   (UnliftIO unlifter) <- lift askUnliftIO   queueIn  <- liftIO $ newTBMQueueIO 1024

@@ -0,0 +1,85 @@+{-# LANGUAGE LambdaCase        #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes        #-}+{-# LANGUAGE RecordWildCards   #-}++module Data.Conduit.Throttle.Internal+  ( Conf+  , newConf+  , setMeasure+  , setInterval+  , setMaxThroughput+  , setBufferSize+  , setEmaAlpha+  , throttleConfPrepare+  ) where++import qualified Control.Concurrent.Throttle as Throttle+import           Data.Function++data Conf a = Conf+  { _measure       :: Throttle.Measure a+  , _interval      :: Double+  , _maxThroughput :: Double+  , _bufferSize    :: Int+  , _emaAlpha      :: Double+  }++newConf :: Conf a+newConf = defaultConf++-- | Default 'ThrottleConf'.+defaultConf :: Conf a+defaultConf = Conf+  { _measure       = const 1+  , _interval      = 1000+  , _maxThroughput = 100+  , _bufferSize    = 1024+  , _emaAlpha      = defaultEmaAlpha }++-- | Set measure function in configuration.+setMeasure :: Throttle.Measure a+           -> Conf a+           -> Conf a+setMeasure measure conf = conf { _measure = measure }++-- | Set interval in configuration.+setInterval :: Double+            -> Conf a+            -> Conf a+setInterval interval conf = conf { _interval = interval }++-- | Set max throughput in configuration.+setMaxThroughput :: Double+                 -> Conf a+                 -> Conf a+setMaxThroughput throughput conf =+  conf { _maxThroughput = throughput }++-- | Set buffer size in configuration.+setBufferSize :: Int+              -> Conf a+              -> Conf a+setBufferSize n conf = conf { _bufferSize = n }++-- | Set exponential weight factor used for computing current item+-- size.+setEmaAlpha :: Double+            -> Conf a+            -> Conf a+setEmaAlpha alpha conf = conf { _emaAlpha = alpha }++-- | Default exponential weight factor for computing current item+-- size.+defaultEmaAlpha :: Double+defaultEmaAlpha = 0.5++throttleConfPrepare :: Conf a -> Throttle.ThrottleConf a+throttleConfPrepare Conf { .. } = Throttle.newThrottleConf+  & Throttle.throttleConfThrottleProducer+  & Throttle.throttleConfSetMeasure _measure+  & Throttle.throttleConfSetInterval _interval+  & Throttle.throttleConfSetMaxThroughput _maxThroughput+  & Throttle.throttleConfSetBufferSize _bufferSize+  & Throttle.throttleConfSetEmaAlpha _emaAlpha+

@@ -0,0 +1,65 @@+{-# LANGUAGE FlexibleContexts      #-}+{-# LANGUAGE LambdaCase            #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings     #-}+{-# LANGUAGE RecordWildCards       #-}++module Data.Conduit.Throttle.MBC+  ( Conf+  , newConf+  , setMeasure+  , setInterval+  , setMaxThroughput+  , setBufferSize+  , setEmaAlpha+  , throttleProducer+  ) where++import           Conduit+import           Control.Concurrent.Async+import           Control.Concurrent.STM+import           Control.Concurrent.STM.TBMQueue+import qualified Control.Concurrent.Throttle     as Throttle+import           Control.Monad+import           Control.Monad.Trans.Control+import           Control.Monad.Trans.Resource+import           Data.Conduit.Throttle.Internal++-- | Given a 'ThrottleConf' and a 'Producer', create and return a new+-- 'Producer', which yields the same stream of values like the+-- provided producer but throttled according to the provided+-- throttling configuration.+throttleProducer :: (MonadIO m, MonadBaseControl IO m, MonadResource m)+                 => Conf o+                 -> ConduitM () o m ()+                 -> ConduitM () o m ()+throttleProducer conf producer = do+  queueIn  <- liftIO $ newTBMQueueIO 1024+  queueOut <- liftIO $ newTBMQueueIO 1+  conduitProcess <- lift $ liftBaseWith $ \runInBase ->+    return $ runInBase $ runConduit (producer .| drainConduit queueIn)+  void $ allocate (async conduitProcess) cancel++  let throttleConf  = throttleConfPrepare conf+      readCallback  = atomically (readTBMQueue queueIn)+      writeCallback = \case+        Just a  -> atomically $ writeTBMQueue queueOut a+        Nothing -> atomically $ closeTBMQueue queueOut++  (_, asyncThrottler) <- allocate+    (Throttle.throttle throttleConf readCallback writeCallback)+    cancel++  liftIO $ link asyncThrottler+  go queueOut++  where go queue = do+          liftIO (atomically (readTBMQueue queue)) >>= \case+            Just a  -> yield a >> go queue+            Nothing -> return ()++        drainConduit queue = do+          await >>= \case+            Just a  -> do liftIO (atomically (writeTBMQueue queue a))+                          drainConduit queue+            Nothing -> liftIO (atomically (closeTBMQueue queue))

@@ -0,0 +1,37 @@+{-# LANGUAGE OverloadedStrings #-}++module Data.Conduit.Throttle.MBC.Test (tests) where++import           Conduit+import           Data.Conduit.Async+import           Data.Conduit.List              (sourceList)+import qualified Data.Conduit.Throttle.MBC      as ConduitThrottle+import           Data.Function                  ((&))+import           Test.Framework                 (Test, testGroup)+import           Test.Framework.Providers.HUnit (testCase)++tests :: [Test.Framework.Test]+tests =+  [ testGroup "Test Suite [MonadBaseControl version]"+    [ testCase "Simple producer throttling"+      simpleProducerThrottling+    , testCase "Simple producer throttling (using stm-conduit)"+      simpleProducerThrottlingStm+    ]+  ]++simpleProducerThrottling :: IO ()+simpleProducerThrottling = do+  let conf = ConduitThrottle.newConf+             & ConduitThrottle.setInterval 1000+             & ConduitThrottle.setMaxThroughput 1+  runResourceT . runConduit $+    ConduitThrottle.throttleProducer conf (sourceList [1..5]) .| printC++simpleProducerThrottlingStm :: IO ()+simpleProducerThrottlingStm = do+  let conf = ConduitThrottle.newConf+             & ConduitThrottle.setInterval 1000+             & ConduitThrottle.setMaxThroughput 1+  runResourceT . runCConduit $+    ConduitThrottle.throttleProducer conf (sourceList [1..5]) =$=& printC

@@ -1,23 +1,18 @@ {-# LANGUAGE OverloadedStrings #-} -module Main where+module Data.Conduit.Throttle.Test (tests) where  import           Conduit import           Data.Conduit.Async import           Data.Conduit.List              (sourceList) import qualified Data.Conduit.Throttle          as ConduitThrottle import           Data.Function                  ((&))-import           Test.Framework                 (Test, defaultMain, testGroup)+import           Test.Framework                 (Test, testGroup) import           Test.Framework.Providers.HUnit (testCase) -main :: IO ()-main = do-  putStrLn ""-  defaultMain tests- tests :: [Test.Framework.Test] tests =-  [ testGroup "Test Suite"+  [ testGroup "Test Suite [UnliftIO version]"     [ testCase "Simple producer throttling"       simpleProducerThrottling     , testCase "Simple producer throttling (using stm-conduit)"

@@ -0,0 +1,18 @@+{-# LANGUAGE OverloadedStrings #-}++module Main where++import           Test.Framework                 (Test, defaultMain)++import qualified Data.Conduit.Throttle.MBC.Test+import qualified Data.Conduit.Throttle.Test++main :: IO ()+main = do+  putStrLn ""+  defaultMain tests++tests :: [Test.Framework.Test]+tests =+  Data.Conduit.Throttle.Test.tests+  ++ Data.Conduit.Throttle.MBC.Test.tests