io-streams 1.0.2.2 → 1.1.0.0

raw patch · 13 files changed

+462/−787 lines, 13 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• io-streams.cabal +9/−1
• src/System/IO/Streams/Builder.hs +18/−16
• src/System/IO/Streams/ByteString.hs +96/−128
• src/System/IO/Streams/Combinators.hs +45/−72
• src/System/IO/Streams/Concurrent.hs +3/−11
• src/System/IO/Streams/Debug.hs +5/−8
• src/System/IO/Streams/Internal.hs +252/−452
• src/System/IO/Streams/Internal/Search.hs +14/−24
• src/System/IO/Streams/List.hs +8/−9
• src/System/IO/Streams/Vector.hs +8/−8
• test/System/IO/Streams/Tests/ByteString.hs +0/−5
• test/System/IO/Streams/Tests/File.hs +3/−15
• test/System/IO/Streams/Tests/Internal.hs +1/−38

@@ -1,5 +1,5 @@ Name:                io-streams-Version:             1.0.2.2+Version:             1.1.0.0 License:             BSD3 License-file:        LICENSE Category:            Data, Network, IO-Streams@@ -79,6 +79,14 @@     * support for spawning processes and communicating with them using streams.   .   /ChangeLog/+  .+    [@1.1.0.0@] Reworked, simplified, and streamlined the internals of the+                library. Exports from "System.IO.Streams.Internal" relying on+                Sources and Sinks were deleted because they are no longer+                necessary: Source(..), Sink(..), defaultPushback,+                withDefaultPushback, nullSource, nullSink, singletonSource,+                simpleSource, sourceToStream, sinkToStream, generatorToSource,+                and consumerToSink.   .     [@1.0.2.2@] Fixed a bug in which \"takeBytes 0\" was erroneously requesting                 input from the wrapped stream.

