Add 'delay' combinator

conduit/ChangeLog.md

@@ -1,5 +1,9 @@
 # ChangeLog for conduit
 
+## 1.3.5
+
+* Add `delay` [#481](https://github.com/snoyberg/conduit/issues/481)
+
 ## 1.3.4.2
 
 * Fix GHC 9.2 build [#473](https://github.com/snoyberg/conduit/pull/473)

conduit/src/Data/Conduit/Combinators.hs

@@ -198,6 +198,7 @@ module Data.Conduit.Combinators
     , mapAccumS
     , peekForever
     , peekForeverE
+    , delay
     ) where
 
 -- BEGIN IMPORTS
@@ -213,6 +214,7 @@ import           Data.ByteString.Lazy.Internal (defaultChunkSize)
 import           Control.Applicative         (Alternative(..), (<$>))
 import           Control.Exception           (catch, throwIO, finally, bracket, try, evaluate)
 import           Control.Category            (Category (..))
+import           Control.Concurrent          (threadDelay)
 import           Control.Monad               (unless, when, (>=>), liftM, forever)
 import           Control.Monad.IO.Unlift     (MonadIO (..), MonadUnliftIO, withRunInIO)
 import           Control.Monad.Primitive     (PrimMonad, PrimState, unsafePrimToPrim)
@@ -2554,3 +2556,11 @@ peekForeverE inner =
         case mx of
             Nothing -> return ()
             Just _ -> inner >> loop
+
+-- | Delays the emission of values for a given number of microseconds.
+--
+-- @since 1.3.5
+delay :: MonadIO m => Int -> ConduitT a a m ()
+delay n = awaitForever $ \v -> do
+  liftIO $ threadDelay n
+  yield v

conduit/src/Data/Conduit/Combinators/Unqualified.hs

@@ -180,6 +180,7 @@ module Data.Conduit.Combinators.Unqualified
     , CC.mapAccumS
     , CC.peekForever
     , CC.peekForeverE
+    , CC.delay
     ) where
 
 -- BEGIN IMPORTS

conduit/test/Spec.hs

@@ -7,10 +7,11 @@ module Spec (spec) where
 
 import Conduit
 import Prelude hiding (FilePath)
-import Data.Maybe (listToMaybe)
+import Data.Maybe (listToMaybe, isNothing)
 import Data.Conduit.Combinators (slidingWindow, chunksOfE, chunksOfExactlyE)
 import Data.List (intersperse, sort, find, mapAccumL)
 import Safe (tailSafe)
+import System.CPUTime (getCPUTime)
 import System.FilePath (takeExtension, (</>))
 import Test.Hspec
 import Test.Hspec.QuickCheck
@@ -21,7 +22,7 @@ import Data.IORef
 import qualified Data.Vector as V
 import qualified Data.Vector.Unboxed as VU
 import qualified Data.Vector.Storable as VS
-import Control.Monad (liftM)
+import Control.Monad (liftM, when, void)
 import Control.Monad.ST (runST)
 import Control.Monad.Trans.Writer
 import qualified System.IO as IO
@@ -637,6 +638,16 @@ spec = do
         res1 <- runConduit $ yieldMany strs .| linesUnboundedC .| sinkList
         res2 <- runConduit $ yieldMany strs .| peekForeverE (lineC $ foldC >>= yield) .| sinkList
         res2 `shouldBe` res1
+    it "delay" $ do
+        let duration = 100000 :: Int
+        elapsed <- runConduit $ yield () .| delay duration .| do
+            start <- liftIO $ getCPUTime
+            mValue <- await
+            void $ when (isNothing mValue) $
+                error "expected 'Just ()'"
+            end <- liftIO $ getCPUTime
+            return (end - start)
+        elapsed `shouldSatisfy` (>= (toInteger duration))
     StreamSpec.spec
 
 evenInt :: Int -> Bool