diff --git a/io-streams.cabal b/io-streams.cabal --- a/io-streams.cabal +++ b/io-streams.cabal @@ -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. diff --git a/src/System/IO/Streams/Builder.hs b/src/System/IO/Streams/Builder.hs --- a/src/System/IO/Streams/Builder.hs +++ b/src/System/IO/Streams/Builder.hs @@ -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' diff --git a/src/System/IO/Streams/ByteString.hs b/src/System/IO/Streams/ByteString.hs --- a/src/System/IO/Streams/ByteString.hs +++ b/src/System/IO/Streams/ByteString.hs @@ -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' diff --git a/src/System/IO/Streams/Combinators.hs b/src/System/IO/Streams/Combinators.hs --- a/src/System/IO/Streams/Combinators.hs +++ b/src/System/IO/Streams/Combinators.hs @@ -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 ------------------------------------------------------------------------------ diff --git a/src/System/IO/Streams/Concurrent.hs b/src/System/IO/Streams/Concurrent.hs --- a/src/System/IO/Streams/Concurrent.hs +++ b/src/System/IO/Streams/Concurrent.hs @@ -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 ------------------------------------------------------------------------------ diff --git a/src/System/IO/Streams/Debug.hs b/src/System/IO/Streams/Debug.hs --- a/src/System/IO/Streams/Debug.hs +++ b/src/System/IO/Streams/Debug.hs @@ -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"] diff --git a/src/System/IO/Streams/Internal.hs b/src/System/IO/Streams/Internal.hs --- a/src/System/IO/Streams/Internal.hs +++ b/src/System/IO/Streams/Internal.hs @@ -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 diff --git a/src/System/IO/Streams/Internal/Search.hs b/src/System/IO/Streams/Internal/Search.hs --- a/src/System/IO/Streams/Internal/Search.hs +++ b/src/System/IO/Streams/Internal/Search.hs @@ -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 [] diff --git a/src/System/IO/Streams/List.hs b/src/System/IO/Streams/List.hs --- a/src/System/IO/Streams/List.hs +++ b/src/System/IO/Streams/List.hs @@ -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 #-} diff --git a/src/System/IO/Streams/Vector.hs b/src/System/IO/Streams/Vector.hs --- a/src/System/IO/Streams/Vector.hs +++ b/src/System/IO/Streams/Vector.hs @@ -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 #-} ------------------------------------------------------------------------------ diff --git a/test/System/IO/Streams/Tests/ByteString.hs b/test/System/IO/Streams/Tests/ByteString.hs --- a/test/System/IO/Streams/Tests/ByteString.hs +++ b/test/System/IO/Streams/Tests/ByteString.hs @@ -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 diff --git a/test/System/IO/Streams/Tests/File.hs b/test/System/IO/Streams/Tests/File.hs --- a/test/System/IO/Streams/Tests/File.hs +++ b/test/System/IO/Streams/Tests/File.hs @@ -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) ------------------------------------------------------------------------------ diff --git a/test/System/IO/Streams/Tests/Internal.hs b/test/System/IO/Streams/Tests/Internal.hs --- a/test/System/IO/Streams/Tests/Internal.hs +++ b/test/System/IO/Streams/Tests/Internal.hs @@ -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)