streaming 0.1.0.3 → 0.1.0.4
raw patch · 4 files changed
+304/−65 lines, 4 files
Files
- Streaming.hs +4/−1
- Streaming/Internal.hs +77/−18
- Streaming/Prelude.hs +206/−34
- streaming.cabal +17/−12
Streaming.hs view
@@ -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,
Streaming/Internal.hs view
@@ -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+
Streaming/Prelude.hs view
@@ -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
streaming.cabal view
@@ -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