content: {-# LANGUAGE BangPatterns       #-}
  {-# LANGUAGE DeriveDataTypeable #-}
  {-# LANGUAGE OverloadedStrings  #-}
  
  -- | Stream operations on 'ByteString'.
  module System.IO.Streams.ByteString
   ( -- * Counting bytes
     countInput
   , countOutput
  
     -- * Treating strings as streams
   , fromByteString
   , fromLazyByteString
  
     -- * Input and output
   , readExactly
   , takeBytesWhile
   , writeLazyByteString
  
     -- * Stream transformers
     -- ** Splitting/Joining
   , splitOn
   , lines
   , unlines
   , words
   , unwords
  
     -- ** Other
   , giveBytes
   , giveExactly
   , takeBytes
   , takeExactly
   , throwIfConsumesMoreThan
   , throwIfProducesMoreThan
  
     -- ** Rate limiting
   , throwIfTooSlow
  
     -- * String search
   , MatchInfo(..)
   , search
  
     -- * Exception types
   , RateTooSlowException
   , ReadTooShortException
   , TooManyBytesReadException
   , TooManyBytesWrittenException
   , TooFewBytesWrittenException
  
   ) where
  
  ------------------------------------------------------------------------------
  import           Control.Exception                 (Exception, throwIO)
  import           Control.Monad                     (when, (>=>))
  import           Data.ByteString                   (ByteString)
  import qualified Data.ByteString.Char8             as S
  import qualified Data.ByteString.Lazy.Char8        as L
  import qualified Data.ByteString.Unsafe            as S
  import           Data.Char                         (isSpace)
  import           Data.Int                          (Int64)
  import           Data.IORef                        (IORef, newIORef, readIORef, writeIORef)
  import           Data.Time.Clock.POSIX             (getPOSIXTime)
  import           Data.Typeable                     (Typeable)
  import           Prelude                           hiding (lines, read, unlines, unwords, words)
  ------------------------------------------------------------------------------
  import           System.IO.Streams.Combinators     (filterM, intersperse, outputFoldM)
  import           System.IO.Streams.Internal        (InputStream (..), OutputStream, makeInputStream, makeOutputStream, read, unRead, write)
  import           System.IO.Streams.Internal.Search (MatchInfo (..), search)
  import           System.IO.Streams.List            (fromList, writeList)
  
  
  ------------------------------------------------------------------------------
  {-# INLINE modifyRef #-}
  modifyRef :: IORef a -> (a -> a) -> IO ()
  modifyRef ref f = do
      x <- readIORef ref
      writeIORef ref $! f x
  
  
  ------------------------------------------------------------------------------
  -- | Writes a lazy 'ByteString' to an 'OutputStream'.
  --
  -- Example:
  --
  -- @
  -- ghci> Streams.'writeLazyByteString' \"Test\\n\" Streams.'System.IO.Streams.stdout'
  -- Test
  -- @
  writeLazyByteString :: L.ByteString             -- ^ string to write to output
                      -> OutputStream ByteString  -- ^ output stream
                      -> IO ()
  writeLazyByteString = writeList . L.toChunks
  {-# INLINE writeLazyByteString #-}
  
  
  ------------------------------------------------------------------------------
  -- | Creates an 'InputStream' from a 'ByteString'.
  fromByteString :: ByteString -> IO (InputStream ByteString)
  fromByteString = fromList . (:[])
  
  
  ------------------------------------------------------------------------------
  -- | Creates an 'InputStream' from a lazy 'ByteString'.
  fromLazyByteString :: L.ByteString -> IO (InputStream ByteString)
  fromLazyByteString = fromList . L.toChunks
  
  
  ------------------------------------------------------------------------------
  -- | Wraps an 'InputStream', counting the number of bytes produced by the
  -- stream as a side effect. Produces a new 'InputStream' as well as an IO
  -- action to retrieve the count of bytes produced.
  --
  -- Strings pushed back to the returned 'InputStream' will be pushed back to the
  -- original stream, and the count of produced bytes will be subtracted
  -- accordingly.
  --
  -- Example:
  --
  -- @
  -- ghci> is <- Streams.'System.IO.Streams.fromList' [\"abc\", \"def\", \"ghi\"::ByteString]
  -- ghci> (is', getCount) <- Streams.'countInput' is
  -- ghci> Streams.'read' is'
  -- Just \"abc\"
  -- ghci> getCount
  -- 3
  -- ghci> Streams.'unRead' \"bc\" is'
  -- ghci> getCount
  -- 1
  -- ghci> Streams.'System.IO.Streams.peek' is
  -- Just \"bc\"
  -- ghci> Streams.'System.IO.Streams.toList' is'
  -- [\"bc\",\"def\",\"ghi\"]
  -- ghci> getCount
  -- 9
  -- @
  --
  countInput :: InputStream ByteString -> IO (InputStream ByteString, IO Int64)
  countInput src = do
      ref    <- newIORef (0 :: Int64)
      return $! (InputStream (prod ref) (pb ref), readIORef ref)
  
    where
      prod ref = read src >>= maybe (return Nothing) (\x -> do
          modifyRef ref (+ (fromIntegral $ S.length x))
          return $! Just x)
  
      pb ref s = do
          modifyRef ref (\x -> x - (fromIntegral $ S.length s))
          unRead s src
  
  
  ------------------------------------------------------------------------------
  -- | Wraps an 'OutputStream', counting the number of bytes consumed by the
  -- stream as a side effect. Produces a new 'OutputStream' as well as an IO
  -- action to retrieve the count of bytes consumed.
  --
  -- Example:
  --
  -- @
  -- ghci> (os :: OutputStream ByteString, getList) <- Streams.'System.IO.Streams.listOutputStream'
  -- ghci> (os', getCount) <- Streams.'countOutput' os
  -- ghci> Streams.'System.IO.Streams.fromList' [\"abc\", \"def\", \"ghi\"] >>= Streams.'System.IO.Streams.connectTo' os'
  -- ghci> getList
  -- [\"abc\",\"def\",\"ghi\"]
  -- ghci> getCount
  -- 9
  -- @
  countOutput :: OutputStream ByteString
              -> IO (OutputStream ByteString, IO Int64)
  countOutput = outputFoldM f 0
    where
      f !count s = return z
        where
          !c = S.length s
          !z = toEnum c + count
  
  
  ------------------------------------------------------------------------------
  -- | Wraps an 'InputStream', producing a new 'InputStream' that will produce at
  -- most @n@ bytes, subsequently yielding end-of-stream forever.
  --
  -- Strings pushed back to the returned 'InputStream' will be propagated
  -- upstream, modifying the count of taken bytes accordingly.
  --
  -- Example:
  --
  -- @
  -- ghci> is <- Streams.'System.IO.Streams.fromList' [\"truncated\", \" string\"::ByteString]
  -- ghci> is' <- Streams.'takeBytes' 9 is
  -- ghci> Streams.'read' is'
  -- Just \"truncated\"
  -- ghci> Streams.'read' is'
  -- Nothing
  -- ghci> Streams.'System.IO.Streams.peek' is
  -- Just \" string\"
  -- ghci> Streams.'unRead' \"cated\" is'
  -- ghci> Streams.'System.IO.Streams.peek' is
  -- Just \"cated\"
  -- ghci> Streams.'System.IO.Streams.peek' is'
  -- Just \"cated\"
  -- ghci> Streams.'read' is'
  -- Just \"cated\"
  -- ghci> Streams.'read' is'
  -- Nothing
  -- ghci> Streams.'read' is
  -- Just \" string\"
  -- @
  takeBytes :: Int64                        -- ^ maximum number of bytes to read
            -> InputStream ByteString       -- ^ input stream to wrap
            -> IO (InputStream ByteString)
  takeBytes k0 = takeBytes' k0 (return Nothing)
  {-# INLINE takeBytes #-}
  
  
  ------------------------------------------------------------------------------
  -- | Like @Streams.'takeBytes'@, but throws 'ReadTooShortException' when
  -- there aren't enough bytes present on the source.
  takeExactly :: Int64                        -- ^ number of bytes to read
              -> InputStream ByteString       -- ^ input stream to wrap
              -> IO (InputStream ByteString)
  takeExactly k0 = takeBytes' k0 (throwIO $ ReadTooShortException k0)
  {-# INLINE takeExactly #-}
  
  
  ------------------------------------------------------------------------------
  -- Helper for the two above.
  takeBytes' :: Int64
             -> IO (Maybe ByteString)
             -- ^ What to do if the input ends before having consumed the
             -- right amount of bytes.
             -> InputStream ByteString
             -> IO (InputStream ByteString)
  takeBytes' k0 h src = do
      kref <- newIORef k0
      return $! InputStream (prod kref) (pb kref)
    where
      prod kref = do
          k <- readIORef kref
          if k <= 0
             then return Nothing
             else read src >>= maybe h (chunk k)
        where
          chunk k s = do
              let l  = fromIntegral $ S.length s
              let k' = k - l
              if k' <= 0
                then let (a,b) = S.splitAt (fromIntegral k) s
                     in do
                         when (not $ S.null b) $ unRead b src
                         writeIORef kref 0
                         return $! Just a
                else writeIORef kref k' >> return (Just s)
  
      pb kref s = do
          modifyRef kref (+ (fromIntegral $ S.length s))
          unRead s src
  {-# INLINE takeBytes' #-}
  
  
  ------------------------------------------------------------------------------
  -- | Splits an 'InputStream' over 'ByteString's using a delimiter predicate.
  --
  -- Note that:
  --
  --   * data pushed back with 'unRead' is *not* propagated upstream here.
  --
  --   * the resulting 'InputStream' may hold an unbounded amount of the
  --     bytestring in memory waiting for the function to return true, so this
  --     function should not be used in unsafe contexts.
  --
  --   * the delimiter is NOT included in the output.
  --
  --   * consecutive delimiters are not merged.
  --
  --   * if the input ends in the delimiter, a final empty string is /not/
  --     emitted. (/Since: 1.5.0.0. Previous versions had the opposite behaviour,
  --     which was changed to match 'Prelude.lines'./)
  --
  -- Example:
  --
  -- @
  -- ghci> Streams.'System.IO.Streams.fromList' [\"the quick br\", \"own  fox\"::'ByteString'] >>=
  --       Streams.'splitOn' (== \' \') >>= Streams.'System.IO.Streams.toList'
  -- [\"the\",\"quick\",\"brown\",\"\",\"fox\"]
  -- @
  --
  splitOn :: (Char -> Bool)               -- ^ predicate used to break the input
                                          -- stream into chunks
          -> InputStream ByteString       -- ^ input stream
          -> IO (InputStream ByteString)
  splitOn p is = do
      ref <- newIORef id
      makeInputStream $ start ref
    where
      start ref = go
        where
          go  = read is >>= maybe end chunk
  
          end = do
              dl <- readIORef ref
              case dl [] of
                [] -> return Nothing
                xs -> writeIORef ref id >>
                      (return $! Just $! S.concat xs)
  
          chunk s = let (a, b) = S.break p s
                    in if S.null b
                         then modifyRef ref (\f -> f . (a:)) >> go
                         else do
                           let !b' = S.unsafeDrop 1 b
                           dl <- readIORef ref
                           when (not $ S.null b') $ unRead b' is
                           writeIORef ref id
                           return $ Just $! S.concat $ dl [a]
  
  
  ------------------------------------------------------------------------------
  -- | Splits a bytestring 'InputStream' into lines. See 'splitOn' and
  -- 'Prelude.lines'.
  --
  -- Example:
  --
  -- @
  -- ghci> is \<- Streams.'System.IO.Streams.fromList' [\"Hello,\\n world!\"] >>= Streams.'lines'
  -- ghci> replicateM 3 (Streams.'read' is)
  -- [Just \"Hello\", Just \", world!\", Nothing]
  -- @
  --
  -- Note that this may increase the chunk size if the input contains extremely
  -- long lines.
  lines :: InputStream ByteString -> IO (InputStream ByteString)
  lines = splitOn (== '\n')
  
  
  ------------------------------------------------------------------------------
  -- | Splits a bytestring 'InputStream' into words. See 'splitOn' and
  -- 'Prelude.words'.
  --
  -- Example:
  --
  -- @
  -- ghci> is \<- Streams.'System.IO.Streams.fromList' [\"Hello, world!\"] >>= Streams.'words'
  -- ghci> replicateM 3 (Streams.'read' is)
  -- [Just \"Hello,\", Just \"world!\", Nothing]
  -- @
  --
  -- Note that this may increase the chunk size if the input contains extremely
  -- long words.
  words :: InputStream ByteString -> IO (InputStream ByteString)
  words = splitOn isSpace >=> filterM (return . not . S.all isSpace)
  
  
  ------------------------------------------------------------------------------
  -- | Intersperses string chunks sent to the given 'OutputStream' with newlines.
  -- See 'intersperse' and 'Prelude.unlines'.
  --
  -- @
  -- ghci> os <- Streams.'unlines' Streams.'System.IO.Streams.stdout'
  -- ghci> Streams.'write' (Just \"Hello,\") os
  -- Hello
  -- ghci> Streams.'write' Nothing os
  -- ghci> Streams.'write' (Just \"world!\") os
  -- world!
  -- @
  unlines :: OutputStream ByteString -> IO (OutputStream ByteString)
  unlines os = makeOutputStream $ \m -> do
      write m os
      case m of
          Nothing -> return $! ()
          Just _  -> write (Just "\n") os
  
  
  ------------------------------------------------------------------------------
  -- | Intersperses string chunks sent to the given 'OutputStream' with spaces.
  -- See 'intersperse' and 'Prelude.unwords'.
  --
  -- @
  -- ghci> os <- Streams.'unwords' Streams.'System.IO.Streams.stdout'
  -- ghci> forM_ [Just \"Hello,\", Nothing, Just \"world!\\n\"] $ \w -> Streams.'write' w os
  -- Hello, world!
  -- @
  unwords :: OutputStream ByteString -> IO (OutputStream ByteString)
  unwords = intersperse " "
  
  
  ------------------------------------------------------------------------------
  -- | Thrown by 'throwIfProducesMoreThan' when too many bytes were read from the
  -- original 'InputStream'.
  data TooManyBytesReadException = TooManyBytesReadException deriving (Typeable)
  
  instance Show TooManyBytesReadException where
      show TooManyBytesReadException = "Too many bytes read"
  
  instance Exception TooManyBytesReadException
  
  
  ------------------------------------------------------------------------------
  -- | Thrown by 'giveExactly' when too few bytes were written to the produced
  -- 'OutputStream'.
  data TooFewBytesWrittenException = TooFewBytesWrittenException deriving (Typeable)
  
  instance Show TooFewBytesWrittenException where
      show TooFewBytesWrittenException = "Too few bytes written"
  
  instance Exception TooFewBytesWrittenException
  
  
  ------------------------------------------------------------------------------
  -- | Thrown by 'throwIfConsumesMoreThan' when too many bytes were sent to the
  -- produced 'OutputStream'.
  data TooManyBytesWrittenException =
      TooManyBytesWrittenException deriving (Typeable)
  
  instance Show TooManyBytesWrittenException where
      show TooManyBytesWrittenException = "Too many bytes written"
  
  instance Exception TooManyBytesWrittenException
  
  
  ------------------------------------------------------------------------------
  -- | Thrown by 'readExactly' and 'takeExactly' when not enough bytes were
  -- available on the input.
  data ReadTooShortException = ReadTooShortException Int64 deriving (Typeable)
  
  instance Show ReadTooShortException where
      show (ReadTooShortException x) = "Short read, expected " ++ show x
                                       ++ " bytes"
  
  instance Exception ReadTooShortException
  
  
  ------------------------------------------------------------------------------
  -- | Wraps an 'InputStream'. If more than @n@ bytes are produced by this
  -- stream, 'read' will throw a 'TooManyBytesReadException'.
  --
  -- If a chunk yielded by the input stream would result in more than @n@ bytes
  -- being produced, 'throwIfProducesMoreThan' will cut the generated string such
  -- that exactly @n@ bytes are yielded by the returned stream, and the
  -- /subsequent/ read will throw an exception. Example:
  --
  -- @
  -- ghci> is \<- Streams.'System.IO.Streams.fromList' [\"abc\", \"def\", \"ghi\"] >>=
  --             Streams.'throwIfProducesMoreThan' 5
  -- ghci> 'Control.Monad.replicateM' 2 ('read' is)
  -- [Just \"abc\",Just \"de\"]
  -- ghci> Streams.'read' is
  -- *** Exception: Too many bytes read
  -- @
  --
  -- Strings pushed back to the returned 'InputStream' will be propagated
  -- upstream, modifying the count of taken bytes accordingly. Example:
  --
  -- @
  -- ghci> is  <- Streams.'System.IO.Streams.fromList' [\"abc\", \"def\", \"ghi\"]
  -- ghci> is' <- Streams.'throwIfProducesMoreThan' 5 is
  -- ghci> Streams.'read' is'
  -- Just \"abc\"
  -- ghci> Streams.'unRead' \"xyz\" is'
  -- ghci> Streams.'System.IO.Streams.peek' is
  -- Just \"xyz\"
  -- ghci> Streams.'read' is
  -- Just \"xyz\"
  -- ghci> Streams.'read' is
  -- Just \"de\"
  -- ghci> Streams.'read' is
  -- *** Exception: Too many bytes read
  -- @
  --
  throwIfProducesMoreThan
      :: Int64                    -- ^ maximum number of bytes to read
      -> InputStream ByteString   -- ^ input stream
      -> IO (InputStream ByteString)
  throwIfProducesMoreThan k0 src = do
      kref <- newIORef k0
      return $! InputStream (prod kref) (pb kref)
    where
      prod kref = read src >>= maybe (return Nothing) chunk
        where
          chunk s = do
              k <- readIORef kref
              let k'    = k - l
              case () of !_ | l == 0  -> return (Just s)
                            | k == 0  -> throwIO TooManyBytesReadException
                            | k' >= 0 -> writeIORef kref k' >> return (Just s)
                            | otherwise -> do
                                  let (!a,!b) = S.splitAt (fromEnum k) s
                                  writeIORef kref 0
                                  unRead b src
                                  return $! Just a
            where
              l     = toEnum $ S.length s
  
      pb kref s = do
          unRead s src
          modifyRef kref (+ (fromIntegral $ S.length s))
  
  
  ------------------------------------------------------------------------------
  -- | Reads an @n@-byte ByteString from an input stream. Throws a
  -- 'ReadTooShortException' if fewer than @n@ bytes were available.
  --
  -- Example:
  --
  -- @
  -- ghci> Streams.'System.IO.Streams.fromList' [\"long string\"] >>= Streams.'readExactly' 6
  -- \"long s\"
  -- ghci> Streams.'System.IO.Streams.fromList' [\"short\"] >>= Streams.'readExactly' 6
  -- *** Exception: Short read, expected 6 bytes
  -- @
  --
  readExactly :: Int                     -- ^ number of bytes to read
              -> InputStream ByteString  -- ^ input stream
              -> IO ByteString
  readExactly n input = go id n
    where
      go !dl 0  = return $! S.concat $! dl []
      go !dl k  =
          read input >>=
          maybe (throwIO $ ReadTooShortException (fromIntegral n))
                (\s -> do
                   let l = S.length s
                   if l >= k
                     then do
                       let (a,b) = S.splitAt k s
                       when (not $ S.null b) $ unRead b input
                       return $! S.concat $! dl [a]
                     else go (dl . (s:)) (k - l))
  
  
  ------------------------------------------------------------------------------
  -- | Takes from a stream until the given predicate is no longer satisfied.
  -- Returns Nothing on end-of-stream, or @Just \"\"@ if the predicate is never
  -- satisfied. See 'Prelude.takeWhile' and 'Data.ByteString.Char8.takeWhile'.
  --
  -- Example:
  --
  -- @
  -- ghci> Streams.'System.IO.Streams.fromList' [\"Hello, world!\"] >>= Streams.'takeBytesWhile' (/= ',')
  -- Just \"Hello\"
  -- ghci> import Data.Char
  -- ghci> Streams.'System.IO.Streams.fromList' [\"7 Samurai\"] >>= Streams.'takeBytesWhile' isAlpha
  -- Just \"\"
  -- ghci> Streams.'System.IO.Streams.fromList' [] >>= Streams.'takeBytesWhile' isAlpha
  -- Nothing
  -- @
  takeBytesWhile :: (Char -> Bool)          -- ^ predicate
                 -> InputStream ByteString  -- ^ input stream
                 -> IO (Maybe ByteString)
  takeBytesWhile p input = read input >>= maybe (return Nothing) (go id)
    where
      go dl !s | S.null b  = read input >>= maybe finish (go dl')
               | otherwise = unRead b input >> finish
        where
          (a, b) = S.span p s
          dl'    = dl . (a:)
          finish = return $! Just $! S.concat $! dl [a]
  
  
  ------------------------------------------------------------------------------
  -- | Wraps an 'OutputStream', producing a new stream that will pass along at
  -- most @n@ bytes to the wrapped stream, throwing any subsequent input away.
  --
  -- Example:
  --
  -- @
  -- ghci> (os :: OutputStream ByteString, getList) <- Streams.'System.IO.Streams.listOutputStream'
  -- ghci> os' <- Streams.'giveBytes' 6 os
  -- ghci> Streams.'System.IO.Streams.fromList' [\"long \", \"string\"] >>= Streams.'System.IO.Streams.connectTo' os'
  -- ghci> getList
  -- [\"long \",\"s\"]
  -- @
  giveBytes :: Int64                        -- ^ maximum number of bytes to send
                                            -- to the wrapped stream
            -> OutputStream ByteString      -- ^ output stream to wrap
            -> IO (OutputStream ByteString)
  giveBytes k0 str = do
      kref <- newIORef k0
      makeOutputStream $ sink kref
    where
      sink _ Nothing        = write Nothing str
      sink kref mb@(Just x) = do
          k <- readIORef kref
          let l  = fromIntegral $ S.length x
          let k' = k - l
          if k' < 0
            then do let a = S.take (fromIntegral k) x
                    when (not $ S.null a) $ write (Just a) str
                    writeIORef kref 0
            else writeIORef kref k' >> write mb str
  
  
  ------------------------------------------------------------------------------
  -- | Wraps an 'OutputStream', producing a new stream that will pass along
  -- exactly @n@ bytes to the wrapped stream. If the stream is sent more or fewer
  -- than the given number of bytes, the resulting stream will throw an exception
  -- (either 'TooFewBytesWrittenException' or 'TooManyBytesWrittenException')
  -- during a call to 'write'.
  --
  -- Example:
  --
  -- @
  -- ghci> is <- Streams.'System.IO.Streams.fromList' [\"ok\"]
  -- ghci> Streams.'System.IO.Streams.outputToList' (Streams.'giveExactly' 2 >=> Streams.'System.IO.Streams.connect' is)
  -- [\"ok\"]
  -- ghci> is <- Streams.'System.IO.Streams.fromList' [\"ok\"]
  -- ghci> Streams.'System.IO.Streams.outputToList' (Streams.'giveExactly' 1 >=> Streams.'System.IO.Streams.connect' is)
  -- *** Exception: Too many bytes written
  -- ghci> is <- Streams.'System.IO.Streams.fromList' [\"ok\"]
  -- ghci> Streams.'System.IO.Streams.outputToList' (Streams.'giveExactly' 3 >=> Streams.'System.IO.Streams.connect' is)
  -- *** Exception: Too few bytes written
  -- @
  giveExactly :: Int64
              -> OutputStream ByteString
              -> IO (OutputStream ByteString)
  giveExactly k0 os = do
      ref <- newIORef k0
      makeOutputStream $ go ref
    where
      go ref chunk = do
          !n <- readIORef ref
          case chunk of
            Nothing -> if n /= 0
                         then throwIO TooFewBytesWrittenException
                         else return $! ()
            Just s  -> let n' = n - fromIntegral (S.length s)
                       in if n' < 0
                            then throwIO TooManyBytesWrittenException
                            else do writeIORef ref n'
          write chunk os
  
  
  ------------------------------------------------------------------------------
  -- | Wraps an 'OutputStream', producing a new stream that will pass along at
  -- most @n@ bytes to the wrapped stream. If more than @n@ bytes are sent to the
  -- outer stream, a 'TooManyBytesWrittenException' will be thrown.
  --
  -- /Note/: if more than @n@ bytes are sent to the outer stream,
  -- 'throwIfConsumesMoreThan' will not necessarily send the first @n@ bytes
  -- through to the wrapped stream before throwing the exception.
  --
  -- Example:
  --
  -- @
  -- ghci> (os :: OutputStream ByteString, getList) <- Streams.'System.IO.Streams.listOutputStream'
  -- ghci> os' <- Streams.'throwIfConsumesMoreThan' 5 os
  -- ghci> Streams.'System.IO.Streams.fromList' [\"short\"] >>= Streams.'System.IO.Streams.connectTo' os'
  -- ghci> getList
  -- [\"short\"]
  -- ghci> os'' <- Streams.'throwIfConsumesMoreThan' 5 os
  -- ghci> Streams.'System.IO.Streams.fromList' [\"long\", \"string\"] >>= Streams.'System.IO.Streams.connectTo' os''
  -- *** Exception: Too many bytes written
  -- @
  throwIfConsumesMoreThan
      :: Int64                    -- ^ maximum number of bytes to send to the
                                  --   wrapped stream
      -> OutputStream ByteString  -- ^ output stream to wrap
      -> IO (OutputStream ByteString)
  throwIfConsumesMoreThan k0 str = do
      kref   <- newIORef k0
      makeOutputStream $ sink kref
    where
      sink _ Nothing        = write Nothing str
  
      sink kref mb@(Just x) = do
          k <- readIORef kref
          let l  = toEnum $ S.length x
          let k' = k - l
          if k' < 0
            then throwIO TooManyBytesWrittenException
            else writeIORef kref k' >> write mb str
  
  
  ------------------------------------------------------------------------------
  -- | Gets the current posix time
  getTime :: IO Double
  getTime = realToFrac `fmap` getPOSIXTime
  
  
  ------------------------------------------------------------------------------
  -- | Thrown by 'throwIfTooSlow' if input is not being produced fast enough by
  -- the given 'InputStream'.
  --
  data RateTooSlowException = RateTooSlowException deriving (Typeable)
  instance Show RateTooSlowException where
      show RateTooSlowException = "Input rate too slow"
  instance Exception RateTooSlowException
  
  
  ------------------------------------------------------------------------------
  -- | Rate-limits an input stream. If the input stream is not read from faster
  -- than the given rate, reading from the wrapped stream will throw a
  -- 'RateTooSlowException'.
  --
  -- Strings pushed back to the returned 'InputStream' will be propagated up to
  -- the original stream.
  throwIfTooSlow
      :: IO ()                   -- ^ action to bump timeout
      -> Double                  -- ^ minimum data rate, in bytes per second
      -> Int                     -- ^ amount of time in seconds to wait before
                                 --   data rate calculation takes effect
      -> InputStream ByteString  -- ^ input stream
      -> IO (InputStream ByteString)
  throwIfTooSlow !bump !minRate !minSeconds' !stream = do
      !_        <- bump
      startTime <- getTime
      bytesRead <- newIORef (0 :: Int64)
      return $! InputStream (prod startTime bytesRead) (pb bytesRead)
  
    where
      prod startTime bytesReadRef = read stream >>= maybe (return Nothing) chunk
        where
          chunk s = do
              let slen = S.length s
              now <- getTime
              let !delta = now - startTime
              nb <- readIORef bytesReadRef
              let newBytes = nb + fromIntegral slen
              when (delta > minSeconds + 1 &&
                    (fromIntegral newBytes /
                      (delta - minSeconds)) < minRate) $
                  throwIO RateTooSlowException
              -- otherwise, bump the timeout and return the input
              !_ <- bump
              writeIORef bytesReadRef newBytes
              return $! Just s
  
      pb bytesReadRef s = do
          modifyRef bytesReadRef $ \x -> x - (fromIntegral $ S.length s)
          unRead s stream
  
      minSeconds = fromIntegral minSeconds'
  