@@ -86,11 +86,12 @@ import           Control.Monad                            (when) import           Data.ByteString.Char8                    (ByteString) import qualified Data.ByteString.Char8                    as S+import           Data.IORef                               (newIORef,+                                                           readIORef,+                                                           writeIORef) ------------------------------------------------------------------------------ import           System.IO.Streams.Internal               (OutputStream,-                                                           Sink (..),-                                                           nullSink,-                                                           sinkToStream,+                                                           makeOutputStream,                                                            write)  @@ -137,38 +138,39 @@                   -> OutputStream ByteString                   -> IO (OutputStream Builder) builderStreamWith (ioBuf0, nextBuf) os = do-    sinkToStream (sink ioBuf0)+    bufRef <- newIORef ioBuf0+    makeOutputStream $ sink bufRef   where-    sink ioBuf = Sink $ maybe eof chunk+    sink bufRef m = do+        buf <- readIORef bufRef+        maybe (eof buf) (chunk buf) m       where-        eof = do+        eof ioBuf = do             buf <- ioBuf             case unsafeFreezeNonEmptyBuffer buf of               Nothing    -> write Nothing os               x@(Just s) -> do-                  when (not $ S.null s) $ write x os-                  write Nothing os--            return nullSink--        chunk c = feed (unBuilder c (buildStep finalStep)) ioBuf+                 when (not $ S.null s) $ write x os+                 write Nothing os +        chunk ioBuf c = feed bufRef (unBuilder c (buildStep finalStep)) ioBuf      finalStep !(BufRange pf _) = return $! Done pf $! () -    feed bStep ioBuf = do+    feed bufRef bStep ioBuf = do         !buf   <- ioBuf         signal <- execBuildStep bStep buf          case signal of-          Done op' _ -> return $ sink (return (updateEndOfSlice buf op'))+          Done op' _ ->+              writeIORef bufRef $ (return (updateEndOfSlice buf op'))            BufferFull minSize op' bStep' -> do               let buf' = updateEndOfSlice buf op'                   {-# INLINE cont #-}                   cont = do                       ioBuf' <- nextBuf minSize buf'-                      feed bStep' ioBuf'+                      feed bufRef bStep' ioBuf'                write (Just $! unsafeFreezeBuffer buf') os               cont@@ -184,4 +186,4 @@               write (Just bs) os                ioBuf' <- nextBuf 1 buf'-              feed bStep' ioBuf'+              feed bufRef bStep' ioBuf'

@@ -68,15 +68,11 @@ ------------------------------------------------------------------------------ import           System.IO.Streams.Combinators     (filterM, intersperse,                                                     outputFoldM)-import           System.IO.Streams.Internal        (InputStream, OutputStream,-                                                    SP (..), Sink (..),-                                                    Source (..),+import           System.IO.Streams.Internal        (InputStream (..),+                                                    OutputStream,                                                     makeInputStream,-                                                    makeOutputStream,-                                                    nullSink, pushback, read,-                                                    sinkToStream,-                                                    sourceToStream, unRead,-                                                    write)+                                                    makeOutputStream, read,+                                                    unRead, write) import           System.IO.Streams.Internal.Search (MatchInfo (..), search) import           System.IO.Streams.List            (fromList, writeList) ------------------------------------------------------------------------------@@ -147,34 +143,19 @@ -- countInput :: InputStream ByteString -> IO (InputStream ByteString, IO Int64) countInput src = do-    ref    <- newIORef 0-    stream <- sourceToStream $ source ref-    return $! (stream, readIORef ref)+    ref    <- newIORef (0 :: Int64)+    return $! (InputStream (prod ref) (pb ref), readIORef ref)    where-    eof !ref = return $! SP (eofSrc ref) Nothing--    eofSrc !ref = Source {-                    produce  = eof ref-                  , pushback = pb ref-                  }+    prod ref = read src >>= maybe (return Nothing) (\x -> do+        modifyRef ref (+ (fromIntegral $ S.length x))+        return $! Just x) -    pb !ref !s = do+    pb ref s = do+        modifyRef ref (\x -> x - (fromIntegral $ S.length s))         unRead s src-        modifyRef ref $ \x -> x - (toEnum $ S.length s)-        return $! source ref -    source ref = Source {-                   produce  = read src >>= maybe (eof ref) chunk-                 , pushback = pb ref-                 }-      where-        chunk s = let !l = toEnum $ S.length s-                  in do-                      modifyRef ref (+ l)-                      return $! SP (source ref) (Just s) - ------------------------------------------------------------------------------ -- | 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@@ -234,31 +215,32 @@ takeBytes :: Int64                        -- ^ maximum number of bytes to read           -> InputStream ByteString       -- ^ input stream to wrap           -> IO (InputStream ByteString)-takeBytes k0 src = sourceToStream $ source k0+takeBytes k0 src = do+    kref <- newIORef k0+    return $! InputStream (prod kref) (pb kref)   where-    eof !n = return $! SP (eofSrc n) Nothing--    eofSrc !n = Source (eof n) (pb n)+    prod kref = do+        k <- readIORef kref+        if k <= 0+           then return Nothing+           else read src >>= maybe (return Nothing) (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 !n s = do+    pb kref s = do+        modifyRef kref (+ (fromIntegral $ S.length s))         unRead s src-        return $! source $! n + toEnum (S.length s) -    source !k = Source grab (pb k)-      where-        grab = if k <= 0-                 then eof k-                 else read src >>= maybe (eof k) chunk-        chunk s = let l  = toEnum $ S.length s-                      k' = k - l-                  in if k' <=  0-                       then let (a,b) = S.splitAt (fromEnum k) s-                            in do-                                when (not $ S.null b) $ unRead b src-                                return $! SP (eofSrc 0) (Just a)-                       else return $! SP (source k') (Just s) - ------------------------------------------------------------------------------ -- | Splits an 'InputStream' over 'ByteString's using a delimiter predicate. --@@ -473,35 +455,31 @@     :: Int64                    -- ^ maximum number of bytes to read     -> InputStream ByteString   -- ^ input stream     -> IO (InputStream ByteString)-throwIfProducesMoreThan k0 src = sourceToStream $ source k0+throwIfProducesMoreThan k0 src = do+    kref <- newIORef k0+    return $! InputStream (prod kref) (pb kref)   where-    eofSrc n = Source {-                 produce  = eof n-               , pushback = pb n-               }--    eof n = return $! SP (eofSrc n) Nothing--    pb n s = do-        unRead s src-        return $! source $! n + toEnum (S.length s)--    source !k = Source prod (pb k)+    prod kref = read src >>= maybe (return Nothing) chunk       where-        prod = read src >>= maybe (eof k) chunk--        chunk s | l == 0    = return $! SP (source k) (Just s)-                | k == 0    = throwIO TooManyBytesReadException-                | k' >= 0   = return $! SP (source k') (Just s)-                | otherwise = do-                     unRead b src-                     return $! SP (source 0) (Just a)+        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-            k'    = k - l-            (a,b) = S.splitAt (fromEnum k) 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.@@ -580,25 +558,20 @@                                           -- to the wrapped stream           -> OutputStream ByteString      -- ^ output stream to wrap           -> IO (OutputStream ByteString)-giveBytes k0 str = sinkToStream $ sink k0+giveBytes k0 str = do+    kref <- newIORef k0+    makeOutputStream $ sink kref   where-    sink !k = Sink g-      where-        g Nothing     = write Nothing str >> return nullSink--        g mb@(Just x) = let l  = toEnum $ S.length x-                            k' = k - l-                        in if k' < 0-                             then do-                                 let a = S.take (fromEnum k) x-                                 when (not $ S.null a) $ write (Just a) str-                                 return nullStr-                             else write mb str >> return (sink k')--    nullStr = Sink h--    h Nothing = write Nothing str >> return nullSink-    h _       = return nullSink+    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   ------------------------------------------------------------------------------@@ -667,17 +640,19 @@                                 --   wrapped stream     -> OutputStream ByteString  -- ^ output stream to wrap     -> IO (OutputStream ByteString)-throwIfConsumesMoreThan k0 str = sinkToStream $ sink k0+throwIfConsumesMoreThan k0 str = do+    kref   <- newIORef k0+    makeOutputStream $ sink kref   where-    sink !k = Sink g-      where-        g Nothing     = write Nothing str >> return nullSink+    sink _ Nothing        = write Nothing str -        g mb@(Just x) = let l  = toEnum $ S.length x-                            k' = k - l-                        in if k' < 0-                             then throwIO TooManyBytesWrittenException-                             else write mb str >> return (sink k')+    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   ------------------------------------------------------------------------------@@ -713,36 +688,29 @@ throwIfTooSlow !bump !minRate !minSeconds' !stream = do     !_        <- bump     startTime <- getTime--    sourceToStream $ source startTime 0+    bytesRead <- newIORef (0 :: Int64)+    return $! InputStream (prod startTime bytesRead) (pb bytesRead)    where-    minSeconds = fromIntegral minSeconds'--    source !startTime = proc+    prod startTime bytesReadRef = read stream >>= maybe (return Nothing) chunk       where-        eof !nb = return $! SP (eofSrc nb) Nothing-        eofSrc !nb = Source { produce = eof nb-                            , pushback = pb nb }--        pb !nb s = do-            unRead s stream-            return $ proc $ nb - S.length s+        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 -        proc !nb = Source prod (pb nb)-          where-            prod = read stream >>=-                   maybe (eof nb)-                         (\s -> do-                            let slen = S.length s-                            now <- getTime-                            let !delta = now - startTime-                            let !newBytes = nb + slen-                            when (delta > minSeconds + 1 &&-                                  (fromIntegral newBytes /-                                    (delta - minSeconds)) < minRate) $-                                throwIO RateTooSlowException+    pb bytesReadRef s = do+        modifyRef bytesReadRef $ \x -> x - (fromIntegral $ S.length s)+        unRead s stream -                            -- otherwise, bump the timeout and return the input-                            !_ <- bump-                            return $! SP (proc newBytes) (Just s))+    minSeconds = fromIntegral minSeconds'

@@ -58,17 +58,16 @@ import           Data.Int                   (Int64) import           Data.IORef                 (atomicModifyIORef, modifyIORef,                                              newIORef, readIORef, writeIORef)+import           Data.Maybe                 (isJust) import           Prelude                    hiding (all, any, drop, filter,                                              map, mapM, mapM_, maximum,                                              minimum, read, take, unzip, zip,                                              zipWith) -------------------------------------------------------------------------------import           System.IO.Streams.Internal (InputStream, OutputStream,-                                             SP (..), Source (..),+import           System.IO.Streams.Internal (InputStream (..), OutputStream,                                              fromGenerator, makeInputStream,-                                             makeOutputStream, read,-                                             sourceToStream, unRead, write,-                                             yield)+                                             makeOutputStream, read, unRead,+                                             write, yield)  ------------------------------------------------------------------------------ -- | A side-effecting fold over an 'OutputStream', as a stream transformer.@@ -449,26 +448,18 @@ filterM :: (a -> IO Bool)         -> InputStream a         -> IO (InputStream a)-filterM p src = sourceToStream source+filterM p src = return $! InputStream prod pb   where-    source = Source {-               produce  = prod-             , pushback = pb-             }-     prod = read src >>= maybe eof chunk      chunk s = do         b <- p s-        if b then return $! SP source (Just s)+        if b then return $! Just s              else prod -    eof = return $! flip SP Nothing Source {-            produce  = eof-          , pushback = pb-          }+    eof = return Nothing -    pb s = unRead s src >> return source+    pb s = unRead s src   ------------------------------------------------------------------------------@@ -487,26 +478,17 @@ filter :: (a -> Bool)        -> InputStream a        -> IO (InputStream a)-filter p src = sourceToStream source+filter p src = return $! InputStream prod pb   where-    source = Source {-               produce  = prod-             , pushback = pb-             }-     prod = read src >>= maybe eof chunk      chunk s = do         let b = p s-        if b then return $! SP source (Just s)+        if b then return $! Just s              else prod -    eof = return $! flip SP Nothing Source {-            produce  = eof-          , pushback = pb-          }--    pb s = unRead s src >> return source+    eof  = return Nothing+    pb s = unRead s src   ------------------------------------------------------------------------------@@ -693,18 +675,22 @@ -- @ -- take :: Int64 -> InputStream a -> IO (InputStream a)-take k0 input = sourceToStream $ source k0+take k0 input = do+    kref <- newIORef k0+    return $! InputStream (prod kref) (pb kref)   where-    eof !n = return $! SP (eofSrc n) Nothing-    eofSrc !n = Source (eof n) (pb n)-    pb !n s = do-        unRead s input-        return $! source $! n + 1+    prod kref = do+        !k <- readIORef kref+        if k <= 0+          then return Nothing+          else do+              m <- read input+              when (isJust m) $ modifyIORef kref $ \x -> x - 1+              return m -    source !k | k <= 0 = eofSrc k-              | otherwise = Source (read input >>= maybe (eof k) chunk) (pb k)-      where-        chunk x = return $! SP (source (k - 1)) (Just x)+    pb kref !s = do+       unRead s input+       modifyIORef kref (+1)   ------------------------------------------------------------------------------@@ -714,26 +700,26 @@ -- Items pushed back to the returned 'InputStream' will be propagated upstream, -- modifying the count of dropped items accordingly. drop :: Int64 -> InputStream a -> IO (InputStream a)-drop k0 input = sourceToStream $ source k0+drop k0 input = do+    kref <- newIORef k0+    return $! InputStream (prod kref) (pb kref)   where-    source !k | k <= 0    = normalSrc k-              | otherwise = Source (discard k) (pb k)-+    prod kref = do+        !k <- readIORef kref+        if k <= 0+          then getInput kref+          else discard kref -    getInput k = read input >>= maybe (eof k)-                                      (return . SP (source (k - 1)) . Just)-    normalSrc k = Source (getInput k) (pb k)+    getInput kref = do+        read input >>= maybe (return Nothing) (\c -> do+            modifyIORef kref (\x -> x - 1)+            return $! Just c) -    eof !n = return $! SP (eofSrc n) Nothing-    eofSrc !n = Source (eof n) (pb n)+    discard kref = getInput kref >>= maybe (return Nothing) (const $ prod kref) -    pb !n s = do+    pb kref s = do         unRead s input-        return $! source $! n + 1--    discard k | k <= 0    = getInput k-              | otherwise = read input >>= maybe (eof k)-                                                 (const $ discard $! k - 1)+        modifyIORef kref (+1)   ------------------------------------------------------------------------------@@ -788,24 +774,11 @@ -- /Since: 1.0.2.0/ -- atEndOfInput :: IO b -> InputStream a -> IO (InputStream a)-atEndOfInput m is = sourceToStream source+atEndOfInput m is = return $! InputStream prod pb   where-    eof = do-        let !x = SP eofSrc Nothing-        void m-        return x--    eofSrc = Source {-               produce  = eof-             , pushback = pb-             }-    pb s = unRead s is >> return source-    source = Source {-               produce  = read is >>= maybe eof chunk-             , pushback = pb-             }-      where-        chunk s = return $! SP source (Just s)+    prod    = read is >>= maybe eof (return . Just)+    eof     = void m >> return Nothing+    pb s    = unRead s is   ------------------------------------------------------------------------------

@@ -18,14 +18,11 @@ import           Control.Exception          (SomeException, mask, throwIO,                                              try) import           Control.Monad              (forM_)-import           Data.Maybe                 (isNothing) import           Prelude                    hiding (read) ------------------------------------------------------------------------------ import           System.IO.Streams.Internal (InputStream, OutputStream,-                                             SP (..), makeInputStream,-                                             makeOutputStream, nullSource,-                                             read, sourceToStream,-                                             withDefaultPushback)+                                             makeInputStream,+                                             makeOutputStream, read)  ------------------------------------------------------------------------------ -- | Writes the contents of an input stream to a channel until the input stream@@ -43,12 +40,7 @@ -- | Turns a 'Chan' into an input stream. -- chanToInput :: Chan (Maybe a) -> IO (InputStream a)-chanToInput ch = sourceToStream src-  where-    src = withDefaultPushback $ do-              mb <- readChan ch-              let src' = if isNothing mb then nullSource else src-              return $! SP src' mb+chanToInput ch = makeInputStream $! readChan ch   ------------------------------------------------------------------------------

@@ -14,7 +14,7 @@  import           Data.ByteString.Char8      (ByteString) import qualified Data.ByteString.Char8      as S-import           System.IO.Streams.Internal (InputStream, OutputStream)+import           System.IO.Streams.Internal (InputStream (..), OutputStream) import qualified System.IO.Streams.Internal as Streams  ------------------------------------------------------------------------------@@ -26,20 +26,17 @@   -> OutputStream ByteString   -- ^ stream the debug info will be sent to   -> InputStream a             -- ^ input stream   -> IO (InputStream a)-debugInput toBS name debugStream inputStream =-    Streams.sourceToStream source+debugInput toBS name debugStream inputStream = return $ InputStream produce pb   where-    source = Streams.Source produce pb-     produce = do         m <- Streams.read inputStream-        Streams.write (Just $ describe m) debugStream-        return $! Streams.SP source m+        Streams.write (Just $! describe m) debugStream+        return m      pb c = do         let s = S.concat [name, ": pushback: ", toBS c, "\n"]         Streams.write (Just s) debugStream-        Streams.unRead c inputStream >> return source+        Streams.unRead c inputStream      describe m = S.concat [name, ": got ", describeChunk m, "\n"] 

@@ -13,23 +13,11 @@ module System.IO.Streams.Internal   ( -- * Types     SP(..)-  , Source(..)-  , Sink(..)   , StreamPair      -- * About pushback     -- $pushback -    -- * Pushback functions-  , defaultPushback-  , withDefaultPushback--    -- * Basic sources and sinks-  , nullSource-  , nullSink-  , singletonSource-  , simpleSource-     -- * Input and output streams   , InputStream(..)   , OutputStream(..)@@ -42,8 +30,6 @@   , atEOF      -- * Building streams-  , sourceToStream-  , sinkToStream   , makeInputStream   , makeOutputStream   , appendInputStream@@ -65,13 +51,11 @@      -- * Generator monad   , Generator-  , generatorToSource   , fromGenerator   , yield      -- * Consumer monad   , Consumer-  , consumerToSink   , fromConsumer   , await   ) where@@ -80,15 +64,14 @@ import           Control.Applicative      (Applicative (..)) import           Control.Concurrent       (newMVar, withMVar) import           Control.Exception        (throwIO)-import           Control.Monad            (liftM, (>=>))+import           Control.Monad            (when) import           Control.Monad.IO.Class   (MonadIO (..)) import           Data.ByteString.Char8    (ByteString) import qualified Data.ByteString.Char8    as S import qualified Data.ByteString.Internal as S import qualified Data.ByteString.Unsafe   as S-import           Data.IORef               (IORef, newIORef, readIORef,-                                           writeIORef)-import           Data.Monoid              (Monoid (..))+import           Data.IORef               (newIORef, readIORef, writeIORef)+import           Data.Maybe               (isNothing) import           Data.Typeable            (Typeable) import           Data.Word                (Word8) import           Foreign.Marshal.Utils    (copyBytes)@@ -105,336 +88,8 @@ data SP a b = SP !a !b   deriving (Typeable) ---------------------------------------------------------------------------------- | A 'Source' generates values of type @c@ in the 'IO' monad.------ 'Source's wrap ordinary values in a 'Just' and signal end-of-stream by--- yielding 'Nothing'.------ All 'Source's define an optional push-back mechanism. You can assume that:------ @--- Streams.'pushback' source c >>= Streams.'produce' = 'return' (source, 'Just' c)--- @------ ... unless a 'Source' documents otherwise.------ 'Source' is to be considered an implementation detail of the library, and--- should only be used in code that needs explicit control over the 'pushback'--- semantics.------ Most library users should instead directly use 'InputStream's, which prevent--- reuse of previous 'Source's.-data Source c = Source {-      produce  :: IO (SP (Source c) (Maybe c))-    , pushback :: c -> IO (Source c)-    } deriving (Typeable) - --------------------------------------------------------------------------------- | A 'Generator' is a coroutine monad that can be used to define complex--- 'InputStream's. You can cause a value of type @Just r@ to appear when the--- 'InputStream' is read by calling 'yield':------ @--- g :: 'Generator' Int ()--- g = do---     Streams.'yield' 1---     Streams.'yield' 2---     Streams.'yield' 3--- @------ A 'Generator' can be turned into an 'InputStream' by calling--- 'fromGenerator':------ @--- m :: 'IO' ['Int']--- m = Streams.'fromGenerator' g >>= Streams.'System.IO.Streams.toList'     \-\- value returned is [1,2,3]--- @------ You can perform IO by calling 'liftIO', and turn a 'Generator' into an--- 'InputStream' with 'fromGenerator'.------ As a general rule, you should not acquire resources that need to be freed--- from a 'Generator', because there is no guarantee the coroutine continuation--- will ever be called, nor can you catch an exception from within a--- 'Generator'.-newtype Generator r a = Generator {-      unG :: IO (Either (SP r (Generator r a)) a)-    } deriving (Typeable)----------------------------------------------------------------------------------generatorBind :: Generator r a -> (a -> Generator r b) -> Generator r b-generatorBind (Generator m) f = Generator (m >>= either step value)-  where-    step (SP v r) = return $! Left $! SP v (generatorBind r f)-    value = unG .  f-{-# INLINE generatorBind #-}----------------------------------------------------------------------------------instance Monad (Generator r) where-   return = Generator . return . Right-   (>>=)  = generatorBind----------------------------------------------------------------------------------instance MonadIO (Generator r) where-    liftIO = Generator . (Right `fmap`)----------------------------------------------------------------------------------instance Functor (Generator r) where-    fmap f (Generator m) = Generator $ m >>= either step value-      where-        step (SP v m') = return $! Left $! SP v (fmap f m')-        value v        = return $! Right $! f v----------------------------------------------------------------------------------instance Applicative (Generator r) where-    pure = Generator . return . Right--    m <*> n = do-        f <- m-        v <- n-        return $! f v------------------------------------------------------------------------------------ | Calling @'yield' x@ causes the value @'Just' x@ to appear on the input--- when this generator is converted to an 'InputStream'. The rest of the--- computation after the call to 'yield' is resumed later when the--- 'InputStream' is 'read' again.-yield :: r -> Generator r ()-yield x = Generator $! return $! Left $! SP x (return $! ())------------------------------------------------------------------------------------ | Turns a 'Generator' into a 'Source' using the default pushback mechanism.-generatorToSource :: Generator r a -> Source r-generatorToSource (Generator m) = withDefaultPushback go-  where-    go              = m >>= either step finish-    finish          = const $ return $! SP nullSource Nothing-    step (SP v gen) = return $! SP (generatorToSource gen) (Just v)------------------------------------------------------------------------------------ | Turns a 'Generator' into an 'InputStream'.-fromGenerator :: Generator r a -> IO (InputStream r)-fromGenerator (Generator m) = do-    ref <- newIORef m-    makeInputStream $! go ref-  where-    go ref = readIORef ref >>= (\n -> n >>= either step finish)-      where-        step (SP v gen) = do-            writeIORef ref $! unG gen-            return $! Just v--        finish _ = return Nothing----------------------------------------------------------------------------------newtype Consumer c a = Consumer {-      unC :: IO (Either (Maybe c -> Consumer c a) a)-    } deriving (Typeable)----------------------------------------------------------------------------------instance Monad (Consumer c) where-    return = Consumer . return . Right--    (Consumer m) >>= f = Consumer $ m >>= either step value-      where-        step g  = return $! Left $! (>>= f) . g-        value v = unC $ f v----------------------------------------------------------------------------------instance MonadIO (Consumer c) where-    liftIO = Consumer . fmap Right----------------------------------------------------------------------------------instance Functor (Consumer r) where-    fmap f (Consumer m) = Consumer (m >>= either step value)-      where-        step g = return $! Left $! (fmap f) . g-        value v = return $! Right $! f v----------------------------------------------------------------------------------instance Applicative (Consumer r) where-    pure = return--    m <*> n = do-        f <- m-        v <- n-        return $! f v----------------------------------------------------------------------------------await :: Consumer r (Maybe r)-await = Consumer $ return (Left return)----------------------------------------------------------------------------------consumerToSink :: Consumer r a -> Sink r-consumerToSink (Consumer m) = Sink $ go m-  where-    go act v = act >>= either step value-      where-        value _ = return nullSink-        step f  = unC (f v) >>=-                  either (\g -> return $! Sink $! go (return $ Left g))-                         (const $ return nullSink)----------------------------------------------------------------------------------fromConsumer :: Consumer r a -> IO (OutputStream r)-fromConsumer = sinkToStream . consumerToSink------------------------------------------------------------------------------------ | A 'Sink' consumes values of type @c@ in the 'IO' monad.------ Sinks are supplied ordinary values by wrapping them in 'Just', and you--- indicate the end of the stream to a 'Sink' by supplying 'Nothing'.------ If you supply a value after a 'Nothing', the behavior is defined by the--- implementer of the given 'Sink'. (All 'Sink' definitions in this library--- will simply discard the extra input.)------ Library users should use 'OutputStream's, which prevent reuse of previous--- 'Sink's.-data Sink c = Sink {-      consume :: Maybe c -> IO (Sink c)-    } deriving (Typeable)------------------------------------------------------------------------------------ | appendSource concatenates two 'Source's, analogous to ('++') for lists.------ The second 'Source' continues where the first 'Source' ends.------ appendSource defines a monoid with 'nullSource' as the identity:------ > nullSource `appendSource` s = s--- >--- > s `appendSource` nullSource = s--- >--- >  (s1 `appendSource` s2) `appendSource` s3--- > = s1 `appendSource` (s2 `appendSource` s3)-appendSource :: Source c -> Source c -> Source c-appendSource !p !q = Source prod pb-  where-    prod = do-        (SP p' c) <- produce p-        maybe (produce q)-              (const $ return $! SP (p' `appendSource` q) c)-              c--    pb c = do-        s' <- pushback p c-        return $! s' `appendSource` q----------------------------------------------------------------------------------instance Monoid (Source a) where-    mempty  = nullSource-    mappend = appendSource---{- TODO: Define better convenience functions for pushback.  These convenience-         functions still require that the user ties the knot to correctly define-         pushback, which is error-prone for non-trivial pushback-         customizations. -}----------------------------------------------------------------------------------- | The default pushback implementation. Given a 'Source' and a value to push--- back, produces a new 'Source' that will 'produce' the value given and yield--- the original 'Source', and where 'pushback' recursively calls--- 'defaultPushback'.-defaultPushback :: Source c -> c -> IO (Source c)-defaultPushback s c = let s' = Source { produce  = return $! SP s (Just c)-                                      , pushback = defaultPushback s'-                                      }-                      in return $! s'------------------------------------------------------------------------------------ | Given an action to use as 'produce', creates a 'Source' that uses--- 'defaultPushback' as its 'pushback'.-withDefaultPushback :: IO (SP (Source c) (Maybe c)) -> Source c-withDefaultPushback prod = let s = Source prod (defaultPushback s)-                           in s------------------------------------------------------------------------------------ | If you have just an @IO (Maybe c)@ action and are happy with the default--- pushback behaviour, this function is slightly more efficient than--- using 'withDefaultPushback'. (It allocates less.)-simpleSource :: IO (Maybe c) -> IO (Source c)-simpleSource m = newIORef [] >>= \ref ->-    let s       = Source prod pb-        prod    = pop ref >>= maybe prodM prodP-        prodM   = m >>= \x -> return $!-                              maybe (SP nullSource Nothing) (const $ SP s x) x-        prodP c = return $! SP s (Just c)-        pb c    = modifyRef ref (c:) >> return s-    in return $! s--  where-    {-# INLINE pop #-}-    pop ref = readIORef ref >>= \l ->-              case l of-                []     -> return Nothing-                (x:xs) -> writeIORef ref xs >> (return $! Just x)----------------------------------------------------------------------------------{-# INLINE modifyRef #-}-modifyRef :: IORef a -> (a -> a) -> IO ()-modifyRef ref f = do-    x <- readIORef ref-    writeIORef ref $! f x------------------------------------------------------------------------------------ | An empty source that immediately yields 'Nothing'.-nullSource :: Source c-nullSource = withDefaultPushback (return $! SP nullSource Nothing)------------------------------------------------------------------------------------ | 'nullSink' discards all values it consumes.-nullSink :: Sink c-nullSink = Sink $ const $ return nullSink------------------------------------------------------------------------------------ | Transforms any value into a 1-element 'Source'.-singletonSource :: c -> Source c-singletonSource c = withDefaultPushback $ return $! SP nullSource (Just c)------------------------------------------------------------------------------------ A note for readers: why are we using IORef inside InputStream and--- OutputStream instead of MVar?------ A modifyMVar takes about 35ns to run on my Macbook, and the equivalent--- readIORef/writeIORef pair takes 6ns.------ Given that we'll be composing these often, we'll give up thread safety in--- order to gain a 6x performance improvement. If you want thread-safe access--- to a stream, you can use lockingInputStream or lockingOutputStream.---------------------------------------------------------------------------------------------------------------------------------------------------------------- -- | An 'InputStream' generates values of type @c@ in the 'IO' monad. -- --  Two primitive operations are defined on 'InputStream':@@ -449,9 +104,12 @@ -- -- @'unRead' c stream >> 'read' stream === 'return' ('Just' c)@ ---newtype InputStream  c = IS (IORef (Source c))-  deriving (Typeable)+data InputStream a = InputStream {+      _read   :: IO (Maybe a)+    , _unRead :: a -> IO ()+    } deriving (Typeable) + ------------------------------------------------------------------------------ -- | An 'OutputStream' consumes values of type @c@ in the 'IO' monad. -- The only primitive operation defined on 'OutputStream' is:@@ -465,8 +123,9 @@ -- implementer of the given 'OutputStream'. (All 'OutputStream' definitions in -- this library will simply discard the extra input.) ---newtype OutputStream c = OS (IORef (Sink   c))-  deriving (Typeable)+data OutputStream a = OutputStream {+      _write :: Maybe a -> IO ()+    } deriving (Typeable)   ------------------------------------------------------------------------------@@ -474,85 +133,19 @@ -- -- Returns either a value wrapped in a 'Just', or 'Nothing' if the end of the -- stream is reached.-read :: InputStream c -> IO (Maybe c)-read (IS ref) = do-    m       <- readIORef ref-    SP m' x <- produce m-    writeIORef ref m'-    return x+read :: InputStream a -> IO (Maybe a)+read = _read {-# INLINE read #-}   --------------------------------------------------------------------------------- | Pushes a value back onto an input stream. 'read' and 'unRead' should--- satisfy the following law, with the possible exception of side effects:------ @--- Streams.'unRead' c stream >> Streams.'read' stream === 'return' ('Just' c)--- @------ Note that this could be used to add values back to the stream that were not--- originally drawn from the stream.-unRead :: c -> InputStream c -> IO ()-unRead c (IS ref) = readIORef ref >>= f >>= writeIORef ref-  where-    f (Source _ pb) = pb c-{-# INLINE unRead #-}------------------------------------------------------------------------------------ | Converts a 'Source' to an 'InputStream'.-sourceToStream :: Source a -> IO (InputStream a)-sourceToStream = liftM IS . newIORef-{-# INLINE sourceToStream #-}------------------------------------------------------------------------------------ | Converts a 'Sink' to an 'OutputStream'.-sinkToStream :: Sink a -> IO (OutputStream a)-sinkToStream = liftM OS . newIORef-{-# INLINE sinkToStream #-}------------------------------------------------------------------------------------ | 'concatInputStreams' concatenates a list of 'InputStream's, analogous to--- ('++') for lists.------ Subsequent 'InputStream's continue where the previous one 'InputStream'--- ends.------ Note: values pushed back to the 'InputStream' returned by--- 'concatInputStreams' are not propagated to any of the source--- 'InputStream's.-concatInputStreams :: [InputStream a] -> IO (InputStream a)-concatInputStreams inputStreams = do-    ref <- newIORef inputStreams-    makeInputStream $! run ref--  where-    run ref = go-      where-        go = do-            streams <- readIORef ref-            case streams of-              []       -> return Nothing-              (s:rest) -> do-                  next <- read s-                  case next of-                    Nothing -> writeIORef ref rest >> go-                    Just _  -> return next------------------------------------------------------------------------------------ | 'appendInputStream' concatenates two 'InputStream's, analogous to ('++')--- for lists.------ The second 'InputStream' continues where the first 'InputStream' ends.+-- | Feeds a value to an 'OutputStream'. Values of type @c@ are written in an+-- 'OutputStream' by wrapping them in 'Just', and the end of the stream is+-- indicated by by supplying 'Nothing'. ----- Note: values pushed back to 'appendInputStream' are not propagated to either--- wrapped 'InputStream'.-appendInputStream :: InputStream a -> InputStream a -> IO (InputStream a)-appendInputStream s1 s2 = concatInputStreams [s1, s2]+write :: Maybe a -> OutputStream a -> IO ()+write = flip _write+{-# INLINE write #-}   ------------------------------------------------------------------------------@@ -564,22 +157,25 @@ -- @ -- Streams.'peek' stream >> Streams.'read' stream === Streams.'read' stream -- @-peek :: InputStream c -> IO (Maybe c)+peek :: InputStream a -> IO (Maybe a) peek s = do     x <- read s-    maybe (return $! ()) (\c -> unRead c s) x+    maybe (return $! ()) (_unRead s) x     return x-{-# INLINE peek #-}   --------------------------------------------------------------------------------- | Feeds a value to an 'OutputStream'. Values of type @c@ are written in an--- 'OutputStream' by wrapping them in 'Just', and the end of the stream is--- indicated by by supplying 'Nothing'.+-- | Pushes a value back onto an input stream. 'read' and 'unRead' should+-- satisfy the following law, with the possible exception of side effects: ---write :: Maybe c -> OutputStream c -> IO ()-write c (OS ref) = readIORef ref >>= (($ c) . consume) >>= writeIORef ref-{-# INLINE write #-}+-- @+-- Streams.'unRead' c stream >> Streams.'read' stream === 'return' ('Just' c)+-- @+--+-- Note that this could be used to add values back to the stream that were not+-- originally drawn from the stream.+unRead :: a -> InputStream a -> IO ()+unRead = flip _unRead   ------------------------------------------------------------------------------@@ -648,7 +244,24 @@ -- from the 'InputStream'. The given action is extended with the default -- pushback mechanism (see "System.IO.Streams.Internal#pushback"). makeInputStream :: IO (Maybe a) -> IO (InputStream a)-makeInputStream = simpleSource >=> sourceToStream+makeInputStream m = do+    doneRef <- newIORef False+    pbRef   <- newIORef []+    return $! InputStream (grab doneRef pbRef) (pb pbRef)+  where+    grab doneRef pbRef = do+        l <- readIORef pbRef+        case l of+          []     -> do done <- readIORef doneRef+                       if done+                         then return Nothing+                         else do+                             x <- m+                             when (isNothing x) $ writeIORef doneRef True+                             return x+          (x:xs) -> writeIORef pbRef xs >> (return $! Just x)++    pb ref x = readIORef ref >>= \xs -> writeIORef ref (x:xs) {-# INLINE makeInputStream #-}  @@ -657,10 +270,7 @@ -- -- (@makeOutputStream f@) runs the computation @f@ on each value fed to it. makeOutputStream :: (Maybe a -> IO ()) -> IO (OutputStream a)-makeOutputStream f = sinkToStream s-  where-    s = Sink (\x -> f x >> return s)-{-# INLINE makeOutputStream #-}+makeOutputStream = return . OutputStream   ------------------------------------------------------------------------------@@ -673,14 +283,11 @@ lockingInputStream :: InputStream a -> IO (InputStream a) lockingInputStream s = do     mv <- newMVar $! ()-    let src = Source { produce = withMVar mv $ const $ do-                           x <- read s-                           return $! SP src x-                     , pushback = \c -> withMVar mv $ const $ do-                                      unRead c s-                                      return src-                     }-    sourceToStream src+    return $! InputStream (grab mv) (pb mv)++  where+    grab mv = withMVar mv $ const $ read s+    pb mv x = withMVar mv $ const $ unRead x s {-# INLINE lockingInputStream #-}  @@ -704,16 +311,57 @@ ------------------------------------------------------------------------------ -- | An empty 'InputStream' that yields 'Nothing' immediately. nullInput :: IO (InputStream a)-nullInput = sourceToStream nullSource+nullInput = makeInputStream $ return Nothing   ------------------------------------------------------------------------------ -- | An empty 'OutputStream' that discards any input fed to it. nullOutput :: IO (OutputStream a)-nullOutput = sinkToStream nullSink+nullOutput = makeOutputStream $ const $ return $! ()   ------------------------------------------------------------------------------+-- | 'appendInputStream' concatenates two 'InputStream's, analogous to ('++')+-- for lists.+--+-- The second 'InputStream' continues where the first 'InputStream' ends.+--+-- Note: values pushed back to 'appendInputStream' are not propagated to either+-- wrapped 'InputStream'.+appendInputStream :: InputStream a -> InputStream a -> IO (InputStream a)+appendInputStream s1 s2 = concatInputStreams [s1, s2]+++------------------------------------------------------------------------------+-- | 'concatInputStreams' concatenates a list of 'InputStream's, analogous to+-- ('++') for lists.+--+-- Subsequent 'InputStream's continue where the previous one 'InputStream'+-- ends.+--+-- Note: values pushed back to the 'InputStream' returned by+-- 'concatInputStreams' are not propagated to any of the source+-- 'InputStream's.+concatInputStreams :: [InputStream a] -> IO (InputStream a)+concatInputStreams inputStreams = do+    ref <- newIORef inputStreams+    makeInputStream $! run ref++  where+    run ref = go+      where+        go = do+            streams <- readIORef ref+            case streams of+              []       -> return Nothing+              (s:rest) -> do+                  next <- read s+                  case next of+                    Nothing -> writeIORef ref rest >> go+                    Just _  -> return next+++------------------------------------------------------------------------------ -- | Checks if an 'InputStream' is at end-of-stream. atEOF :: InputStream a -> IO Bool atEOF s = read s >>= maybe (return True) (\k -> unRead k s >> return False)@@ -743,8 +391,6 @@ -- @  --                  --------------------------------------------                  -- Typeclass instances for Handle support --                  --------------------------------------------@@ -865,3 +511,157 @@                  -> IO (H.Buffer Word8) emptyWriteBuffer buf     = return buf { H.bufL=0, H.bufR=0, H.bufState = H.WriteBuffer }+++------------------------------------------------------------------------------+-- | A 'Generator' is a coroutine monad that can be used to define complex+-- 'InputStream's. You can cause a value of type @Just r@ to appear when the+-- 'InputStream' is read by calling 'yield':+--+-- @+-- g :: 'Generator' Int ()+-- g = do+--     Streams.'yield' 1+--     Streams.'yield' 2+--     Streams.'yield' 3+-- @+--+-- A 'Generator' can be turned into an 'InputStream' by calling+-- 'fromGenerator':+--+-- @+-- m :: 'IO' ['Int']+-- m = Streams.'fromGenerator' g >>= Streams.'System.IO.Streams.toList'     \-\- value returned is [1,2,3]+-- @+--+-- You can perform IO by calling 'liftIO', and turn a 'Generator' into an+-- 'InputStream' with 'fromGenerator'.+--+-- As a general rule, you should not acquire resources that need to be freed+-- from a 'Generator', because there is no guarantee the coroutine continuation+-- will ever be called, nor can you catch an exception from within a+-- 'Generator'.+newtype Generator r a = Generator {+      unG :: IO (Either (SP r (Generator r a)) a)+    } deriving (Typeable)+++------------------------------------------------------------------------------+generatorBind :: Generator r a -> (a -> Generator r b) -> Generator r b+generatorBind (Generator m) f = Generator (m >>= either step value)+  where+    step (SP v r) = return $! Left $! SP v (generatorBind r f)+    value = unG .  f+{-# INLINE generatorBind #-}+++------------------------------------------------------------------------------+instance Monad (Generator r) where+   return = Generator . return . Right+   (>>=)  = generatorBind+++------------------------------------------------------------------------------+instance MonadIO (Generator r) where+    liftIO = Generator . (Right `fmap`)+++------------------------------------------------------------------------------+instance Functor (Generator r) where+    fmap f (Generator m) = Generator $ m >>= either step value+      where+        step (SP v m') = return $! Left $! SP v (fmap f m')+        value v        = return $! Right $! f v+++------------------------------------------------------------------------------+instance Applicative (Generator r) where+    pure = Generator . return . Right++    m <*> n = do+        f <- m+        v <- n+        return $! f v+++------------------------------------------------------------------------------+-- | Calling @'yield' x@ causes the value @'Just' x@ to appear on the input+-- when this generator is converted to an 'InputStream'. The rest of the+-- computation after the call to 'yield' is resumed later when the+-- 'InputStream' is 'read' again.+yield :: r -> Generator r ()+yield x = Generator $! return $! Left $! SP x (return $! ())+++------------------------------------------------------------------------------+-- | Turns a 'Generator' into an 'InputStream'.+fromGenerator :: Generator r a -> IO (InputStream r)+fromGenerator (Generator m) = do+    ref <- newIORef m+    makeInputStream $! go ref+  where+    go ref = readIORef ref >>= (\n -> n >>= either step finish)+      where+        step (SP v gen) = do+            writeIORef ref $! unG gen+            return $! Just v++        finish _ = return Nothing+++------------------------------------------------------------------------------+newtype Consumer c a = Consumer {+      unC :: IO (Either (Maybe c -> Consumer c a) a)+    } deriving (Typeable)+++------------------------------------------------------------------------------+instance Monad (Consumer c) where+    return = Consumer . return . Right++    (Consumer m) >>= f = Consumer $ m >>= either step value+      where+        step g  = return $! Left $! (>>= f) . g+        value v = unC $ f v+++------------------------------------------------------------------------------+instance MonadIO (Consumer c) where+    liftIO = Consumer . fmap Right+++------------------------------------------------------------------------------+instance Functor (Consumer r) where+    fmap f (Consumer m) = Consumer (m >>= either step value)+      where+        step g = return $! Left $! (fmap f) . g+        value v = return $! Right $! f v+++------------------------------------------------------------------------------+instance Applicative (Consumer r) where+    pure = return++    m <*> n = do+        f <- m+        v <- n+        return $! f v+++------------------------------------------------------------------------------+await :: Consumer r (Maybe r)+await = Consumer $ return (Left return)+++------------------------------------------------------------------------------+fromConsumer :: Consumer r a -> IO (OutputStream r)+fromConsumer c0 = newIORef c0 >>= makeOutputStream . go+  where+    go ref mb = do+        c  <- readIORef ref+        c' <- unC c >>= either step (const $! return c)+        writeIORef ref c'+      where+        force c = do e <- unC c+                     return $! Consumer $! return e+        step g  = force $! g mb

@@ -7,18 +7,17 @@   ) where  ------------------------------------------------------------------------------+import           Control.Monad               (when)+import           Control.Monad.IO.Class      (liftIO) import           Data.ByteString.Char8       (ByteString) import qualified Data.ByteString.Char8       as S import           Data.ByteString.Unsafe      as S-import           Data.Monoid                 (mappend, mconcat) import qualified Data.Vector.Unboxed         as V import qualified Data.Vector.Unboxed.Mutable as MV import           Prelude                     hiding (last, read) -------------------------------------------------------------------------------import           System.IO.Streams.Internal  (InputStream, SP (..),-                                              nullSource, produce, read,-                                              singletonSource, sourceToStream,-                                              withDefaultPushback)+import           System.IO.Streams.Internal  (InputStream)+import qualified System.IO.Streams.Internal  as Streams  ------------------------------------------------------------------------------ -- | 'MatchInfo' provides match information when performing string search.@@ -67,16 +66,14 @@ search :: ByteString                   -- ^ \"needle\" to look for        -> InputStream ByteString       -- ^ input stream to wrap        -> IO (InputStream MatchInfo)-search needle stream = do-    --debug $ "boyermoore: needle=" ++ show needle-    sourceToStream (withDefaultPushback $-                    lookahead nlen >>= either finishAndEOF startSearch)+search needle stream = Streams.fromGenerator $+                       lookahead nlen >>= either finishAndEOF startSearch    where     --------------------------------------------------------------------------     finishAndEOF x = if S.null x-                       then return $! SP nullSource Nothing-                       else return $! SP nullSource (Just $! NoMatch x)+                       then return $! ()+                       else Streams.yield $! NoMatch x      --------------------------------------------------------------------------     startSearch !haystack =@@ -158,21 +155,14 @@                       else do                           let sidx = p - leftLen                           let (!crumb, rest) = S.splitAt sidx nextHaystack-                          let s1 = singletonSource $ NoMatch $-                                   S.concat [haystack, crumb]-                          let s2 = withDefaultPushback $ startSearch rest-                          produce $ s1 `mappend` s2+                          Streams.yield $! NoMatch $! S.append haystack crumb+                          startSearch rest      --------------------------------------------------------------------------     produceMatch nomatch aftermatch = do-        let !s1 = singletonSource $! NoMatch nomatch-        let !s2 = singletonSource $! Match needle-        let s3 = withDefaultPushback $ startSearch aftermatch--        produce $ mconcat $ if S.null nomatch-                              then [s2, s3]-                              else [s1, s2, s3]-+        when (not $ S.null nomatch) $ Streams.yield $! NoMatch nomatch+        Streams.yield $! Match needle+        startSearch aftermatch      --------------------------------------------------------------------------     !nlen = S.length needle@@ -195,7 +185,7 @@     --------------------------------------------------------------------------     lookahead n = go id n       where-        go dlist !k = read stream >>= maybe eof chunk+        go dlist !k = liftIO (Streams.read stream) >>= maybe eof chunk           where             eof = return $! Left $! S.concat $ dlist [] 

@@ -19,10 +19,10 @@ import           Control.Monad.IO.Class     (MonadIO (..)) import           Data.IORef                 (newIORef, readIORef, writeIORef) import           Prelude                    hiding (read)-import           System.IO.Streams.Internal (InputStream, OutputStream,Sink (..), connect, fromGenerator,-                                             makeInputStream, nullSink,-                                             read, sinkToStream,-                                              write, yield)+import           System.IO.Streams.Internal (InputStream, OutputStream, await,+                                             connect, fromConsumer,+                                             fromGenerator, makeInputStream,+                                             read, write, yield)   ------------------------------------------------------------------------------@@ -65,16 +65,15 @@ listOutputStream :: IO (OutputStream c, IO [c]) listOutputStream = do     r <- newMVar id-    c <- sinkToStream $ consumer r+    c <- fromConsumer $ consumer r     return (c, flush r)    where     consumer r = go       where-        go = Sink $ maybe (return nullSink)-                          (\c -> do-                               modifyMVar_ r $ \dl -> return (dl . (c:))-                               return go)+        go = await >>= (maybe (return $! ()) $ \c -> do+                            liftIO $ modifyMVar_ r $ \dl -> return (dl . (c:))+                            go)      flush r = modifyMVar r $ \dl -> return (id, dl []) {-# INLINE listOutputStream #-}

@@ -38,8 +38,7 @@ import           Data.Vector.Generic.Mutable (MVector) import qualified Data.Vector.Generic.Mutable as VM import           System.IO.Streams.Internal  (InputStream, OutputStream,-                                              Sink (..), fromGenerator,-                                              nullSink, sinkToStream, yield)+                                              fromGenerator, yield) import qualified System.IO.Streams.Internal  as S  @@ -220,21 +219,22 @@                                -> IO (OutputStream c, IO (v (PrimState IO) c)) mutableVectorOutputStreamSized initialSize = do     r <- vfNew initialSize >>= newMVar-    c <- sinkToStream $ consumer r+    c <- S.fromConsumer $ consumer r     return (c, flush r)    where     consumer r = go       where-        go = Sink $ maybe (return nullSink)-                          (\c -> do-                               modifyMVar_ r $ flip vfAppend c-                               return go)+        go = S.await >>=+             (maybe (return $! ()) $ \c -> do+                 liftIO $ modifyMVar_ r $ flip vfAppend c+                 go)+     flush r = modifyMVar r $ \vfi -> do                                 !v   <- vfFinish vfi                                 vfi' <- vfNew initialSize                                 return $! (vfi', v)-{-# INLINE mutableVectorOutputStream #-}+{-# INLINE mutableVectorOutputStreamSized #-}   ------------------------------------------------------------------------------

@@ -286,11 +286,6 @@              l' <- liftM L.fromChunks grab              assertEqual "throwIfConsumesMoreThan" l l' -             -- cover nullSink behaviour-             write (Just "blah") os'-             nil <- liftM L.fromChunks grab-             assertEqual "nil after eof" "" nil-  ------------------------------------------------------------------------------ testGiveExactly :: Test

@@ -1,10 +1,8 @@-{-# LANGUAGE BangPatterns      #-} {-# LANGUAGE OverloadedStrings #-}  module System.IO.Streams.Tests.File (tests) where  -------------------------------------------------------------------------------import           Control.Concurrent import           Control.Exception import           Control.Monad                  hiding (mapM) import           Data.ByteString.Char8          (ByteString)@@ -16,7 +14,6 @@ import           System.FilePath import           System.IO import           System.IO.Streams              hiding (intersperse, mapM_)-import           System.IO.Streams.Internal import           Test.Framework import           Test.Framework.Providers.HUnit import           Test.HUnit                     hiding (Test)@@ -32,18 +29,9 @@ ------------------------------------------------------------------------------ copyingListOutputStream :: IO (OutputStream ByteString, IO [ByteString]) copyingListOutputStream = do-    r <- newMVar id-    c <- sinkToStream $ consumer r-    return (c, flush r)--  where-    consumer r = Sink $ maybe (return nullSink)-                              (\c0 -> do-                                   let !c = S.copy c0-                                   modifyMVar_ r $ \dl -> return (dl . (c:))-                                   return $ consumer r)--    flush r = modifyMVar r $ \dl -> return (id, dl [])+    (os, grab) <- listOutputStream+    os' <- contramap S.copy os >>= lockingOutputStream+    return (os', grab)   ------------------------------------------------------------------------------

@@ -8,7 +8,6 @@ import           Control.Monad                  hiding (mapM) import           Control.Monad.IO.Class         (liftIO) import           Data.IORef-import           Data.Monoid import           Prelude                        hiding (mapM, read) import           Test.Framework import           Test.Framework.Providers.HUnit@@ -19,43 +18,19 @@ import           System.IO.Streams.Tests.Common  tests :: [Test]-tests = [ testSourceConcat-        , testAppendInput+tests = [ testAppendInput         , testConst         , testCoverLockingStream         , testPeek         , testNullInput         , testGenerator         , testGeneratorInstances-        , testGeneratorSource         , testConsumer         , testTrivials         ]   -------------------------------------------------------------------------------testSourceConcat :: Test-testSourceConcat = testCase "internal/sourceConcat" $ do-    is  <- sourceToStream $ mconcat $-           map singletonSource [1::Int, 2, 3]--    unRead 7 is--    l   <- toList is--    assertEqual "sourceConcat" [7,1,2,3] l--    is' <- sourceToStream $ mconcat $-           map singletonSource [1::Int, 2, 3]--    unRead 7 is'--    l'  <- toList is'--    assertEqual "sourceConcat2" [7,1,2,3] l'--------------------------------------------------------------------------------- testAppendInput :: Test testAppendInput = testCase "internal/appendInputStream" $ do     s1 <- fromList [1::Int, 2, 3]@@ -141,16 +116,6 @@   -------------------------------------------------------------------------------testGeneratorSource :: Test-testGeneratorSource = testCase "internal/generatorSource" $ do-    let src = generatorToSource $ sequence $-              Prelude.map ((>>= yield) . (liftIO . return)) [1..5::Int]-    is  <- sourceToStream src-    toList is >>= assertEqual "generator" [1..5]-    read is >>= assertEqual "read after EOF" Nothing--------------------------------------------------------------------------------- testGeneratorInstances :: Test testGeneratorInstances = testCase "internal/generatorInstances" $ do     fromGenerator g1 >>= toList@@ -197,8 +162,6 @@ testTrivials = testCase "internal/trivials" $ do     coverTypeableInstance (undefined :: InputStream Int)     coverTypeableInstance (undefined :: OutputStream Int)-    coverTypeableInstance (undefined :: Source Int)-    coverTypeableInstance (undefined :: Sink Int)     coverTypeableInstance (undefined :: Generator Int ())     coverTypeableInstance (undefined :: Consumer Int ())     coverTypeableInstance (undefined :: SP Int Int)