diff --git a/Streaming.hs b/Streaming.hs --- a/Streaming.hs +++ b/Streaming.hs @@ -8,6 +8,9 @@ unfold, for, construct, + replicates, + repeats, + repeatsM, -- * Transforming streams maps, @@ -26,7 +29,7 @@ iterT, -- * Splitting and joining 'Stream's - split, + splitsAt, chunksOf, concats, diff --git a/Streaming/Internal.hs b/Streaming/Internal.hs --- a/Streaming/Internal.hs +++ b/Streaming/Internal.hs @@ -8,6 +8,9 @@ -- * Introducing a stream , construct , unfold + , replicates + , repeats + , repeatsM -- * Eliminating a stream , destroy @@ -16,7 +19,6 @@ , iterT , iterTM - -- * Inspecting a stream step by step , inspect @@ -27,7 +29,7 @@ -- * Splitting streams , chunksOf - , split + , splitsAt ) where import Control.Monad @@ -254,6 +256,17 @@ Thus dissolving the segmentation into @Stream f m@ layers. > concats stream = destroy stream join (join . lift) return + +>>> S.print $ concats $ maps (cons 1776) $ chunksOf 2 (each [1..5]) +1776 +1 +2 +1776 +3 +4 +1776 +5 + -} concats :: (MonadTrans t, Monad (t m), Monad m) => @@ -261,27 +274,41 @@ concats stream = destroy stream join (join . lift) return {-# INLINE concats #-} --- | Split a succession of layers after some number, returning a streaming or --- effectful pair. -split :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m (Stream f m r) -split = loop where +{-| Split a succession of layers after some number, returning a streaming or + effectful pair. + +>>> rest <- S.print $ S.splitAt 1 $ each [1..3] +1 +>>> S.print rest +2 +3 +-} +splitsAt :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m (Stream f m r) +splitsAt = loop where loop !n stream - | n <= 1 = Return stream + | n <= 0 = Return stream | otherwise = case stream of Return r -> Return (Return r) Delay m -> Delay (liftM (loop n) m) Step fs -> case n of 0 -> Return (Step fs) _ -> Step (fmap (loop (n-1)) fs) -{-# INLINABLE split #-} +{-# INLINABLE splitsAt #-} --- | Break a stream into substreams each with n functorial layers. +{-| Break a stream into substreams each with n functorial layers. + +>>> S.print $ maps' sum' $ chunksOf 2 $ each [1,1,1,1,1,1,1] +2 +2 +2 +1 +-} chunksOf :: (Monad m, Functor f) => Int -> Stream f m r -> Stream (Stream f m) m r chunksOf n0 = loop where loop stream = case stream of Return r -> Return r Delay m -> Delay (liftM loop m) - Step fs -> Step $ Step $ fmap (fmap loop . split n0) fs + Step fs -> Step $ Step $ fmap (fmap loop . splitsAt (n0-1)) fs {-# INLINABLE chunksOf #-} {- | Make it possible to \'run\' the underlying transformed monad. A simple @@ -291,21 +318,21 @@ > loop n = do > S.yield n > s <- lift get -> liftIO $ putStr "state is: " >> print s +> liftIO $ putStr "Current state is: " >> print s > lift $ put (s + n :: Int) > loop s >>> S.print $ S.take 4 $ S.drop 4 $ debugFibs -state is: 1 -state is: 2 -state is: 3 -state is: 5 +Current state is: 1 +Current state is: 2 +Current state is: 3 +Current state is: 5 5 -state is: 8 +Current state is: 8 8 -state is: 13 +Current state is: 13 13 -state is: 21 +Current state is: 21 21 -} @@ -316,3 +343,35 @@ Return r -> lift $ Return r Delay tmstr -> hoist lift tmstr >>= distribute Step fstr -> join $ lift (Step (fmap (Return . distribute) fstr)) + +-- | Repeat a functorial layer, command or instruction forever. +repeats :: (Monad m, Functor f) => f () -> Stream f m r +repeats f = loop where + loop = Step $ fmap (\_ -> loop) f + +-- Repeat a functorial layer, command or instruction forever. +repeatsM :: (Monad m, Functor f) => m (f ()) -> Stream f m r +repeatsM mf = loop where + loop = Delay $ do + f <- mf + return $ Step $ fmap (\_ -> loop) f + +-- | Repeat a functorial layer, command or instruct several times. +replicates :: (Monad m, Functor f) => Int -> f () -> Stream f m () +replicates n f = splitsAt n (repeats f) >> return () + +{-| Construct an infinite stream by cycling a finite one + +> cycles = forever + +>>> S.print $ S.take 3 $ forever $ S.each "hi" +'h' +'i' +'h' +> S.sum $ S.take 13 $ forever $ S.each [1..3] +25 +-} + +cycles :: (Monad m, Functor f) => Stream f m () -> Stream f m r +cycles = forever + diff --git a/Streaming/Prelude.hs b/Streaming/Prelude.hs --- a/Streaming/Prelude.hs +++ b/Streaming/Prelude.hs @@ -1,4 +1,27 @@ -{-| This module is very closely modeled on Pipes.Prelude +{-| This module is very closely modeled on Pipes.Prelude. + + Import qualified thus: + +> import Streaming +> import qualified Streaming as S + + The @Streaming@ exports types, functor-general operations and some other kit; + it may clash with @free@ and @pipes-group@. + + Interoperation with @pipes@ is accomplished with this isomorphism, which + uses @Pipes.Prelude.unfoldr@ from @HEAD@: + +> Pipes.unfoldr Streaming.next :: Stream (Of a) m r -> Producer a m r +> Streaming.unfoldr Pipes.next :: Producer a m r -> Stream (Of a) m r + + Interoperation with `iostreams` is thus: + +> Streaming.reread IOStreams.read :: InputStream a -> Stream (Of a) IO () +> IOStreams.unfoldM Streaming.uncons :: Stream (Of a) IO () -> IO (InputStream a) + + A simple exit to conduit would be, for example: + +> Conduit.unfoldM Streaming.uncons :: Stream (Of a) m () -> Source m a -} {-# LANGUAGE RankNTypes, BangPatterns, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveTraversable #-} @@ -9,18 +32,19 @@ , Of (..) , lazily , strictly - + -- * Introducing streams of elements -- $producers - , each , yield + , each , unfoldr , stdinLn , readLn , fromHandle + , repeat , repeatM , replicateM - + -- * Consuming streams of elements -- $consumers , stdoutLn @@ -54,14 +78,15 @@ , chain , read , show - , seq + , cons -- * Splitting and inspecting streams of elements , next , uncons - , split + , splitAt , break , span + -- , split -- * Folds -- $folds @@ -99,6 +124,10 @@ -- * Zips , zip , zipWith + + -- * Interoperation + , reread + ) where import Streaming.Internal @@ -136,6 +165,19 @@ strictly = \(a,b) -> a :> b {-# INLINE strictly #-} +{-| Break a sequence when a element falls under a predicate, keeping the rest of + the stream as the return value. + +>>> rest <- S.print $ S.break even $ each [1,1,2,3] +1 +1 +>>> S.print rest +2 +3 + + +-} + break :: Monad m => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m (Stream (Of a) m r) break pred = loop where @@ -170,12 +212,33 @@ 'i' 'h' 'o' + +>>> S.print $ S.concat (S.each [Just 1, Nothing, Just 2, Nothing]) +1 +2 + +>>> S.print $ S.concat (S.each [Right 1, Left "error!", Right 2]) +1 +2 -} concat :: (Monad m, Foldable f) => Stream (Of (f a)) m r -> Stream (Of a) m r concat str = for str each {-# INLINE concat #-} +-- +{-| The natural @cons@ for a @Stream (Of a)@. +> cons a stream = yield a >> stream + +Useful for interoperation + +-} + +cons :: (Monad m) => a -> Stream (Of a) m r -> Stream (Of a) m r +cons a str = Step (a :> str) +{-# INLINE cons #-} + + -- --------------- -- drain -- --------------- @@ -537,10 +600,10 @@ {-| Inspect the first item in a stream of elements, without a return value. - Useful for unfolding into another streaming type. + @uncons@ provides convenient exit into another streaming type: > IOStreams.unfoldM uncons :: Stream (Of a) IO b -> IO (InputStream a) -> Conduit.unfoldM uncons :: Stream (Of o) m r -> Conduit.Source m o +> Conduit.unfoldM uncons :: Stream (Of a) m r -> Conduit.Source m a -} uncons :: Monad m => Stream (Of a) m () -> m (Maybe (a, Stream (Of a) m ())) @@ -559,7 +622,7 @@ {-| Fold a 'Stream' of numbers into their product with the return value -> mapsFold product' :: Stream (Stream (Of Int)) m r -> Stream (Of Int) m r +> maps' product' :: Stream (Stream (Of Int)) m r -> Stream (Of Int) m r -} product' :: (Monad m, Num a) => Stream (Of a) m r -> m (a,r) product' = fold' (*) 1 id @@ -579,11 +642,28 @@ -- --------------- -- repeat -- --------------- +{-| Repeat an element /ad inf./ . +>>> S.print $ S.take 3 $ S.repeat 1 +1 +1 +1 +-} + repeat :: a -> Stream (Of a) m r repeat a = loop where loop = Step (a :> loop) {-# INLINE repeat #-} + +{-| Repeat a monadic action /ad inf./, streaming its results. + +>>> L.purely fold L.list $ S.take 2 $ repeatM getLine +hello +world +["hello","world"] + +-} + repeatM :: Monad m => m a -> Stream (Of a) m r repeatM ma = loop where loop = Delay $ do @@ -611,7 +691,21 @@ return (Step $ a :> loop (n-1)) {-# INLINEABLE replicateM #-} +{-| Read an @IORef (Maybe a)@ or a similar device until it reads @Nothing@. + @reread@ provides convenient exit from the @io-streams@ library +> reread readIORef :: IORef (Maybe a) -> Stream (Of a) IO () +> reread Streams.read :: System.IO.Streams.InputStream a -> Stream (Of a) IO () +-} +reread :: Monad m => (s -> m (Maybe a)) -> s -> Stream (Of a) m () +reread step s = loop where + loop = Delay $ do + m <- step s + case m of + Nothing -> return (Return ()) + Just a -> return (Step (a :> loop)) +{-# INLINEABLE reread #-} + {-| Strict left scan, streaming, e.g. successive partial results. > Control.Foldl.purely scan :: Monad m => Fold a b -> Stream (Of a) m r -> Stream (Of b) m r @@ -639,6 +733,15 @@ {-| Strict, monadic left scan > Control.Foldl.impurely scanM :: Monad m => FoldM a m b -> Stream (Of a) m r -> Stream (Of b) m r + +>>> let v = L.impurely scanM L.vector $ each [1..4::Int] :: Stream (Of (U.Vector Int)) IO () +>>> S.print v +fromList [] +fromList [1] +fromList [1,2] +fromList [1,2,3] +fromList [1,2,3,4] + -} scanM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> Stream (Of a) m r -> Stream (Of b) m r scanM step begin done str = do @@ -695,7 +798,7 @@ {-| Fold a 'Stream' of numbers into their sum with the return value -> mapsFold sum' :: Stream (Stream (Of Int)) m r -> Stream (Of Int) m r +> maps' sum' :: Stream (Stream (Of Int)) m r -> Stream (Of Int) m r -} sum' :: (Monad m, Num a) => Stream (Of a) m r -> m (a, r) sum' = fold' (+) 0 id @@ -718,6 +821,40 @@ {-# INLINEABLE span #-} +{-| Split a succession of layers after some number, returning a streaming or +-- effectful pair. This function is the same as the 'splitsAt' exported by the +-- @Streaming@ module, but since this module is imported qualified, it can +-- usurp a Prelude name. It specializes to: + +> splitAt :: (Monad m, Functor f) => Int -> Stream (Of a) m r -> Stream (Of a) m (Stream (Of a) m r) + +-} +splitAt :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m (Stream f m r) +splitAt = splitsAt +{-# INLINE splitAt #-} + +-- {-| Split a stream of elements on each occurrence of a value, omitting the value; +-- if it appears as the last item in the stream, an empty stream will follow. +-- -} +-- split :: (Monad m, Eq a) => a -> Stream (Of a) m r -> Stream (Stream (Of a) m) m r +-- split a stream = -- loop where +-- -- loop stream = +-- case stream of +-- Return r -> Return r +-- Delay m -> Delay (liftM (split a) m) +-- Step (a' :> rest) -> if a == a' +-- then Step $ do +-- e <- lift $ inspect $ split a rest +-- case e of +-- Left r -> Return (Return r) +-- Right b -> b +-- else Step $ do +-- yield a' +-- e <- lift $ inspect $ split a rest +-- case e of +-- Left r -> Return (Return r) +-- Right b -> b + -- --------------- -- take -- --------------- @@ -749,7 +886,7 @@ --- | Convert a pure 'Stream (Of a) into a list of a +-- | Convert a pure @Stream (Of a)@ into a list of @as@ toList :: Stream (Of a) Identity () -> [a] toList = loop where @@ -759,12 +896,13 @@ Step (a :> rest) -> a : loop rest {-# INLINABLE toList #-} -{-| Convert an effectful 'Stream (Of a)' into a list of a +{-| Convert an effectful 'Stream (Of a)' into a list of @as@ - Note: 'toListM' is not an idiomatic use of @pipes@, but I provide it for - simple testing purposes. Idiomatic @pipes@ style consumes the elements - immediately as they are generated instead of loading all elements into - memory. + Note: Needless to say this function does not stream properly. + It is basically the same as 'mapM' which, like 'replicateM', + 'sequence' and similar operations on traversable containers + is a leading cause of space leaks. + -} toListM :: Monad m => Stream (Of a) m () -> m [a] toListM = fold (\diff a ls -> diff (a: ls)) id (\diff -> diff []) @@ -773,21 +911,16 @@ {-| Convert an effectful 'Stream' into a list alongside the return value - Note: 'toListM'' is not an idiomatic use of @streaming@, but I provide it for - simple testing purposes. Idiomatic @streaming@ style, like idiomatic @pipes@ style - consumes the elements as they are generated instead of loading all elements into - memory. - -> mapsFold toListM' :: Stream (Stream (Of a)) m r -> Stream (Of [a]) m +> maps' toListM' :: Stream (Stream (Of a)) m r -> Stream (Of [a]) m -} toListM' :: Monad m => Stream (Of a) m r -> m ([a], r) toListM' = fold' (\diff a ls -> diff (a: ls)) id (\diff -> diff []) {-# INLINE toListM' #-} {-| Build a @Stream@ by unfolding steps starting from a seed. - This is one natural way to consume a 'Pipes.Producer'. The - more general 'unfold' would require dealing with the left-strict pair - we are using. + This is one natural way to consume a 'Pipes.Producer'. It is worth + adding it to the functor-general 'unfold' to avoid dealing with + the left-strict pairing we are using in place of Haskell pairing. > unfoldr Pipes.next :: Monad m => Producer a m r -> Stream (Of a) m r > unfold (curry (:>) . Pipes.next) :: Monad m => Producer a m r -> Stream (Of a) m r @@ -807,7 +940,28 @@ -- yield -- --------------------------------------- --- | A singleton stream +{-| A singleton stream + +>>> S.sum $ do {S.yield 1; lift $ putStrLn "hello"; S.yield 2; lift $ putStrLn "goodbye"; S.yield 3} +hello +goodbye +6 + +>>> S.sum $ S.take 3 $ forever $ do {lift $ putStrLn "enter a number" ; n <- lift $ readLn; S.yield n } +enter a number +100 +enter a number +200 +enter a number +300 +600 + +enter a number +1 +enter a number +1000 +1001 +-} yield :: Monad m => a -> Stream (Of a) m () yield a = Step (a :> Return ()) {-# INLINE yield #-} @@ -841,12 +995,27 @@ -- IO fripperies -- -------------- --- | repeatedly stream lines as 'String' from stdin +{-| repeatedly stream lines as 'String' from stdin + +>>> S.stdoutLn $ S.show (S.each [1..3]) +1 +2 +3 + +-} stdinLn :: MonadIO m => Stream (Of String) m () stdinLn = fromHandle IO.stdin {-# INLINABLE stdinLn #-} --- | 'read' values from 'IO.stdin', ignoring failed parses +{-| Read values from 'IO.stdin', ignoring failed parses + +>>> S.sum $ S.take 2 $ forever S.readLn :: IO Int +3 +#$%^&\^? +1000 +1003 +-} + readLn :: (MonadIO m, Read a) => Stream (Of a) m () readLn = for stdinLn $ \str -> case readMaybe str of Nothing -> return () @@ -887,14 +1056,17 @@ liftIO (Prelude.print a) loop rest --- | Evaluate all values flowing downstream to WHNF -seq :: Monad m => Stream (Of a) m r -> Stream (Of a) m r -seq str = for str $ \a -> yield $! a -{-# INLINABLE seq #-} +-- -- | Evaluate all values flowing downstream to WHNF +-- seq :: Monad m => Stream (Of a) m r -> Stream (Of a) m r +-- seq str = for str $ \a -> yield $! a +-- {-# INLINABLE seq #-} -{-| Write 'String's to 'IO.stdout' using 'putStrLn' +{-| Write 'String's to 'IO.stdout' using 'putStrLn'; terminates on a broken output pipe - Unlike 'toHandle', 'stdoutLn' gracefully terminates on a broken output pipe +>>> S.stdoutLn $ S.show (S.each [1..3]) +1 +2 +3 -} stdoutLn :: MonadIO m => Stream (Of String) m () -> m () stdoutLn = loop diff --git a/streaming.cabal b/streaming.cabal --- a/streaming.cabal +++ b/streaming.cabal @@ -1,31 +1,36 @@ name: streaming -version: 0.1.0.3 +version: 0.1.0.4 cabal-version: >=1.10 build-type: Simple -synopsis: A general free monad transformer - optimized for streaming applications. +synopsis: A free monad transformer optimized for streaming applications. description: `Stream` can be used wherever `FreeT` is used. The compiler is better able to optimize operations written in terms of `Stream`. . - An associated prelude of functions, closely following - @Pipes.Prelude@ is focused on on employment with a base + See the examples in @Streaming.Prelude@ for a sense of things. + It closely follows + @Pipes.Prelude@, and focused on on employment with a base functor like @((,) a)@ (here called @Of a@) which generates effectful sequences or producers - @Pipes.Producer@, @Conduit.Source@, @IOStreams.InputStream@, @IOStreams.Generator@ and the like. . - Interoperation with `pipes` is accomplished with this isomorphism: + Interoperation with @pipes@ is accomplished with this isomorphism, which + uses @Pipes.Prelude.unfoldr@ from @HEAD@: . - > Pipes.unfoldr Streaming.next :: Stream (Of a) m r -> Producer a m r - > Stream.unfoldr Pipes.next :: Producer a m r -> Stream (Of a) m r + > Pipes.unfoldr Streaming.next :: Stream (Of a) m r -> Producer a m r + > Streaming.unfoldr Pipes.next :: Producer a m r -> Stream (Of a) m r . - Exit to `conduit` and `iostreams` is thus: + Interoperation with `iostreams` is thus: . - > Conduit.unfoldM Streaming.uncons :: Stream (Of a) m () -> Source m a - > IOStreams.unfoldM Streaming.uncons :: Stream (Of a) IO () -> InputStream a - + > Streaming.reread IOStreams.read :: InputStream a -> Stream (Of a) IO () + > IOStreams.unfoldM Streaming.uncons :: Stream (Of a) IO () -> IO (InputStream a) + . + for example. A simple exit to conduit would be, e.g.: + . + > Conduit.unfoldM Streaming.uncons :: Stream (Of a) m () -> Source m a + license: BSD3 license-file: LICENSE