System.IO.Streams.Combinators

Generic stream manipulations

Folds

inputFoldM

Arguments

:: (a -> b -> IO a)

fold function

-> a

initial seed

-> InputStream b

input stream

-> IO (InputStream b, IO a)

returns a new stream as well as an IO action to fetch the updated seed value.

A side-effecting fold over an InputStream, as a stream transformer.

The IO action returned by inputFoldM can be used to fetch the updated seed value. Example:

 ghci> is <- Streams.fromList [1, 2, 3::Int]
 ghci> (is', getSeed) <- Streams.inputFoldM (\x y -> return (x+y)) 0 is
 ghci> Streams.toList is'
 [1,2,3]
 ghci> getSeed
 6

outputFoldM

Arguments

:: (a -> b -> IO a)

fold function

-> a

initial seed

-> OutputStream b

output stream

-> IO (OutputStream b, IO a)

returns a new stream as well as an IO action to fetch the updated seed value.

A side-effecting fold over an OutputStream, as a stream transformer.

The IO action returned by outputFoldM can be used to fetch the updated seed value. Example:

 ghci> is <- Streams.fromList [1, 2, 3::Int]
 ghci> (os, getList) <- Streams.listOutputStream
 ghci> (os', getSeed) <- Streams.outputFoldM (\x y -> return (x+y)) 0 os
 ghci> Streams.connect is os'
 ghci> getList
 [1,2,3]
 ghci> getSeed
 6

fold

Arguments

:: (s -> a -> s)

fold function

-> s

initial seed

-> InputStream a

input stream

-> IO s

A left fold over an input stream. The input stream is fully consumed. See foldl.

Example:

 ghci> Streams.fromList [1..10] >>= Streams.fold (+) 0
 55

foldM

Arguments

:: (s -> a -> IO s)

fold function

-> s

initial seed

-> InputStream a

input stream

-> IO s

A side-effecting left fold over an input stream. The input stream is fully consumed. See foldl.

Example:

 ghci> Streams.fromList [1..10] >>= Streams.foldM (x y -> return (x + y)) 0
 55

any :: (a -> Bool) -> InputStream a -> IO Bool

any predicate stream returns True if any element in stream matches the predicate.

any consumes as few elements as possible, ending consumption if an element satisfies the predicate.

 ghci> is <- Streams.fromList [1, 2, 3]
 ghci> Streams.any (> 0) is    -- Consumes one element
 True
 ghci> Streams.read is
 Just 2
 ghci> Streams.any even is     -- Only 3 remains
 False

all :: (a -> Bool) -> InputStream a -> IO Bool

all predicate stream returns True if every element in stream matches the predicate.

all consumes as few elements as possible, ending consumption if any element fails the predicate.

 ghci> is <- Streams.fromList [1, 2, 3]
 ghci> Streams.all (< 0) is    -- Consumes one element
 False
 ghci> Streams.read is
 Just 2
 ghci> Streams.all odd is      -- Only 3 remains
 True

maximum :: Ord a => InputStream a -> IO (Maybe a)

maximum stream returns the greatest element in stream or Nothing if the stream is empty.

maximum consumes the entire stream.

 ghci> is <- Streams.fromList [1, 2, 3]
 ghci> Streams.maximum is
 3
 ghci> Streams.read is     -- The stream is now empty
 Nothing

minimum :: Ord a => InputStream a -> IO (Maybe a)

minimum stream returns the greatest element in stream

minimum consumes the entire stream.

 ghci> is <- Streams.fromList [1, 2, 3]
 ghci> Streams.minimum is
 1
 ghci> Streams.read is    -- The stream is now empty
 Nothing

Unfolds

unfoldM :: (b -> IO (Maybe (a, b))) -> b -> IO (InputStream a)

unfoldM f seed builds an InputStream from successively applying f to the seed value, continuing if f produces Just and halting on Nothing.

 ghci> is <- Streams.unfoldM (n -> return $ if n < 3 then Just (n, n + 1) else Nothing) 0
 ghci> Streams.toList is
 [0,1,2]

Maps

map :: (a -> b) -> InputStream a -> IO (InputStream b)

Maps a pure function over an InputStream.

map f s passes all output from s through the function f.

Satisfies the following laws:

 Streams.map (g . f) === Streams.map f >=> Streams.map g
 Streams.map id === Streams.makeInputStream . Streams.read

mapM :: (a -> IO b) -> InputStream a -> IO (InputStream b)

Maps an impure function over an InputStream.

mapM f s passes all output from s through the IO action f.

Satisfies the following laws:

 Streams.mapM (f >=> g) === Streams.mapM f >=> Streams.mapM g
 Streams.mapM return === Streams.makeInputStream . Streams.read

mapM_ :: (a -> IO b) -> InputStream a -> IO (InputStream a)

Maps a side effect over an InputStream.

mapM_ f s produces a new input stream that passes all output from s through the side-effecting IO action f.

Example:

 ghci> Streams.fromList [1,2,3] >>=
       Streams.mapM_ (putStrLn . show . (*2)) >>=
       Streams.toList
 2
 4
 6
 [1,2,3]

contramap :: (a -> b) -> OutputStream b -> IO (OutputStream a)

Contravariant counterpart to map.

contramap f s passes all input to s through the function f.

Satisfies the following laws:

 Streams.contramap (g . f) === Streams.contramap g >=> Streams.contramap f
 Streams.contramap id === return

contramapM :: (a -> IO b) -> OutputStream b -> IO (OutputStream a)

Contravariant counterpart to mapM.

contramapM f s passes all input to s through the IO action f

Satisfies the following laws:

 Streams.contramapM (f >=> g) = Streams.contramapM g >=> Streams.contramapM f
 Streams.contramapM return = return

contramapM_ :: (a -> IO b) -> OutputStream a -> IO (OutputStream a)

Equivalent to mapM_ for output.

contramapM f s passes all input to s through the side-effecting IO action f.

Filter

filter :: (a -> Bool) -> InputStream a -> IO (InputStream a)

Drops chunks from an input stream if they fail to match a given filter predicate. See filter.

Items pushed back to the returned stream are propagated back upstream.

Example:

 ghci> Streams.fromList ["the", "quick", "brown", "fox"] >>=
       Streams.filter (/= "brown") >>= Streams.toList
 ["the","quick","fox"]

filterM :: (a -> IO Bool) -> InputStream a -> IO (InputStream a)

Drops chunks from an input stream if they fail to match a given filter predicate. See filter.

Items pushed back to the returned stream are propagated back upstream.

Example:

 ghci> Streams.fromList ["the", "quick", "brown", "fox"] >>=
       Streams.filterM (return . (/= "brown")) >>= Streams.toList
 ["the","quick","fox"]

filterOutput :: (a -> Bool) -> OutputStream a -> IO (OutputStream a)

Filters output to be sent to the given OutputStream using a pure function. See filter.

Example:

 ghci> import qualified Data.ByteString.Char8 as S
 ghci> os1 <- Streams.stdout >>= Streams.'System.IO.Streams.unlines
 ghci> os2 <- os1 >>= Streams.contramap (S.pack . show) >>= Streams.filterOutput even
 ghci> Streams.write (Just 3) os2
 ghci> Streams.write (Just 4) os2
 4

filterOutputM :: (a -> IO Bool) -> OutputStream a -> IO (OutputStream a)

Filters output to be sent to the given OutputStream using a predicate function in IO. See filterM.

Example:

 ghci> let check a = putStrLn a (Allow  ++ show a ++ ?) >> readLn :: IO Bool
 ghci> import qualified Data.ByteString.Char8 as S
 ghci> os1 <- Streams.unlines Streams.stdout
 ghci> os2 <- os1 >>= Streams.contramap (S.pack . show) >>= Streams.filterOutputM check
 ghci> Streams.write (Just 3) os2
 Allow 3?
 False<Enter>
 ghci> Streams.write (Just 4) os2
 Allow 4?
 True<Enter>
 4

Takes and drops

give :: Int64 -> OutputStream a -> IO (OutputStream a)

Wraps an OutputStream, producing a new OutputStream that will pass at most n items on to the wrapped stream, subsequently ignoring the rest of the input.

take :: Int64 -> InputStream a -> IO (InputStream a)

Wraps an InputStream, producing a new InputStream that will produce at most n items, subsequently yielding end-of-stream forever.

Items pushed back to the returned InputStream will be propagated upstream, modifying the count of taken items accordingly.

Example:

 ghci> is <- Streams.fromList [1..9::Int]
 ghci> is' <- Streams.take 1 is
 ghci> Streams.read is'
 Just 1
 ghci> Streams.read is'
 Nothing
 ghci> Streams.peek is
 Just 2
 ghci> Streams.unRead 11 is'
 ghci> Streams.peek is
 Just 11
 ghci> Streams.peek is'
 Just 11
 ghci> Streams.read is'
 Just 11
 ghci> Streams.read is'
 Nothing
 ghci> Streams.read is
 Just 2
 ghci> Streams.toList is
 [3,4,5,6,7,8,9]

drop :: Int64 -> InputStream a -> IO (InputStream a)

Wraps an InputStream, producing a new InputStream that will drop the first n items produced by the wrapped stream. See drop.

Items pushed back to the returned InputStream will be propagated upstream, modifying the count of dropped items accordingly.

ignore :: Int64 -> OutputStream a -> IO (OutputStream a)

Wraps an OutputStream, producing a new OutputStream that will ignore the first n items received, subsequently passing the rest of the input on to the wrapped stream.

Zip and unzip

zip :: InputStream a -> InputStream b -> IO (InputStream (a, b))

Combines two input streams. Continues yielding elements from both input streams until one of them finishes.

zipWith :: (a -> b -> c) -> InputStream a -> InputStream b -> IO (InputStream c)

Combines two input streams using the supplied function. Continues yielding elements from both input streams until one of them finishes.

zipWithM :: (a -> b -> IO c) -> InputStream a -> InputStream b -> IO (InputStream c)

Combines two input streams using the supplied monadic function. Continues yielding elements from both input streams until one of them finishes.

unzip :: InputStream (a, b) -> IO (InputStream a, InputStream b)

Takes apart a stream of pairs, producing a pair of input streams. Reading from either of the produced streams will cause a pair of values to be pulled from the original stream if necessary. Note that reading n values from one of the returned streams will cause n values to be buffered at the other stream.

Access to the original stream is thread safe, i.e. guarded by a lock.

Utility

intersperse :: a -> OutputStream a -> IO (OutputStream a)

The function intersperse v s wraps the OutputStream s, creating a new output stream that writes its input to s interspersed with the provided value v. See intersperse.

Example:

 ghci> import Control.Monad ((>=>))
 ghci> is <- Streams.fromList ["nom", "nom", "nom"::ByteString]
 ghci> Streams.outputToList (Streams.intersperse "burp!" >=> Streams.connect is)
 ["nom","burp!","nom","burp!","nom"]

skipToEof :: InputStream a -> IO ()

Drives an InputStream to end-of-stream, discarding all of the yielded values.