diff --git a/Streaming.hs b/Streaming.hs --- a/Streaming.hs +++ b/Streaming.hs @@ -4,38 +4,41 @@ -- * Free monad transformer -- $stream Stream, - -- * Constructing a 'Stream' on a base functor + -- * Constructing a 'Stream' on a given functor unfold, - construct, - for, yields, replicates, repeats, repeatsM, - mwrap, + effect, wrap, + streamBuild, -- * Transforming streams decompose, maps, mapsM, + mapped, distribute, - separate, - unseparate, groups, -- * Inspecting a stream inspect, - -- * Zipping streams - zips, + + -- * Zipping and unzipping streams zipsWith, + zips, + unzips, interleaves, + separate, + unseparate, -- * Eliminating a 'Stream' iterTM, iterT, destroy, + streamFold, mapsM_, run, @@ -45,19 +48,29 @@ chunksOf, concats, intercalates, - + -- period, + -- periods, + -- * Base functor for streams of individual items Of (..), lazily, strictly, + -- * ResourceT help + bracketStream, + -- * re-exports MFunctor(..), MMonad(..), MonadTrans(..), MonadIO(..), Compose(..), + MonadThrow(..), + MonadResource(..), + MonadBase(..), + ResourceT(..), + runResourceT, join, liftA2, liftA3, @@ -72,32 +85,43 @@ import Control.Monad.Trans import Data.Functor.Compose +import Control.Monad.Base +import Control.Monad.Trans.Resource {- $stream - The 'Stream' data type is equivalent to @FreeT@ and can represent any effectful - succession of steps, where the form of the steps or 'commands' is - specified by the first (functor) parameter. The present module exports - functions that pertain to that general case. So for example, if the - functor is + The 'Stream' data type can be used to represent any effectful + succession of steps arising in some monad. + The form of the steps is specified by the first (\"functor\") + parameter in @Stream f m r@, the monad of effects by the second. + This module exports combinators that pertain to that general case. + Some of these are quite abstract and pervade any use of the library, + e.g. - > data Split r = Split r r +> maps :: (forall x . f x -> g x) -> Stream f m r -> Stream g m r +> mapped :: (forall x . f x -> m (g x)) -> Stream f m r -> Stream g m r +> hoist :: (forall x . m x -> n x) -> Stream f m r -> Stream f n r +> concats :: Stream (Stream f m) m r -> Stream f m r + + (assuming here and thoughout that @m@ or @n@ satisfies a @Monad@ constraint, and + @f@ or @g@ a @Functor@ constraint.) - The @Stream Split m r@ will the type of binary trees with @r@ at the leaves - and in which each episode of branching results from an @m@-effect. - + Others are surprisingly determinate in content: +> chunksOf :: Int -> Stream f m r -> Stream (Stream f m) m r +> splits :: - In the simplest case, the base functor is @ (,) a @. Here the news - or /command/ at each step is an /individual element of type/ @ a @, - i.e. the command is a @yield@ statement. The associated - @Streaming@ 'Streaming.Prelude' - uses the left-strict pair @Of a b@ in place of the Haskell pair @(a,b)@ + One way to see that /any/ streaming library needs some such general type is + that it is required to represent the segmentation of a stream, and to + express the equivalents of @Prelude/Data.List@ combinators that involve + 'lists of lists' and the like. The module @Streaming.Prelude@ exports + combinators relating to -and operations like e.g. +> Stream (Of a) m r -> chunksOf :: Monad m => Int -> Stream f m r -> Stream (Stream f m) m r -> mapsM Streaming.Prelude.length' :: Stream (Stream (Of a) m) r -> Stream (Of Int) m r + where @Of a r = !a :> r@ is a left-strict pair. + +> mapped Streaming.Prelude.length :: Stream (Stream (Of a) m) r -> Stream (Of Int) m r -} diff --git a/Streaming/Internal.hs b/Streaming/Internal.hs --- a/Streaming/Internal.hs +++ b/Streaming/Internal.hs @@ -1,19 +1,20 @@ {-# LANGUAGE RankNTypes, StandaloneDeriving,DeriveDataTypeable, BangPatterns #-} -{-# LANGUAGE UndecidableInstances, CPP #-} -- for show, data instances +{-# LANGUAGE UndecidableInstances, CPP, FlexibleInstances, MultiParamTypeClasses #-} module Streaming.Internal ( -- * The free monad transformer -- $stream Stream (..) -- * Introducing a stream - , construct , unfold , replicates , repeats , repeatsM - , mwrap + , effect , wrap , yields + , streamBuild + , cycles -- * Eliminating a stream , intercalates @@ -21,6 +22,7 @@ , iterT , iterTM , destroy + , streamFold -- * Inspecting a stream wrap by wrap , inspect @@ -28,12 +30,11 @@ -- * Transforming streams , maps , mapsM + , mapped , decompose , mapsM_ , run , distribute - , separate - , unseparate , groups -- , groupInL @@ -41,16 +42,26 @@ , chunksOf , splitsAt , takes + -- , period + -- , periods - -- * Zipping streams + -- * Zipping and unzipping streams , zipsWith , zips + , unzips , interleaves + , separate + , unseparate + -- * Assorted Data.Functor.x help , switch + -- * ResourceT help + + , bracketStream + -- * For use in implementation , unexposed , hoistExposed @@ -74,6 +85,12 @@ import Prelude hiding (splitAt) import Data.Functor.Compose import Data.Functor.Sum +-- import Data.Time (getCurrentTime, diffUTCTime, picosecondsToDiffTime, addUTCTime) + +import Control.Monad.Base +import Control.Monad.Trans.Resource +import Control.Monad.Catch (MonadCatch (..)) + {- $stream The 'Stream' data type is equivalent to @FreeT@ and can represent any effectful @@ -133,6 +150,8 @@ Step f -> Step (fmap loop f) {-# INLINABLE (>>) #-} (>>=) = _bind + {-#INLINE (>>=) #-} + -- stream >>= f = -- loop stream where -- loop stream0 = case stream0 of @@ -142,7 +161,8 @@ -- {-# INLINABLE (>>=) #-} fail = lift . fail - + {-#INLINE fail #-} + _bind :: (Functor f, Monad m) @@ -154,7 +174,8 @@ Step fstr -> Step (fmap go fstr) Effect m -> Effect (m >>= \s -> return (go s)) Return r -> f r - +{-#INLINABLE[0] _bind #-} + {-# RULES "_bind (Step fstr) f" forall fstr f . _bind (Step fstr) f = Step (fmap (\p -> _bind p f) fstr); @@ -170,7 +191,7 @@ pure = Return {-# INLINE pure #-} streamf <*> streamx = do {f <- streamf; x <- streamx; return (f x)} - {-# INLINABLE (<*>) #-} + {-# INLINE (<*>) #-} -- stra0 *> strb = loop stra0 where -- loop stra = case stra of -- Return _ -> strb @@ -207,21 +228,47 @@ instance (MonadIO m, Functor f) => MonadIO (Stream f m) where liftIO = Effect . liftM Return . liftIO {-# INLINE liftIO #-} + +instance (MonadBase b m, Functor f) => MonadBase b (Stream f m) where + liftBase = effect . fmap return . liftBase + {-#INLINE liftBase #-} + +instance (MonadThrow m, Functor f) => MonadThrow (Stream f m) where + throwM = lift . throwM + {-#INLINE throwM #-} + +instance (MonadCatch m, Functor f) => MonadCatch (Stream f m) where + catch str f = go str + where + go p = case p of + Step f -> Step (fmap go f) + Return r -> Return r + Effect m -> Effect (catch (do + p' <- m + return (go p')) + (\e -> return (f e)) ) + {-#INLINABLE catch #-} + +instance (MonadResource m, Functor f) => MonadResource (Stream f m) where + liftResourceT = lift . liftResourceT + {-#INLINE liftResourceT #-} + +bracketStream :: (Functor f, MonadResource m) => + IO a -> (a -> IO ()) -> (a -> Stream f m b) -> Stream f m b +bracketStream alloc free inside = do + (key, seed) <- lift (allocate alloc free) + clean key (inside seed) + where + clean key = loop where + loop str = case str of + Return r -> Effect (release key >> return (Return r)) + Effect m -> Effect (liftM loop m) + Step f -> Step (fmap loop f) +{-#INLINABLE bracketStream #-} + {-| Map a stream directly to its church encoding; compare @Data.List.foldr@ - It permits distinctions that should be hidden, as can be seen from - e.g. - -isPure stream = destroy_ (const True) (const False) (const True) - - and similar nonsense. The crucial - constraint is that the @m x -> x@ argument is an /Eilenberg-Moore algebra/. - See Atkey "Reasoning about Stream Processing with Effects" - - The destroy exported by the safe modules is - -destroy str = destroy (observe str) -} destroy :: (Functor f, Monad m) => @@ -234,42 +281,58 @@ {-# INLINABLE destroy #-} -{-| 'destroyWith' reorders the arguments of 'destroy' to be more akin +{-| 'streamFold' reorders the arguments of 'destroy' to be more akin to @foldr@ It is more convenient to query in ghci to figure out - what kind of \'algebra\' you need to write. + what kind of \'algebra\' you need to write. ->>> :t destroyWith join return +>>> :t streamFold return join (Monad m, Functor f) => (f (m a) -> m a) -> Stream f m a -> m a -- iterT ->>> :t destroyWith (join . lift) return + +>>> :t streamFold return (join . lift) (Monad m, Monad (t m), Functor f, MonadTrans t) => (f (t m a) -> t m a) -> Stream f m a -> t m a -- iterTM ->>> :t destroyWith effect return -(Monad m, Functor f, Functor f1) => - (f (Stream f1 m r) -> Stream f1 m r) -> Stream f m r -> Stream f1 m r ->>> :t destroyWith effect return (wrap . lazily) -Monad m => - Stream (Of a) m r -> Stream ((,) a) m r ->>> :t destroyWith effect return (wrap . strictly) -Monad m => - Stream ((,) a) m r -> Stream (Of a) m r ->>> :t destroyWith Data.ByteString.Streaming.effect return + +>>> :t streamFold return effect +(Monad m, Functor f, Functor g) => + (f (Stream g m r) -> Stream g m r) -> Stream f m r -> Stream g m r + +>>> :t \f -> streamFold return effect (wrap . f) +(Monad m, Functor f, Functor g) => + (f (Stream g m a) -> g (Stream g m a)) + -> Stream f m a -> Stream g m a -- maps + +>>> :t \f -> streamFold return effect (effect . liftM wrap . f) +(Monad m, Functor f, Functor g) => + (f (Stream g m a) -> m (g (Stream g m a))) + -> Stream f m a -> Stream g m a -- mapped + + So for example, when we realize that + +>>> :t streamFold return Q.mwrap (Monad m, Functor f) => - (f (ByteString m r) -> ByteString m r) -> Stream f m r -> ByteString m r ->>> :t destroyWith Data.ByteString.Streaming.effect return (\(a:>b) -> consChunk a b) -Monad m => - Stream (Of B.ByteString) m r -> ByteString m r -- fromChunks + (f (Q.ByteString m a) -> Q.ByteString m a) + -> Stream f m a -> Q.ByteString m a + + it is easy to see how to write @fromChunks@: + +>>> streamFold return Q.mwrap (\(a:>b) -> Q.chunk a >> b) +Monad m => Stream (Of B.ByteString) m a -> Q.ByteString m a -- fromChunks -} -destroyWith +streamFold :: (Functor f, Monad m) => - (m b -> b) -> (r -> b) -> (f b -> b) -> Stream f m r -> b -destroyWith effect done construct stream = destroy stream construct effect done + (r -> b) -> (m b -> b) -> (f b -> b) -> Stream f m r -> b +streamFold done effect construct stream = destroy stream construct effect done +{-#INLINE streamFold #-} --- | Reflect a church-encoded stream; cp. @GHC.Exts.build@ -construct +{- | Reflect a church-encoded stream; cp. @GHC.Exts.build@ + +> destroy a b c (streamBuild psi) = +-} +streamBuild :: (forall b . (f b -> b) -> (m b -> b) -> (r -> b) -> b) -> Stream f m r -construct = \phi -> phi Step Effect Return -{-# INLINE construct #-} +streamBuild = \phi -> phi Step Effect Return +{-# INLINE streamBuild #-} {-| Inspect the first stage of a freely layered sequence. @@ -296,9 +359,9 @@ > unfold (curry (:>) . Pipes.next) :: Monad m => Producer a m r -> Stream (Of a) m r -} - unfold :: (Monad m, Functor f) - => (s -> m (Either r (f s))) -> s -> Stream f m r + => (s -> m (Either r (f s))) + -> s -> Stream f m r unfold step = loop where loop s0 = Effect $ do e <- step s0 @@ -308,7 +371,13 @@ {-# INLINABLE unfold #-} --- | Map layers of one functor to another with a transformation +{- | Map layers of one functor to another with a transformation. Compare + hoist, which has a similar effect on the 'monadic' parameter. + +> maps id = id +> maps f . maps g = maps (f . g) + +-} maps :: (Monad m, Functor f) => (forall x . f x -> g x) -> Stream f m r -> Stream g m r maps phi = loop where @@ -319,27 +388,6 @@ {-# INLINABLE maps #-} - - --- newtype NT g f = NT {runNT :: forall x . f x -> g x} --- newtype NTM g m f = NTM {runNTM :: forall x . f x -> m (g x)} --- compNTNT :: NT f g -> NT g h -> NT f h --- compNTNT (NT f) (NT g) = NT (f . g) --- compNTNTM :: Monad m => NT f g -> NTM g m h -> NTM f m h --- compNTNTM (NT f) (NTM g) = NTM (liftM f . g) --- compNTMNT :: Monad m => NTM f m g -> NT g h -> NTM f m h --- compNTMNT (NTM f) (NT g) = NTM (f . g) --- compNTMNTM :: Monad m => NTM f m g -> NTM g m h -> NTM f m h --- compNTMNTM (NTM f) (NTM g) = NTM (f <=< g) --- --- {-# NOINLINE [0] mapsNT #-} --- mapsNT :: (Functor f, Functor g, Monad m) => NT g f -> Stream f m r -> Stream g m r --- mapsNT (NT phi) = loop where --- loop stream = case stream of --- Return r -> Return r --- Effect m -> Effect (liftM loop m) --- Step f -> Step (phi (fmap loop f)) - {- | Map layers of one functor to another with a transformation involving the base monad @maps@ is more fundamental than @mapsM@, which is best understood as a convenience for effecting this frequent composition: @@ -354,6 +402,26 @@ Step f -> Effect (liftM Step (phi (fmap loop f))) {-# INLINABLE mapsM #-} +{- | Map layers of one functor to another with a transformation involving the base monad + @maps@ is more fundamental than @mapped@, which is best understood as a convenience + for effecting this frequent composition: + +> mapped = mapsM +> mapsM phi = decompose . maps (Compose . phi) + + @mapped@ obeys these rules: + +> mapped return = id +> mapped f . mapped g = mapped (f <=< g) +> map f . mapped g = mapped (liftM f . g) +> mapped f . map g = mapped (f . g) + +-} + +mapped :: (Monad m, Functor f) => (forall x . f x -> m (g x)) -> Stream f m r -> Stream g m r +mapped = mapsM +{-#INLINE mapped #-} + {-| Resort a succession of layers of the form @m (f x)@. Though @mapsM@ is best understood as: @@ -385,11 +453,11 @@ {-# INLINABLE run #-} -{-| Map each layer to an effect in the base monad, and run them all. +{-| Map each layer to an effect, and run them all. -} mapsM_ :: (Functor f, Monad m) => (forall x . f x -> m x) -> Stream f m r -> m r -mapsM_ f str = run (maps f str) -{-# INLINABLE mapsM_ #-} +mapsM_ f = run . maps f +{-# INLINE mapsM_ #-} {-| Interpolate a layer at each segment. This specializes to e.g. @@ -397,7 +465,7 @@ > intercalates :: (Monad m, Functor f) => Stream f m () -> Stream (Stream f m) m r -> Stream f m r -} intercalates :: (Monad m, Monad (t m), MonadTrans t) => - t m a -> Stream (t m) m b -> t m b + t m x -> Stream (t m) m r -> t m r intercalates sep = go0 where go0 f = case f of @@ -415,9 +483,9 @@ go1 f' {-# INLINABLE intercalates #-} -{-| Specialized fold +{-| Specialized fold following the usage of @Control.Monad.Trans.Free@ -> iterTM alg stream = destroy stream alg (join . lift) return +> iterTM alg = streamFold return (join . lift) -} iterTM :: (Functor f, Monad m, MonadTrans t, @@ -426,9 +494,9 @@ iterTM out stream = destroyExposed stream out (join . lift) return {-# INLINE iterTM #-} -{-| Specialized fold +{-| Specialized fold following the usage of @Control.Monad.Trans.Free@ -> iterT alg stream = destroy stream alg join return +> iterT alg = streamFold return join alg -} iterT :: (Functor f, Monad m) => (f (m a) -> m a) -> Stream f m a -> m a @@ -437,18 +505,6 @@ {-| Dissolves the segmentation into layers of @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 :: (Monad m, Functor f) => Stream (Stream f m) m r -> Stream f m r concats = loop where @@ -466,6 +522,20 @@ >>> S.print rest 2 3 + +> splitAt 0 = return +> splitAt n >=> splitAt m = splitAt (m+n) + + Thus, e.g. + +>>> rest <- S.print $ splitsAt 2 >=> splitsAt 2 $ each [1..5] +1 +2 +3 +4 +>>> S.print rest +5 + -} splitsAt :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m (Stream f m r) splitsAt = loop where @@ -478,15 +548,40 @@ 0 -> Return (Step fs) _ -> Step (fmap (loop (n-1)) fs) {-# INLINABLE splitsAt #-} - + +{- Functor-general take. + + @takes 3@ can take three individual values + +>>> S.print $ takes 3 $ each [1..] +1 +2 +3 + + + or three sub-streams + +>>> S.print $ mapped S.toList $ takes 3 $ chunksOf 2 $ each [1..] +[1,2] +[3,4] +[5,6] + + Or, using 'Data.ByteString.Streaming.Char' (here called @Q@), + three byte streams. + +>>> Q.stdout $ Q.unlines $ takes 3 $ Q.lines $ Q.chunk "a\nb\nc\nd\ne\nf" +a +b +c + +-} takes :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m () takes n = void . splitsAt n {-# INLINE takes #-} {-| 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 +>>> S.print $ mapped S.sum $ chunksOf 2 $ each [1,1,1,1,1] 2 2 1 @@ -494,9 +589,9 @@ 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 - Effect m -> Effect (liftM loop m) - Step fs -> Step $ Step $ fmap (fmap loop . splitsAt (n0-1)) fs + Return r -> Return r + Effect m -> Effect (liftM loop m) + Step fs -> Step (Step (fmap (fmap loop . splitsAt (n0-1)) fs)) {-# INLINABLE chunksOf #-} {- | Make it possible to \'run\' the underlying transformed monad. @@ -529,12 +624,7 @@ > 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 @@ -598,11 +688,13 @@ -mwrap :: (Monad m, Functor f ) => m (Stream f m r) -> Stream f m r -mwrap = Effect +effect :: (Monad m, Functor f ) => m (Stream f m r) -> Stream f m r +effect = Effect +{-#INLINE effect #-} wrap :: (Monad m, Functor f ) => f (Stream f m r) -> Stream f m r wrap = Step +{-#INLINE wrap #-} {-| Lift for items in the base functor. Makes a singleton or @@ -614,6 +706,7 @@ yields :: (Monad m, Functor f) => f r -> Stream f m r yields fr = Step (fmap Return fr) +{-#INLINE yields #-} zipsWith :: (Monad m, Functor h) @@ -678,49 +771,34 @@ with the streaming on the other functor as the governing monad. This is useful for acting on one or the other functor with a fold. ->>> let odd_even = S.maps (S.distinguish even) $ S.each [1..10] ->>> :t S.effects $ separate odd_even +>>> let odd_even = S.maps (S.distinguish even) $ S.each [1..10::Int] +>>> :t separate odd_even +separate odd_even + :: Monad m => Stream (Of Int) (Stream (Of Int) m) () Now, for example, it is convenient to fold on the left and right values separately: ->>> toListM' $ toList' (separate odd_even) +>>> toList $ toList $ separate odd_even [2,4,6,8,10] :> ([1,3,5,7,9] :> ()) ->>> S.toListM' $ S.print $ separate $ odd_even -1 -3 -5 -7 -9 -[2,4,6,8,10] :> () - - We can easily use this device in place of filter: - -> filter = S.effects . separate . maps (distinguish f) - ->>> :t hoist S.effects $ separate odd_even -hoist S.effects $ separate odd_even :: Monad n => Stream (Of Int) n () ->>> S.print $ effects $ separate odd_even -2 -4 -6 -8 -10 ->>> S.print $ hoist effects $ separate odd_even -1 -3 -5 -7 -9 + We can achieve the above effect more simply + in the case of @Stream (Of a) m r@ by using 'Streaming.Prelude.duplicate' + +>>> S.toList . S.filter even $ S.toList . S.filter odd $ S.duplicate $ each [1..10::Int] +[2,4,6,8,10] :> ([1,3,5,7,9] :> ()) + + + But 'separate' and 'unseparate' are functor-general. + -} separate :: (Monad m, Functor f, Functor g) => Stream (Sum f g) m r -> Stream f (Stream g m) r separate str = destroyExposed str - (\x -> case x of InL fss -> wrap fss; InR gss -> mwrap (yields gss)) - (mwrap . lift) + (\x -> case x of InL fss -> wrap fss; InR gss -> effect (yields gss)) + (effect . lift) return -{-#INLINE separate #-} +{-#INLINABLE separate #-} unseparate :: (Monad m, Functor f, Functor g) => Stream f (Stream g m) r -> Stream (Sum f g) m r unseparate str = destroyExposed @@ -728,11 +806,20 @@ (wrap . InL) (join . maps InR) return -{-#INLINE unseparate #-} - +{-#INLINABLE unseparate #-} + + +unzips :: (Monad m, Functor f, Functor g) => + Stream (Compose f g) m r -> Stream f (Stream g m) r +unzips str = destroyExposed + str + (\(Compose fgstr) -> Step (fmap (Effect . yields) fgstr)) + (Effect . lift) + return +{-#INLINABLE unzips #-} + {-| Group layers in an alternating stream into adjoining sub-streams of one type or another. -= -} groups :: (Monad m, Functor f, Functor g) => Stream (Sum f g) m r @@ -767,7 +854,8 @@ Left r -> return (return r) Right (InL fstr) -> return (wrap (InL fstr)) Right (InR gstr) -> wrap (fmap loop gstr) - +{-#INLINABLE groups #-} + -- groupInL :: (Monad m, Functor f, Functor g) -- => Stream (Sum f g) m r -- -> Stream (Sum (Stream f m) g) m r @@ -789,3 +877,76 @@ -- Left r -> return (return r) -- Right (InL fstr) -> wrap (fmap loop fstr) -- Right (InR gstr) -> return (wrap (InR gstr)) + +-- {-| Permit streamed actions to proceed unless the clock has run out. +-- +-- -} +-- period :: (MonadIO m, Functor f) => Double -> Stream f m r -> Stream f m (Stream f m r) +-- period seconds str = do +-- utc <- liftIO getCurrentTime +-- let loop s = do +-- utc' <- liftIO getCurrentTime +-- if diffUTCTime utc' utc > (cutoff / 1000000000) +-- then return s +-- else case s of +-- Return r -> Return (Return r) +-- Effect m -> Effect (liftM loop m) +-- Step f -> Step (fmap loop f) +-- loop str +-- where +-- cutoff = fromInteger (truncate (1000000000 * seconds)) +-- {-#INLINABLE period #-} +-- +-- +-- {-| Divide a succession of phases according to a specified time interval. If time runs out +-- while an action is proceeding, it is allowed to run to completion. The clock is only then +-- restarted. +-- -} +-- periods :: (MonadIO m, Functor f) => Double -> Stream f m r -> Stream (Stream f m) m r +-- periods seconds s = do +-- utc <- liftIO getCurrentTime +-- loop (addUTCTime cutoff utc) s +-- +-- where +-- cutoff = fromInteger (truncate (1000000000 * seconds)) / 1000000000 +-- loop final stream = do +-- utc <- liftIO getCurrentTime +-- if utc > final +-- then loop (addUTCTime cutoff utc) stream +-- else case stream of +-- Return r -> Return r +-- Effect m -> Effect $ liftM (loop final) m +-- Step fstr -> Step $ fmap (periods seconds) (cutoff_ final (Step fstr)) +-- +-- -- do +-- -- let sloop s = do +-- -- utc' <- liftIO getCurrentTime +-- -- if final < utc' +-- -- then return s +-- -- else case s of +-- -- Return r -> Return (Return r) +-- -- Effect m -> Effect (liftM sloop m) +-- -- Step f -> Step (fmap sloop f) +-- -- Step (Step (fmap (fmap (periods seconds) . sloop) fstr)) +-- -- str <- m +-- -- utc' <- liftIO getCurrentTime +-- -- if diffUTCTime utc' utc > (cutoff / 1000000000) +-- -- then return (loop utc' str) +-- -- else return (loop utc str) +-- -- Step fs -> do +-- -- let check str = do +-- -- utc' <- liftIO getCurrentTime +-- -- loop utc' str +-- -- +-- {-# INLINABLE periods #-} +-- +-- cutoff_ final str = do +-- let loop s = do +-- utc' <- liftIO getCurrentTime +-- if utc' > final +-- then Return s +-- else case s of +-- Return r -> Return (Return r) +-- Effect m -> Effect (liftM loop m) +-- Step f -> Step (fmap loop f) +-- loop str diff --git a/Streaming/Prelude.hs b/Streaming/Prelude.hs --- a/Streaming/Prelude.hs +++ b/Streaming/Prelude.hs @@ -1,1850 +1,2378 @@ -{-| This module is very closely modeled on Pipes.Prelude; it attempts to - simplify and optimize the conception of Producer manipulation contained - in Pipes.Group, Pipes.Parse and the like. This is very simple and unmysterious; - it is independent of piping and conduiting, and can be used with any - rational \"streaming IO\" system. - - Import qualified thus: - -> import Streaming -> import qualified Streaming.Prelude as S - - For the examples below, one sometimes needs - -> import Streaming.Prelude (each, yield, stdoutLn, stdinLn) - - Other libraries that come up in passing are - -> import qualified Control.Foldl as L -- cabal install foldl -> import qualified Pipes as P -> import qualified Pipes.Prelude as P -> import qualified System.IO as IO - - Here are some correspondences between the types employed here and elsewhere: - -> streaming | pipes | conduit | io-streams -> ------------------------------------------------------------------------------------------------------------------- -> Stream (Of a) m () | Producer a m () | Source m a | InputStream a -> | ListT m a | ConduitM () o m () | Generator r () -> ------------------------------------------------------------------------------------------------------------------- -> Stream (Of a) m r | Producer a m r | ConduitM () o m r | Generator a r -> ------------------------------------------------------------------------------------------------------------------- -> Stream (Of a) m (Stream (Of a) m r) | Producer a m (Producer a m r) | -> -------------------------------------------------------------------------------------------------------------------- -> Stream (Stream (Of a) m) r | FreeT (Producer a m) m r | -> -------------------------------------------------------------------------------------------------------------------- -> -------------------------------------------------------------------------------------------------------------------- -> ByteString m () | Producer ByteString m () | Source m ByteString | InputStream ByteString -> -------------------------------------------------------------------------------------------------------------------- -> --} -{-# LANGUAGE RankNTypes, BangPatterns, DeriveDataTypeable, TypeFamilies, - DeriveFoldable, DeriveFunctor, DeriveTraversable #-} - -module Streaming.Prelude ( - -- * Types - Of (..) - - -- * Introducing streams of elements - -- $producers - , yield - , each - , unfoldr - , stdinLn - , readLn - , fromHandle - , iterate - , repeat - , replicate - , cycle - , repeatM - , replicateM - , enumFrom - , enumFromThen - - -- * Consuming streams of elements - -- $consumers - , stdoutLn - , stdoutLn' - , mapM_ - , print - , toHandle - , effects - , drained - - -- * Stream transformers - -- $pipes - , map - , mapM - , chain - , maps - , sequence - , mapFoldable - , filter - , filterM - , for - , delay - , take - , takeWhile --- , takeWhile' - , drop - , dropWhile - , concat - -- , elemIndices - -- , findIndices - , scan - , scanM - , scanned - , read - , show - , cons - , duplicate - - -- * Splitting and inspecting streams of elements - , next - , uncons - , splitAt - , split - , breaks - , break - , breakWhen - , span - , group - , groupBy - , groupedBy - , timed - -- , split - - - -- * Sum and Compose manipulation - - , distinguish - , switch - , separate - , unseparate - , eitherToSum - , sumToCompose - , composeToSum - - -- * Folds - -- $folds - , fold - , fold_ - , foldM - , foldM_ - , sum - , sum_ - , product - , product_ - , length - , length_ - , toList - , toList_ - , mconcat - , mconcat_ - , foldrM - , foldrT - - - -- , all - -- , any - -- , and - -- , or - -- , elem - -- , notElem - -- , find - -- , findIndex - -- , head - -- , index - -- , last - -- , length - -- , maximum - -- , minimum - -- , null - - -- * Zips - , zip - , zipWith - , zip3 - , zipWith3 - - -- * Pair manipulation - , lazily - , strictly - , fst' - , snd' - - -- * Interoperation - , reread - - -- * Basic Type - , Stream - - ) where -import Streaming.Internal - -import Control.Monad hiding (filterM, mapM, mapM_, foldM, foldM_, replicateM, sequence) -import Data.Data ( Data, Typeable ) -import Data.Functor.Identity -import Data.Functor.Sum -import Control.Monad.Trans -import Control.Applicative (Applicative (..)) -import Data.Functor (Functor (..), (<$)) - -import qualified Prelude as Prelude -import Data.Foldable (Foldable) -import Data.Traversable (Traversable) -import qualified Data.Foldable as Foldable -import Text.Read (readMaybe) -import Prelude hiding (map, mapM, mapM_, filter, drop, dropWhile, take, mconcat, sum, product - , iterate, repeat, cycle, replicate, splitAt - , takeWhile, enumFrom, enumFromTo, enumFromThen, length - , print, zipWith, zip, zipWith3, zip3, seq, show, read - , readLn, sequence, concat, span, break) - -import qualified GHC.IO.Exception as G -import qualified System.IO as IO -import Foreign.C.Error (Errno(Errno), ePIPE) -import Control.Exception (throwIO, try) -import Data.Monoid (Monoid (mappend, mempty)) -import Data.String (IsString (..)) -import Control.Concurrent (threadDelay) -import Data.Time (getCurrentTime, diffUTCTime, picosecondsToDiffTime) -import Data.Functor.Classes -import Data.Functor.Compose --- | A left-strict pair; the base functor for streams of individual elements. -data Of a b = !a :> b - deriving (Data, Eq, Foldable, Ord, - Read, Show, Traversable, Typeable) -infixr 5 :> - -instance (Monoid a, Monoid b) => Monoid (Of a b) where - mempty = mempty :> mempty - {-#INLINE mempty #-} - mappend (m :> w) (m' :> w') = mappend m m' :> mappend w w' - {-#INLINE mappend #-} - -instance Functor (Of a) where - fmap f (a :> x) = a :> f x - {-#INLINE fmap #-} - a <$ (b :> x) = b :> a - {-#INLINE (<$) #-} - -instance Monoid a => Applicative (Of a) where - pure x = mempty :> x - {-#INLINE pure #-} - m :> f <*> m' :> x = mappend m m' :> f x - {-#INLINE (<*>) #-} - m :> x *> m' :> y = mappend m m' :> y - {-#INLINE (*>) #-} - m :> x <* m' :> y = mappend m m' :> x - {-#INLINE (<*) #-} - -instance Monoid a => Monad (Of a) where - return x = mempty :> x - {-#INLINE return #-} - m :> x >> m' :> y = mappend m m' :> y - {-#INLINE (>>) #-} - m :> x >>= f = let m' :> y = f x in mappend m m' :> y - {-#INLINE (>>=) #-} - -instance (r ~ (), Monad m, f ~ Of Char) => IsString (Stream f m r) where - fromString = each - -instance (Eq a) => Eq1 (Of a) where eq1 = (==) -instance (Ord a) => Ord1 (Of a) where compare1 = compare -instance (Read a) => Read1 (Of a) where readsPrec1 = readsPrec -instance (Show a) => Show1 (Of a) where showsPrec1 = showsPrec - -{-| Note that 'lazily', 'strictly', 'fst'', and 'mapOf' are all so-called /natural transformations/ on the primitive @Of a@ functor - If we write - -> type f ~~> g = forall x . f x -> g x - - then we can restate some types as follows: - -> mapOf :: (a -> b) -> Of a ~~> Of b -- bifunctor lmap -> lazily :: Of a ~~> (,) a -> Identity . fst' :: Of a ~~> Identity a - - Manipulation of a @Stream f m r@ by mapping often turns on recognizing natural transformations of @f@, - thus @maps@ is far more general the the @map@ of the present module, which can be - defined thus: - -> S.map :: (a -> b) -> Stream (Of a) m r -> Stream (Of b) m r -> S.map f = maps (mapOf f) - - This rests on recognizing that @mapOf@ is a natural transformation; note though - that it results in such a transformation as well: - -> S.map :: (a -> b) -> Stream (Of a) m ~> Stream (Of b) m - --} -lazily :: Of a b -> (a,b) -lazily = \(a:>b) -> (a,b) -{-# INLINE lazily #-} - -strictly :: (a,b) -> Of a b -strictly = \(a,b) -> a :> b -{-# INLINE strictly #-} - -fst' :: Of a b -> a -fst' (a :> b) = a - -snd' :: Of a b -> b -snd' (a :> b) = b - -mapOf :: (a -> b) -> Of a r -> Of b r -mapOf f (a:> b) = (f a :> b) - -{-| 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 - loop str = case str of - Return r -> Return (Return r) - Effect m -> Effect $ liftM loop m - Step (a :> rest) -> if (pred a) - then Return (Step (a :> rest)) - else Step (a :> loop rest) -{-# INLINEABLE break #-} - -{-| Yield elements, using a fold to maintain state, until the accumulated - value satifies the supplied predicate. The fold will then be short-circuited - and the element that breaks it will be included with the stream returned. - This function is easiest to use with 'Control.Foldl.purely' - ->>> rest <- S.print $ L.purely S.breakWhen L.sum even $ S.each [1,2,3,4] -1 -2 ->>> S.print rest -3 -4 - --} -breakWhen :: Monad m => (x -> a -> x) -> x -> (x -> b) -> (b -> Bool) -> Stream (Of a) m r -> Stream (Of a) m (Stream (Of a) m r) -breakWhen step begin done pred = loop0 begin - where - loop0 x stream = case stream of - Return r -> return (return r) - Effect mn -> Effect $ liftM (loop0 x) mn - Step (a :> rest) -> loop a (step x a) rest - loop a !x stream = do - if pred (done x) - then return (yield a >> stream) - else case stream of - Return r -> yield a >> return (return r) - Effect mn -> Effect $ liftM (loop a x) mn - Step (a' :> rest) -> do - yield a - loop a' (step x a') rest -{-# INLINABLE breakWhen #-} - -{- Break during periods where the predicate is not satisfied, grouping the periods when it is. - ->>> S.print $ mapsM S.toList $ S.breaks not $ S.each [False,True,True,False,True,True,False] -[True,True] -[True,True] ->>> S.print $ mapsM S.toList $ S.breaks id $ S.each [False,True,True,False,True,True,False] -[False] -[False] -[False] --} -breaks - :: Monad m => - (a -> Bool) -> Stream (Of a) m r -> Stream (Stream (Of a) m) m r -breaks thus = loop where - loop stream = Effect $ do - e <- next stream - return $ case e of - Left r -> Return r - Right (a, p') -> - if not (thus a) - then Step $ fmap loop (yield a >> break thus p') - else loop p' -{-#INLINABLE breaks #-} - - -{-| Apply an action to all values flowing downstream - - ->>> S.product (S.chain Prelude.print (S.each [2..4])) >>= Prelude.print -2 -3 -4f -24 :> () - --} - -chain :: Monad m => (a -> m ()) -> Stream (Of a) m r -> Stream (Of a) m r -chain f = loop where - loop str = case str of - Return r -> return r - Effect mn -> Effect (liftM loop mn) - Step (a :> rest) -> Effect $ do - f a - return (Step (a :> loop rest)) -{-# INLINE chain #-} - -{-| Make a stream of traversable containers into a stream of their separate elements. - This is just - -> concat = for str each - ->>> S.print $ S.concat (each ["xy","z"]) -'x' -'y' -'z' - - Note that it also has the effect of 'Data.Maybe.catMaybes' and 'Data.Either.rights' - - ->>> S.print $ S.concat $ S.each [Just 1, Nothing, Just 2] -1 -2 ->>> S.print $ S.concat $ S.each [Right 1, Left "Error!", Right 2] -1 -2 - - @concat@ is not to be confused with the functor-general - -> concats :: (Monad m, Functor f) => Stream (Stream f m) m r -> Stream f m r -- specializing - ->>> S.stdoutLn $ concats $ maps (<* yield "--\n--") $ chunksOf 2 $ S.show (each [1..5]) -1 -2 --- --- -3 -4 --- --- -5 --- --- --} - -concat :: (Monad m, Foldable.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: - -> Data.Text.foldr S.cons (return ()) :: Text -> Stream (Of Char) m () -> Lazy.foldrChunks S.cons (return ()) :: Lazy.ByteString -> Stream (Of Strict.ByteString) m () - - and so on. --} - -cons :: (Monad m) => a -> Stream (Of a) m r -> Stream (Of a) m r -cons a str = Step (a :> str) -{-# INLINE cons #-} - -{- | Cycle repeatedly through the layers of a stream, /ad inf./ This - function is functor-general - -> cycle = forever - ->>> rest <- S.print $ S.splitAt 3 $ S.cycle (yield True >> yield False) -True -False -True ->>> S.print $ S.take 3 rest -False -True -False - --} - -cycle :: (Monad m, Functor f) => Stream f m r -> Stream f m s -cycle = forever - - -{-| Effect each element by the supplied number of seconds. -mapM :: Monad m => (a -> m b) -> Stream (Of a) m r -> Stream (Of b) m r - --} -delay :: MonadIO m => Double -> Stream (Of a) m r -> Stream (Of a) m r -delay seconds = mapM go where - go a = liftIO (threadDelay (truncate (seconds * 1000000))) >> return a --- --------------- --- effects --- --------------- - -{- | Reduce a stream, performing its actions but ignoring its elements. - This might just be called @effects@ or @runEffects@. - ->>> let effect = lift (putStrLn "Effect!") ->>> let stream = do {yield 1; effect; yield 2; effect; return (2^100)} - ->>> S.effects stream -Effect! -Effect! -1267650600228229401496703205376 - ->>> S.effects $ S.takeWhile (<2) stream -Effect! - --} -effects :: Monad m => Stream (Of a) m r -> m r -effects = loop where - loop stream = case stream of - Return r -> return r - Effect m -> m >>= loop - Step (_ :> rest) -> loop rest -{-#INLINABLE effects #-} - -{-| Where a transformer returns a stream, run the effects of the stream, keeping - the return value. This is usually used at the type - -> drained :: Monad m => Stream (Of a) m (Stream (Of b) m r) -> Stream (Of a) m r - -> drained = join . fmap (lift . effects) - ->>> let take' n = S.drained . S.splitAt n ->>> S.print $ concats $ maps (take' 1) $ S.group $ S.each "wwwwarrrrr" -'w' -'a' -'r' - - --} -drained :: (Monad m, Monad (t m), Functor (t m), MonadTrans t) => t m (Stream (Of a) m r) -> t m r -drained = join . fmap (lift . effects) -{-#INLINE drained #-} - --- --------------- --- drop --- --------------- - --- | Ignore the first n elements of a stream, but carry out the actions -drop :: (Monad m) => Int -> Stream (Of a) m r -> Stream (Of a) m r -drop = loop where - loop 0 stream = stream - loop n stream = case stream of - Return r -> Return r - Effect ma -> Effect (liftM (loop n) ma) - Step (a :> as) -> loop (n-1) as -{-# INLINE drop #-} - --- --------------- --- dropWhile --- --------------- - -{- | Ignore elements of a stream until a test succeeds. - ->>> IO.withFile "distribute.hs" IO.ReadMode $ S.stdoutLn . S.take 2 . S.dropWhile (isPrefixOf "import") . S.fromHandle -main :: IO () -main = do - - --} -dropWhile :: Monad m => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m r -dropWhile pred = loop where - loop stream = case stream of - Return r -> Return r - Effect ma -> Effect (liftM loop ma) - Step (a :> as) -> if pred a - then loop as - else Step (a :> as) -{-# INLINE dropWhile #-} - --- --------------- --- each --- --------------- - -{- | Stream the elements of a foldable container. - ->>> S.print $ S.map (*100) $ each [1..3] -100 -200 -300 - ->>> S.print $ S.map (*100) $ each [1..3] >> lift readLn >>= yield -100 -200 -300 -4 -400 --} -each :: (Monad m, Foldable.Foldable f) => f a -> Stream (Of a) m () -each = Foldable.foldr (\a p -> Step (a :> p)) (Return ()) -{-# INLINE each #-} - --- ----- --- enumFrom --- ------ - -{-| An infinite stream of enumerable values, starting from a given value. - @Streaming.Prelude.enumFrom@ is more desirable that @each [x..]@ for - the infinite case, because it has a polymorphic return type. - ->>> S.print $ S.take 3 $ S.enumFrom 'a' -'a' -'b' -'c' - - Because their return type is polymorphic, @enumFrom@ and @enumFromThen@ - are useful for example with @zip@ - and @zipWith@, which require the same return type in the zipped streams. - With @each [1..]@ the following would be impossible. - ->>> rest <- S.print $ S.zip (S.enumFrom 'a') $ S.splitAt 3 $ S.enumFrom 1 -('a',1) -('b',2) -('c',3) ->>> S.print $ S.take 3 rest -4 -5 -6 - - Where a final element is specified, as in @each [1..10]@ a special combinator - is unneeded, since the return type would be @()@ anyway. - --} -enumFrom :: (Monad m, Enum n) => n -> Stream (Of n) m r -enumFrom = loop where - loop !n = Step (n :> loop (succ n)) -{-# INLINEABLE enumFrom #-} - - -{-| An infinite sequence of enumerable values at a fixed distance, determined - by the first and second values. See the discussion of 'Streaming.enumFrom' - ->>> S.print $ S.take 3 $ S.enumFromThen 100 200 -100 -200 -300 - --} -enumFromThen:: (Monad m, Enum a) => a -> a -> Stream (Of a) m r -enumFromThen first second = Streaming.Prelude.map toEnum (loop _first) - where - _first = fromEnum first - _second = fromEnum second - diff = _second - _first - loop !s = Step (s :> loop (s+diff)) -{-# INLINEABLE enumFromThen #-} - --- --------------- --- filter --- --------------- - --- | Skip elements of a stream that fail a predicate -filter :: (Monad m) => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m r -filter pred = loop where - loop str = case str of - Return r -> Return r - Effect m -> Effect (liftM loop m) - Step (a :> as) -> if pred a - then Step (a :> loop as) - else loop as -{-# INLINE filter #-} - --- --------------- --- filterM --- --------------- - --- | Skip elements of a stream that fail a monadic test -filterM :: (Monad m) => (a -> m Bool) -> Stream (Of a) m r -> Stream (Of a) m r -filterM pred = loop where - loop str = case str of - Return r -> Return r - Effect m -> Effect $ liftM loop m - Step (a :> as) -> Effect $ do - bool <- pred a - if bool - then return $ Step (a :> loop as) - else return $ loop as -{-# INLINEABLE filterM #-} - --- --------------- --- fold --- --------------- - -{- $folds - Use these to fold the elements of a 'Stream'. - ->>> S.fold_ (+) 0 id $ S.each [1..0] -50 - - The general folds 'fold', fold_', 'foldM' and 'foldM_' are arranged - for use with 'Control.Foldl' - ->>> L.purely fold_ L.sum $ each [1..10] -55 ->>> L.purely fold_ (liftA3 (,,) L.sum L.product L.list) $ each [1..10] -(55,3628800,[1,2,3,4,5,6,7,8,9,10]) - - All functions marked with an underscore omit - (e.g. @fold_@, @sum_@) the stream's return value in a left-strict pair. - They are good for exiting streaming completely, - but when you are, e.g. @mapsM@-ing over a @Stream (Stream (Of a) m) m r@, - which is to be compared with @[[a]]@. Specializing, we have e.g. - -> mapsM sum :: (Monad m, Num n) => Stream (Stream (Of Int)) IO () -> Stream (Of n) IO () -> mapsM (fold mappend mempty id) :: Stream (Stream (Of Int)) IO () -> Stream (Of Int) IO () - ->>> S.print $ mapsM S.sum $ chunksOf 3 $ S.each [1..10] -6 -15 -24 -10 - ->>> let three_folds = L.purely S.fold (liftA3 (,,) L.sum L.product L.list) ->>> S.print $ mapsM three_folds $ chunksOf 3 (each [1..10]) -(6,6,[1,2,3]) -(15,120,[4,5,6]) -(24,504,[7,8,9]) -(10,10,[10]) --} - -{-| Strict fold of a 'Stream' of elements - -> Control.Foldl.purely fold :: Monad m => Fold a b -> Stream (Of a) m () -> m b --} -fold_ :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> m b -fold_ step begin done stream0 = loop stream0 begin - where - loop !stream !x = case stream of - Return r -> return (done x) - Effect m -> m >>= \s -> loop s x - Step (a :> rest) -> loop rest (step x a) -{-# INLINABLE fold_ #-} - -{-| Strict fold of a 'Stream' of elements that preserves the return value. - ->>> S.sum $ each [1..10] -55 :> () - ->>> (n :> rest) <- S.sum $ S.splitAt 3 (each [1..10]) ->>> print n -6 ->>> (m :> rest') <- S.sum $ S.splitAt 3 rest ->>> print m -15 ->>> S.print rest' -7 -8 -9 - - The type provides for interoperation with the foldl library. - -> Control.Foldl.purely fold :: Monad m => Fold a b -> Stream (Of a) m r -> m (Of b r) - - Thus, specializing a bit: - -> L.purely fold L.sum :: Stream (Of Int) Int r -> m (Of Int r) -> maps (L.purely fold L.sum) :: Stream (Stream (Of Int)) IO r -> Stream (Of Int) IO r - - ->>> S.print $ mapsM (L.purely S.fold (liftA2 (,) L.list L.sum)) $ chunksOf 3 $ each [1..10] -([1,2,3],6) -([4,5,6],15) -([7,8,9],24) -([10],10) --} - -fold :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> m (Of b r) -fold step begin done s0 = loop s0 begin - where - loop stream !x = case stream of - Return r -> return (done x :> r) - Effect m -> m >>= \s -> loop s x - Step (a :> rest) -> loop rest (step x a) -{-# INLINABLE fold #-} - -{-| Strict, monadic fold of the elements of a 'Stream (Of a)' - -> Control.Foldl.impurely foldM :: Monad m => FoldM a b -> Stream (Of a) m () -> m b --} -foldM_ - :: Monad m - => (x -> a -> m x) -> m x -> (x -> m b) -> Stream (Of a) m r -> m b -foldM_ step begin done s0 = do - x0 <- begin - loop s0 x0 - where - loop stream !x = case stream of - Return r -> done x - Effect m -> m >>= \s -> loop s x - Step (a :> rest) -> do - x' <- step x a - loop rest x' -{-# INLINABLE foldM_ #-} - -{-| Strict, monadic fold of the elements of a 'Stream (Of a)' - -> Control.Foldl.impurely foldM' :: Monad m => FoldM a b -> Stream (Of a) m r -> m (b, r) --} -foldM - :: Monad m - => (x -> a -> m x) -> m x -> (x -> m b) -> Stream (Of a) m r ->m (Of b r) -foldM step begin done str = do - x0 <- begin - loop str x0 - where - loop stream !x = case stream of - Return r -> done x >>= \b -> return (b :> r) - Effect m -> m >>= \s -> loop s x - Step (a :> rest) -> do - x' <- step x a - loop rest x' -{-# INLINABLE foldM #-} - - - -{-| A natural right fold for consuming a stream of elements. - See also the more general 'iterTM' in the 'Streaming' module - and the still more general 'destroy' - -> foldrT (\a p -> Pipes.yield a >> p) :: Monad m => Stream (Of a) m r -> Producer a m r -> foldrT (\a p -> Conduit.yield a >> p) :: Monad m => Stream (Of a) m r -> Conduit a m r - --} - -foldrT :: (Monad m, MonadTrans t, Monad (t m)) - => (a -> t m r -> t m r) -> Stream (Of a) m r -> t m r -foldrT step = loop where - loop stream = case stream of - Return r -> return r - Effect m -> lift m >>= loop - Step (a :> as) -> step a (loop as) -{-# INLINABLE foldrT #-} - -{-| A natural right fold for consuming a stream of elements. - See also the more general 'iterT' in the 'Streaming' module and the - still more general 'destroy' --} -foldrM :: Monad m - => (a -> m r -> m r) -> Stream (Of a) m r -> m r -foldrM step = loop where - loop stream = case stream of - Return r -> return r - Effect m -> m >>= loop - Step (a :> as) -> step a (loop as) -{-# INLINABLE foldrM #-} - --- --------------- --- for --- --------------- - --- | @for@ replaces each element of a stream with an associated stream. Note that the --- associated stream may layer any functor. -for :: (Monad m, Functor f) => Stream (Of a) m r -> (a -> Stream f m x) -> Stream f m r -for str0 act = loop str0 where - loop str = case str of - Return r -> Return r - Effect m -> Effect $ liftM loop m - Step (a :> rest) -> do - act a - loop rest -{-# INLINEABLE for #-} - -{-| Group layers of any functor by comparisons on a preliminary annotation - --} -groupedBy - :: (Monad m, Functor f) => - (a -> a -> Bool) - -> Stream (Compose (Of a) f) m r - -> Stream (Stream (Compose (Of a) f) m) m r -groupedBy equals = loop where - loop stream = Effect $ do - e <- inspect stream - return $ case e of - Left r -> Return r - Right s@(Compose (a :> p')) -> Step $ - fmap loop (Step $ Compose (a :> fmap (span' (equals a)) p')) - span' :: (Monad m, Functor f) => (a -> Bool) -> Stream (Compose (Of a) f) m r - -> Stream (Compose (Of a) f) m (Stream (Compose (Of a) f) m r) - span' pred = loop where - loop str = case str of - Return r -> Return (Return r) - Effect m -> Effect $ liftM loop m - Step s@(Compose (a :> rest)) -> case pred a of - True -> Step (Compose (a :> fmap loop rest)) - False -> Return (Step s) -{-# INLINEABLE groupedBy #-} - -{-| Group elements of a stream by comparisons on a preliminary annotation - --} -groupBy :: Monad m - => (a -> a -> Bool) - -> Stream (Of a) m r - -> Stream (Stream (Of a) m) m r -groupBy equals = loop where - loop stream = Effect $ do - e <- next stream - return $ case e of - Left r -> Return r - Right (a, p') -> Step $ - fmap loop (yield a >> span (equals a) p') - -{-# INLINEABLE groupBy #-} - -group :: (Monad m, Eq a) => Stream (Of a) m r -> Stream (Stream (Of a) m) m r -group = groupBy (==) - - --- --------------- --- iterate --- --------------- - --- | Iterate a pure function from a seed value, streaming the results forever -iterate :: (a -> a) -> a -> Stream (Of a) m r -iterate f = loop where - loop a' = Step (a' :> loop (f a')) -{-# INLINEABLE iterate #-} - --- | Iterate a monadic function from a seed value, streaming the results forever -iterateM :: Monad m => (a -> m a) -> m a -> Stream (Of a) m r -iterateM f = loop where - loop ma = Effect $ do - a <- ma - return (Step (a :> loop (f a))) -{-# INLINEABLE iterateM #-} - - --- --------------- --- length --- --------------- - -{-| Run a stream, remembering only its length: - ->>> S.length $ S.each [1..10] -10 - --} -length_ :: Monad m => Stream (Of a) m r -> m Int -length_ = fold_ (\n _ -> n + 1) 0 id -{-#INLINE length_#-} - -{-| Run a stream, keeping its length and its return value. - ->>> S.print $ mapsM S.length $ chunksOf 3 $ S.each [1..10] -3 -3 -3 -1 - --} - -length :: Monad m => Stream (Of a) m r -> m (Of Int r) -length = fold (\n _ -> n + 1) 0 id -{-#INLINE length #-} --- --------------- --- map --- --------------- - --- | Standard map on the elements of a stream. -map :: Monad m => (a -> b) -> Stream (Of a) m r -> Stream (Of b) m r -map f = loop where - loop stream = case stream of - Return r -> Return r - Effect m -> Effect (liftM loop m) - Step (a :> as) -> Step (f a :> loop as) -{-# INLINE map #-} - --- --------------- --- mapFoldable --- --------------- - -{-| For each element of a stream, stream a foldable container of elements instead; compare - 'Pipes.Prelude.mapFoldable'. - -> mapFoldable f str = for str (\a -> each (f a)) - ->>> S.print $ S.mapFoldable show $ yield 12 -'1' -'2' - --} -mapFoldable :: (Monad m, Foldable.Foldable t) => (a -> t b) -> Stream (Of a) m r -> Stream (Of b) m r -mapFoldable f str = for str (\a -> each (f a)) -- as in pipes - --- | Replace each element of a stream with the result of a monadic action -mapM :: Monad m => (a -> m b) -> Stream (Of a) m r -> Stream (Of b) m r -mapM f = loop where - loop str = case str of - Return r -> Return r - Effect m -> Effect (liftM loop m) - Step (a :> as) -> Effect $ do - a' <- f a - return (Step (a' :> loop as) ) -{-# INLINEABLE mapM #-} - - - -{-| Reduce a stream to its return value with a monadic action. - ->>> mapM_ Prelude.print $ each [1..3] >> return True -1 -2 -3 -True - --} -mapM_ :: Monad m => (a -> m b) -> Stream (Of a) m r -> m r -mapM_ f = loop where - loop str = case str of - Return r -> return r - Effect m -> m >>= loop - Step (a :> as) -> do - f a - loop as -{-# INLINEABLE mapM_ #-} - - -mconcat :: (Monad m, Monoid w) => Stream (Of w) m r -> m (Of w r) -mconcat = fold mappend mempty id -{-#INLINE mconcat #-} - -mconcat_ :: (Monad m, Monoid w) => Stream (Of w) m r -> m w -mconcat_ = fold_ mappend mempty id - -{-| The standard way of inspecting the first item in a stream of elements, if the - stream is still \'running\'. The @Right@ case contains a - Haskell pair, where the more general @inspect@ would return a left-strict pair. - There is no reason to prefer @inspect@ since, if the @Right@ case is exposed, - the first element in the pair will have been evaluated to whnf. - -> next :: Monad m => Stream (Of a) m r -> m (Either r (a, Stream (Of a) m r)) -> inspect :: Monad m => Stream (Of a) m r -> m (Either r (Of a (Stream (Of a) m r))) - - Interoperate with @pipes@ producers thus: - -> Pipes.unfoldr Stream.next :: Stream (Of a) m r -> Producer a m r -> Stream.unfoldr Pipes.next :: Producer a m r -> Stream (Of a) m r - - Similarly: - -> IOStreams.unfoldM (liftM (either (const Nothing) Just) . next) :: Stream (Of a) IO b -> IO (InputStream a) -> Conduit.unfoldM (liftM (either (const Nothing) Just) . next) :: Stream (Of a) m r -> Source a m r - - But see 'uncons', which is better fitted to these @unfoldM@s --} -next :: Monad m => Stream (Of a) m r -> m (Either r (a, Stream (Of a) m r)) -next = loop where - loop stream = case stream of - Return r -> return (Left r) - Effect m -> m >>= loop - Step (a :> rest) -> return (Right (a,rest)) -{-# INLINABLE next #-} - - -{-| Inspect the first item in a stream of elements, without a return value. - @uncons@ provides convenient exit into another streaming type: - -> IOStreams.unfoldM uncons :: Stream (Of a) IO b -> IO (InputStream a) -> 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 ())) -uncons = loop where - loop stream = case stream of - Return () -> return Nothing - Effect m -> m >>= loop - Step (a :> rest) -> return (Just (a,rest)) -{-# INLINABLE uncons #-} - - --- | Fold a 'Stream' of numbers into their product -product_ :: (Monad m, Num a) => Stream (Of a) m () -> m a -product_ = fold_ (*) 1 id -{-# INLINE product_ #-} - -{-| Fold a 'Stream' of numbers into their product with the return value - -> maps' product' :: Stream (Stream (Of Int)) m r -> Stream (Of Int) m r --} -product :: (Monad m, Num a) => Stream (Of a) m r -> m (Of a r) -product = fold (*) 1 id -{-# INLINE product #-} - - --- --------------- --- read --- --------------- - --- | Make a stream of strings into a stream of parsed values, skipping bad cases -read :: (Monad m, Read a) => Stream (Of String) m r -> Stream (Of a) m r -read stream = for stream $ \str -> case readMaybe str of - Nothing -> return () - Just r -> yield r -{-# INLINE read #-} - --- --------------- --- 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. - ->>> S.toListM $ S.take 2 (repeatM getLine) -hello -world -["hello","world"] --} - -repeatM :: Monad m => m a -> Stream (Of a) m r -repeatM ma = loop where - loop = do - a <- lift ma - yield a - loop -{-# INLINEABLE repeatM #-} - --- --------------- --- replicate --- --------------- - --- | Repeat an element several times -replicate :: Monad m => Int -> a -> Stream (Of a) m () -replicate n a = loop n where - loop 0 = Return () - loop m = Step (a :> loop (m-1)) -{-# INLINEABLE replicate #-} - -{-| Repeat an action several times, streaming the results. - ->>> S.print $ S.replicateM 2 getCurrentTime -2015-08-18 00:57:36.124508 UTC -2015-08-18 00:57:36.124785 UTC - --} -replicateM :: Monad m => Int -> m a -> Stream (Of a) m () -replicateM n ma = loop n where - loop 0 = Return () - loop n = Effect $ do - a <- ma - 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 = Effect $ 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 - ->>> S.print $ L.purely S.scan L.list $ each [3..5] -[] -[3] -[3,4] -[3,4,5] - - A simple way of including the scanned item with the accumulator is to use - 'Control.Foldl.last'. See also 'Streaming.Prelude.scanned' - ->>> let a >< b = (,) <$> a <*> b ->>> S.print $ L.purely S.scan (L.last >< L.sum) $ S.each [1..3] -(Nothing,0) -(Just 1,1) -(Just 2,3) -(Just 3,6) - --} -scan :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> Stream (Of b) m r -scan step begin done = loop begin - where - loop !x stream = Step $ - done x :> case stream of - Return r -> Return r - Effect m -> Effect $ liftM (loop x) m - Step (a :> rest) -> do - let !x' = step x a - loop x' rest -{-# INLINE scan #-} - -{-| 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 - x <- lift begin - loop x str - where - loop !x stream = do - b <- lift (done x) - yield b - case stream of - Return r -> Return r - Effect m -> Effect $ liftM (loop x) m - Step (a :> rest) -> do - x' <- lift $ step x a - loop x' rest -{-# INLINABLE scanM #-} - -{- Label each element in a stream with a value accumulated according to a fold. - - ->>> S.print $ S.scanned (*) 1 id $ S.each [100,200,300] -(100,100) -(200,20000) -(300,6000000) - ->>> S.print $ L.purely S.scanned L.product $ S.each [100,200,300] -(100,100) -(200,20000) -(300,6000000) - --} - -data Maybe' a = Just' a | Nothing' - -scanned :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> Stream (Of (a,b)) m r -scanned step begin done = loop Nothing' begin - where - loop !m !x stream = do - case stream of - Return r -> return r - Effect mn -> Effect $ liftM (loop m x) mn - Step (a :> rest) -> do - case m of - Nothing' -> do - let !acc = step x a - yield (a, done acc) - loop (Just' a) acc rest - Just' _ -> do - let !acc = done (step x a) - yield (a, acc) - loop (Just' a) (step x a) rest -{-# INLINABLE scanned #-} - - --- --------------- --- sequence --- --------------- - -{-| Like the 'Data.List.sequence' but streaming. The result type is a - stream of a\'s, /but is not accumulated/; the effects of the elements - of the original stream are interleaved in the resulting stream. Compare: - -> sequence :: Monad m => [m a] -> m [a] -> sequence :: Monad m => Stream (Of (m a)) m r -> Stream (Of a) m r --} -sequence :: Monad m => Stream (Of (m a)) m r -> Stream (Of a) m r -sequence = loop where - loop stream = case stream of - Return r -> Return r - Effect m -> Effect $ liftM loop m - Step (ma :> rest) -> Effect $ do - a <- ma - return (Step (a :> loop rest)) -{-# INLINEABLE sequence #-} - --- --------------- --- show --- --------------- - -show :: (Monad m, Show a) => Stream (Of a) m r -> Stream (Of String) m r -show = map Prelude.show -{-# INLINE show #-} --- --------------- --- sum --- --------------- - --- | Fold a 'Stream' of numbers into their sum -sum_ :: (Monad m, Num a) => Stream (Of a) m () -> m a -sum_ = fold_ (+) 0 id -{-# INLINE sum_ #-} - -{-| Fold a 'Stream' of numbers into their sum with the return value - -> maps' sum' :: Stream (Stream (Of Int)) m r -> Stream (Of Int) m r --} -sum :: (Monad m, Num a) => Stream (Of a) m r -> m (Of a r) -sum = fold (+) 0 id -{-# INLINE sum #-} - --- --------------- --- span --- --------------- - --- | Stream elements until one fails the condition, return the rest. -span :: Monad m => (a -> Bool) -> Stream (Of a) m r - -> Stream (Of a) m (Stream (Of a) m r) -span pred = loop where - loop str = case str of - Return r -> Return (Return r) - Effect m -> Effect $ liftM loop m - Step (a :> rest) -> if pred a - then Step (a :> loop rest) - else Return (Step (a :> rest)) -{-# INLINEABLE span #-} - - -{-| Split a stream of elements wherever a given element arises. - The action is like that of 'Prelude.words'. - ->>> S.stdoutLn $ mapsM S.toList $ split ' ' "hello world " -hello -world ->>> Prelude.mapM_ Prelude.putStrLn (Prelude.words "hello world ") -hello -world - --} - -split :: (Eq a, Monad m) => - a -> Stream (Of a) m r -> Stream (Stream (Of a) m) m r -split t = loop where - loop stream = do - e <- lift $ next stream - case e of - Left r -> Return r - Right (a, p') -> - if a /= t - then Step $ fmap loop (yield a >> break (== t) p') - else loop p' -{-#INLINABLE split #-} - -{-| 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 #-} - - --- --------------- --- take --- --------------- - -{-| End a stream after n elements; the original return value is thus lost. - 'splitAt' preserves this information. Note that, like @splitAt@, this - function is functor-general, so that, for example, you can @take@ not - just a number of items from a stream of elements, but a number - of substreams and the like. - ->>> S.print $ mapsM S.sum $ S.take 2 $ chunksOf 3 $ each [1..] -6 -- sum of first group of 3 -15 -- sum of second group of 3 - --} - -take :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m () -take = loop where - loop 0 p = return () - loop n p = - case p of Step fas -> Step (fmap (loop (n-1)) fas) - Effect m -> Effect (liftM (loop n) m) - Return r -> Return () -{-# INLINE take #-} - --- --------------- --- takeWhile --- --------------- - --- | End stream when an element fails a condition; the original return value is lost --- 'span' preserves this information. -takeWhile :: Monad m => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m () -takeWhile pred = loop where - loop str = case str of - Step (a :> as) -> when (pred a) (Step (a :> loop as)) - Effect m -> Effect (liftM loop m) - Return r -> Return () -{-# INLINE takeWhile #-} - -{- Break a stream after the designated number of seconds. - - ->>> rest <- S.print $ S.timed 1 $ S.delay 0.3 $ S.each [1..] -1 -2 -3 ->>> S.print $ S.take 3 rest -4 -5 -6 - - - - --} - -timed :: MonadIO m => Double -> Stream (Of a) m r -> Stream (Of a) m (Stream (Of a) m r) -timed seconds str = do - utc <- liftIO getCurrentTime - loop utc str - where - cutoff = fromInteger $ truncate (1000000000 * seconds) - loop utc str = do - utc' <- liftIO getCurrentTime - if diffUTCTime utc' utc > (cutoff / 1000000000) - then return str - else case str of - Return r -> return (return r) - Effect m -> Effect (liftM (loop utc) m) - Step (a:>rest) -> yield a >> loop utc rest - - - -{-| Convert an effectful 'Stream (Of a)' into a list of @as@ - - 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. - --} -toList_ :: Monad m => Stream (Of a) m () -> m [a] -toList_ = fold_ (\diff a ls -> diff (a: ls)) id (\diff -> diff []) -{-# INLINE toList_ #-} - - -{-| Convert an effectful 'Stream' into a list alongside the return value - -> mapsM toListM :: Stream (Stream (Of a)) m r -> Stream (Of [a]) m --} -toList :: Monad m => Stream (Of a) m r -> m (Of [a] r) -toList = fold (\diff a ls -> diff (a: ls)) id (\diff -> diff []) -{-# INLINE toList #-} - -{-| Build a @Stream@ by unfolding steps starting from a seed. - - The seed can of course be anything, but this is one natural way - to consume a @pipes@ 'Pipes.Producer'. Consider: - ->>> S.stdoutLn $ S.take 2 (S.unfoldr P.next P.stdinLn) -hello -hello -goodbye -goodbye - ->>> S.stdoutLn $ S.unfoldr P.next (P.stdinLn P.>-> P.take 2) -hello -hello -goodbye -goodbye - ->>> S.effects $ S.unfoldr P.next (P.stdinLn P.>-> P.take 2 P.>-> P.stdoutLn) -hello -hello -goodbye -goodbye - - If the intended \"coalgebra\" is complicated it might be pleasant to - write it with the state monad: - -> \state seed -> S.unfoldr (runExceptT . runStateT state) seed :: Monad m => StateT s (ExceptT r m) a -> s -> P.Producer a m r - ->>> let state = do {n <- get ; if n >= 3 then lift (throwE "Got to three"); else put (n+1); return n} ->>> S.print $ S.unfoldr (runExceptT . runStateT state) 0 -0 -1 -2 -"Got to three" --} -unfoldr :: Monad m - => (s -> m (Either r (a, s))) -> s -> Stream (Of a) m r -unfoldr step = loop where - loop s0 = Effect $ do - e <- step s0 - case e of - Left r -> return (Return r) - Right (a,s) -> return (Step (a :> loop s)) -{-# INLINABLE unfoldr #-} - --- --------------------------------------- --- yield --- --------------------------------------- - -{-| A singleton stream - ->>> stdoutLn $ yield "hello" -hello - ->>> S.sum $ do {yield 1; yield 2} -3 - ->>> S.sum $ do {yield 1; lift $ putStrLn "/* 1 was yielded */"; yield 2; lift $ putStrLn "/* 2 was yielded */"} -/* 1 was yielded */ -/* 2 was yielded */ -3 - - ->>> let prompt :: IO Int; prompt = putStrLn "Enter a number:" >> readLn ->>> S.sum $ do {lift prompt >>= yield ; lift prompt >>= yield ; lift prompt >>= yield} -Enter a number: -3 -Enter a number: -20 -Enter a number: -100 -123 - --} - -yield :: Monad m => a -> Stream (Of a) m () -yield a = Step (a :> Return ()) -{-# INLINE yield #-} - --- | Zip two 'Streams's -zip :: Monad m - => (Stream (Of a) m r) - -> (Stream (Of b) m r) - -> (Stream (Of (a,b)) m r) -zip = zipWith (,) -{-# INLINE zip #-} - --- | Zip two 'Streams's using the provided combining function -zipWith :: Monad m - => (a -> b -> c) - -> (Stream (Of a) m r) - -> (Stream (Of b) m r) - -> (Stream (Of c) m r) -zipWith f = loop - where - loop str0 str1 = case str0 of - Return r -> Return r - Effect m -> Effect $ liftM (\str -> loop str str1) m - Step (a :> rest0) -> case str1 of - Return r -> Return r - Effect m -> Effect $ liftM (loop str0) m - Step (b :> rest1) -> Step (f a b :>loop rest0 rest1) -{-# INLINABLE zipWith #-} - - --- | Zip three 'Stream's with a combining function -zipWith3 :: Monad m => - (a -> b -> c -> d) - -> Stream (Of a) m r - -> Stream (Of b) m r - -> Stream (Of c) m r - -> Stream (Of d) m r -zipWith3 op = loop where - loop str0 str1 str2 = do - e0 <- lift (next str0) - case e0 of - Left r0 -> return r0 - Right (a0,rest0) -> do - e1 <- lift (next str1) - case e1 of - Left r1 -> return r1 - Right (a1,rest1) -> do - e2 <- lift (next str2) - case e2 of - Left r2 -> return r2 - Right (a2,rest2) -> do - yield (op a0 a1 a2) - loop rest0 rest1 rest2 -{-# INLINABLE zipWith3 #-} - - --- | Zip three streams together -zip3 :: Monad m - => (Stream (Of a) m r) - -> (Stream (Of b) m r) - -> (Stream (Of c) m r) - -> (Stream (Of (a,b,c)) m r) -zip3 = zipWith3 (,,) -{-# INLINABLE zip3 #-} - --- -------------- --- IO fripperies --- -------------- - -{-| repeatedly stream lines as 'String' from stdin - ->>> stdoutLn $ S.show (S.each [1..3]) -1 -2 -3 - ->>> stdoutLn stdinLn -hello -hello -world -world -^CInterrupted. - - ->>> stdoutLn $ S.map reverse stdinLn -hello -olleh -world -dlrow -^CInterrupted. - --} -stdinLn :: MonadIO m => Stream (Of String) m () -stdinLn = fromHandle IO.stdin -{-# INLINABLE stdinLn #-} - -{-| Read values from 'IO.stdin', ignoring failed parses - ->>> S.sum $ S.take 2 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 () - Just n -> yield n -{-# INLINABLE readLn #-} - -{-| Read 'String's from a 'IO.Handle' using 'IO.hGetLine' - - Terminates on end of input - ->>> withFile "distribute.hs" ReadMode $ stdoutLn . S.take 3 . fromHandle -import Streaming -import qualified Streaming.Prelude as S -import Control.Monad.Trans.State.Strict - --} -fromHandle :: MonadIO m => IO.Handle -> Stream (Of String) m () -fromHandle h = go - where - go = do - eof <- liftIO $ IO.hIsEOF h - unless eof $ do - str <- liftIO $ IO.hGetLine h - yield str - go -{-# INLINABLE fromHandle #-} - -toHandle :: MonadIO m => IO.Handle -> Stream (Of String) m r -> m r -toHandle handle = loop where - loop str = case str of - Return r -> return r - Effect m -> m >>= loop - Step (s :> rest) -> do - liftIO $ IO.hPutStrLn handle s - loop rest -{-# INLINABLE toHandle #-} - -{-| Print the elements of a stream as they arise. --} -print :: (MonadIO m, Show a) => Stream (Of a) m r -> m r -print = loop where - loop stream = case stream of - Return r -> return r - Effect m -> m >>= loop - Step (a :> rest) -> do - 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 #-} - -{-| Write 'String's to 'IO.stdout' using 'putStrLn'; terminates on a broken output pipe - (compare 'Pipes.Prelude.stdoutLn'). - ->>> S.stdoutLn $ S.show (S.each [1..3]) -1 -2 -3 --} -stdoutLn :: MonadIO m => Stream (Of String) m () -> m () -stdoutLn = loop - where - loop stream = case stream of - Return _ -> return () - Effect m -> m >>= loop - Step (s :> rest) -> do - x <- liftIO $ try (putStrLn s) - case x of - Left (G.IOError { G.ioe_type = G.ResourceVanished - , G.ioe_errno = Just ioe }) - | Errno ioe == ePIPE - -> return () - Left e -> liftIO (throwIO e) - Right () -> loop rest -{-# INLINABLE stdoutLn #-} - - -{-| Write 'String's to 'IO.stdout' using 'putStrLn' - - This does not handle a broken output pipe, but has a polymorphic return - value, which makes this possible: - ->>> rest <- stdoutLn' $ S.show $ S.splitAt 3 (each [1..5]) -1 -2 -3 ->>> S.sum rest -9 - --} - -stdoutLn' :: MonadIO m => Stream (Of String) m r -> m r -stdoutLn' = loop where - loop stream = case stream of - Return r -> return r - Effect m -> m >>= loop - Step (s :> rest) -> liftIO (putStrLn s) >> loop rest -{-# INLINE stdoutLn' #-} - - --- -- * Producers --- -- $producers --- stdinLn -- --- , readLn -- --- , fromHandle -- --- , repeatM -- --- , replicateM -- --- --- -- * Consumers --- -- $consumers --- , stdoutLn -- --- , stdoutLn' -- --- , mapM_ -- --- , print -- --- , toHandle -- --- , effects -- --- --- -- * Pipes --- -- $pipes --- , map -- --- , mapM -- --- , sequence -- --- , mapFoldable -- --- , filter -- --- , filterM -- --- , take -- --- , takeWhile -- --- , takeWhile' -- --- , drop -- --- , dropWhile -- --- , concat -- --- , elemIndices --- , findIndices --- , scan -- --- , scanM -- --- , chain -- --- , read -- --- , show -- --- , seq -- --- --- -- * Folds --- -- $folds --- , fold -- --- , fold' -- --- , foldM -- --- , foldM' -- --- , all --- , any --- , and --- , or --- , elem --- , notElem --- , find --- , findIndex --- , head --- , index --- , last --- , length --- , maximum --- , minimum --- , null --- , sum -- --- , product -- --- , toList -- --- , toListM -- --- , toListM' -- --- --- -- * Zips --- , zip -- --- , zipWith -- --- - -distinguish :: (a -> Bool) -> Of a r -> Sum (Of a) (Of a) r -distinguish predicate (a :> b) = if predicate a then InR (a :> b) else InL (a :> b) -{-#INLINE distinguish #-} - - -eitherToSum :: Of (Either a b) r -> Sum (Of a) (Of b) r -eitherToSum s = case s of - Left a :> r -> InL (a :> r) - Right b :> r -> InR (b :> r) -{-#INLINE eitherToSum #-} -composeToSum :: Compose (Of Bool) f r -> Sum f f r -composeToSum x = case x of - Compose (True :> f) -> InR f - Compose (False :> f) -> InL f -{-#INLINE composeToSum #-} - -sumToCompose :: Sum f f r -> Compose (Of Bool) f r -sumToCompose x = case x of - InR f -> Compose (True :> f) - InL f -> Compose (False :> f) -{-#INLINE sumToCompose #-} - -duplicate - :: Monad m => - Stream (Of a) m r -> Stream (Of a) (Stream (Of a) m) r -duplicate = loop where -loop str = do - e <- lift (lift (next str)) - case e of - Left r -> return r - Right (a, rest) -> do - yield a - mwrap $ do - yield a - return (loop rest) -{-#INLINABLE duplicate#-} - +{-| This module is very closely modeled on Pipes.Prelude, Pipes.Group and Pipes.Parse. It + maybe said to give independent expression to the conception of Producer manipulation + articulated in the latter two modules. Because we dispense with piping and + conduiting, the distinction between all of these modules collapses. + The leading type is chosen to permit an api that is as close as possible to that + of Data.List and the Prelude. Thecan + be used with any + rational \"streaming IO\" system. + + Import qualified thus: + +> import Streaming +> import qualified Streaming.Prelude as S + + For the examples below, one sometimes needs + +> import Streaming.Prelude (each, yield, stdoutLn, stdinLn) +> import Data.Function ((&)) + + Other libraries that come up in passing are + +> import qualified Control.Foldl as L -- cabal install foldl +> import qualified Pipes as P +> import qualified Pipes.Prelude as P +> import qualified System.IO as IO + + Here are some correspondences between the types employed here and elsewhere: + +> streaming | pipes | conduit | io-streams +> ------------------------------------------------------------------------------------------------------------------- +> Stream (Of a) m () | Producer a m () | Source m a | InputStream a +> | ListT m a | ConduitM () o m () | Generator r () +> ------------------------------------------------------------------------------------------------------------------- +> Stream (Of a) m r | Producer a m r | ConduitM () o m r | Generator a r +> ------------------------------------------------------------------------------------------------------------------- +> Stream (Of a) m (Stream (Of a) m r) | Producer a m (Producer a m r) | +> -------------------------------------------------------------------------------------------------------------------- +> Stream (Stream (Of a) m) r | FreeT (Producer a m) m r | +> -------------------------------------------------------------------------------------------------------------------- +> -------------------------------------------------------------------------------------------------------------------- +> ByteString m () | Producer ByteString m () | Source m ByteString | InputStream ByteString +> -------------------------------------------------------------------------------------------------------------------- +> +-} +{-# LANGUAGE RankNTypes, BangPatterns, DeriveDataTypeable, TypeFamilies, + DeriveFoldable, DeriveFunctor, DeriveTraversable #-} + +module Streaming.Prelude ( + -- * Types + Of (..) + + -- * Introducing streams of elements + -- $producers + , yield + , each + , unfoldr + , stdinLn + , readLn + , fromHandle + , readFile + , iterate + , repeat + , replicate + , cycle + , repeatM + , replicateM + , enumFrom + , enumFromThen + , seconds + + -- * Consuming streams of elements + -- $consumers + , stdoutLn + , stdoutLn' + , mapM_ + , print + , toHandle + , writeFile + , effects + , drained + + + -- * Stream transformers + -- $pipes + , map + , mapM + , chain + , maps + , sequence + , nub + , filter + , filterM + , for + , with + , delay + , intersperse + , take + , takeWhile +-- , takeWhile' + , drop + , dropWhile + , concat + -- , elemIndices + -- , findIndices + , scan + , scanM + , scanned + , read + , show + , cons + , duplicate + , store + + -- * Splitting and inspecting streams of elements + , next + , uncons + , splitAt + , split +-- , breaks + , break + , breakWhen + , span + , group + , groupBy + -- , groupedBy + -- , split + + + -- * Sum and Compose manipulation + + , distinguish + , switch + , separate + , unseparate + , eitherToSum + , sumToEither + , sumToCompose + , composeToSum + + -- * Folds + -- $folds + , fold + , fold_ + , foldM + , foldM_ + , all + , all_ + , any + , any_ + , sum + , sum_ + , product + , product_ + , head + , head_ + , last + , last_ + , elem + , elem_ + , length + , length_ + , toList + , toList_ + , mconcat + , mconcat_ + , minimum + , minimum_ + , maximum + , maximum_ + , foldrM + , foldrT + + + -- , all + -- , any + -- , and + -- , or + -- , elem + -- , notElem + -- , find + -- , findIndex + -- , head + -- , index + -- , last + -- , length + -- , maximum + -- , minimum + -- , null + + -- * Zips and unzips + , zip + , zipWith + , zip3 + , zipWith3 + , unzip + + -- * Pair manipulation + , lazily + , strictly + , fst' + , snd' + + -- * Interoperation + , reread + + -- * Basic Type + , Stream + + ) where +import Streaming.Internal + +import Control.Monad hiding (filterM, mapM, mapM_, foldM, foldM_, replicateM, sequence) +import Data.Data ( Data, Typeable ) +import Data.Functor.Identity +import Data.Functor.Sum +import Control.Monad.Trans +import Control.Applicative (Applicative (..)) +import Data.Functor (Functor (..), (<$)) + +import qualified Prelude as Prelude +import Data.Foldable (Foldable) +import Data.Traversable (Traversable) +import qualified Data.Foldable as Foldable +import Text.Read (readMaybe) +import Prelude hiding (map, mapM, mapM_, filter, drop, dropWhile, take, mconcat, sum, product + , iterate, repeat, cycle, replicate, splitAt + , takeWhile, enumFrom, enumFromTo, enumFromThen, length + , print, zipWith, zip, zipWith3, zip3, unzip, seq, show, read + , readLn, sequence, concat, span, break, readFile, writeFile + , minimum, maximum, elem, intersperse, all, any, head, last) + +import qualified GHC.IO.Exception as G +import qualified System.IO as IO +import Foreign.C.Error (Errno(Errno), ePIPE) +import Control.Exception (throwIO, try) +import Data.Monoid (Monoid (mappend, mempty)) +import Data.String (IsString (..)) +import Control.Concurrent (threadDelay) +import Data.Time (getCurrentTime, diffUTCTime, picosecondsToDiffTime) +import Data.Functor.Classes +import Data.Functor.Compose +import Control.Monad.Trans.Resource +import qualified Data.Set as Set + +import GHC.Exts ( SpecConstrAnnotation(..) ) + + +data SPEC = SPEC | SPEC2 + +{-# ANN type SPEC ForceSpecConstr #-} +-- | A left-strict pair; the base functor for streams of individual elements. +data Of a b = !a :> b + deriving (Data, Eq, Foldable, Ord, + Read, Show, Traversable, Typeable) +infixr 5 :> + +instance (Monoid a, Monoid b) => Monoid (Of a b) where + mempty = mempty :> mempty + {-#INLINE mempty #-} + mappend (m :> w) (m' :> w') = mappend m m' :> mappend w w' + {-#INLINE mappend #-} + +instance Functor (Of a) where + fmap f (a :> x) = a :> f x + {-#INLINE fmap #-} + a <$ (b :> x) = b :> a + {-#INLINE (<$) #-} + +instance Monoid a => Applicative (Of a) where + pure x = mempty :> x + {-#INLINE pure #-} + m :> f <*> m' :> x = mappend m m' :> f x + {-#INLINE (<*>) #-} + m :> x *> m' :> y = mappend m m' :> y + {-#INLINE (*>) #-} + m :> x <* m' :> y = mappend m m' :> x + {-#INLINE (<*) #-} + +instance Monoid a => Monad (Of a) where + return x = mempty :> x + {-#INLINE return #-} + m :> x >> m' :> y = mappend m m' :> y + {-#INLINE (>>) #-} + m :> x >>= f = let m' :> y = f x in mappend m m' :> y + {-#INLINE (>>=) #-} + +instance (r ~ (), Monad m, f ~ Of Char) => IsString (Stream f m r) where + fromString = each + +instance (Eq a) => Eq1 (Of a) where eq1 = (==) +instance (Ord a) => Ord1 (Of a) where compare1 = compare +instance (Read a) => Read1 (Of a) where readsPrec1 = readsPrec +instance (Show a) => Show1 (Of a) where showsPrec1 = showsPrec + +{-| Note that 'lazily', 'strictly', 'fst'', and 'mapOf' are all so-called /natural transformations/ on the primitive @Of a@ functor + If we write + +> type f ~~> g = forall x . f x -> g x + + then we can restate some types as follows: + +> mapOf :: (a -> b) -> Of a ~~> Of b -- bifunctor lmap +> lazily :: Of a ~~> (,) a +> Identity . fst' :: Of a ~~> Identity a + + Manipulation of a @Stream f m r@ by mapping often turns on recognizing natural transformations of @f@, + thus @maps@ is far more general the the @map@ of the present module, which can be + defined thus: + +> S.map :: (a -> b) -> Stream (Of a) m r -> Stream (Of b) m r +> S.map f = maps (mapOf f) + + This rests on recognizing that @mapOf@ is a natural transformation; note though + that it results in such a transformation as well: + +> S.map :: (a -> b) -> Stream (Of a) m ~> Stream (Of b) m + +-} +lazily :: Of a b -> (a,b) +lazily = \(a:>b) -> (a,b) +{-# INLINE lazily #-} + +strictly :: (a,b) -> Of a b +strictly = \(a,b) -> a :> b +{-# INLINE strictly #-} + +fst' :: Of a b -> a +fst' (a :> b) = a + +snd' :: Of a b -> b +snd' (a :> b) = b + +mapOf :: (a -> b) -> Of a r -> Of b r +mapOf f (a:> b) = (f a :> b) + + +all :: Monad m => (a -> Bool) -> Stream (Of a) m r -> m (Of Bool r) +all thus = loop True where + loop b str = case str of + Return r -> return (b :> r) + Effect m -> m >>= loop b + Step (a :> rest) -> if thus a + then loop True rest + else do + r <- effects rest + return (False :> r) +{-#INLINABLE all #-} + +all_ :: Monad m => (a -> Bool) -> Stream (Of a) m r -> m Bool +all_ thus = loop True where + loop b str = case str of + Return r -> return b + Effect m -> m >>= loop b + Step (a :> rest) -> if thus a + then loop True rest + else return False +{-#INLINABLE all_ #-} + + +any :: Monad m => (a -> Bool) -> Stream (Of a) m r -> m (Of Bool r) +any thus = loop False where + loop b str = case str of + Return r -> return (b :> r) + Effect m -> m >>= loop b + Step (a :> rest) -> if thus a + then do + r <- effects rest + return (True :> r) + else loop False rest +{-#INLINABLE any #-} + +any_ :: Monad m => (a -> Bool) -> Stream (Of a) m r -> m Bool +any_ thus = loop False where + loop b str = case str of + Return r -> return b + Effect m -> m >>= loop b + Step (a :> rest) -> if thus a + then return True + else loop False rest +{-#INLINABLE any_ #-} + +{-| Break a sequence upon meeting element falls under a predicate, + keeping it and 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 + loop str = case str of + Return r -> Return (Return r) + Effect m -> Effect $ liftM loop m + Step (a :> rest) -> if (pred a) + then Return (Step (a :> rest)) + else Step (a :> loop rest) +{-# INLINABLE break #-} + +{-| Yield elements, using a fold to maintain state, until the accumulated + value satifies the supplied predicate. The fold will then be short-circuited + and the element that breaks it will be put after the break. + This function is easiest to use with 'Control.Foldl.purely' + +>>> rest <- each [1..10] & L.purely S.breakWhen L.sum (>10) & S.print +1 +2 +3 +4 +>>> S.print rest +5 +6 +7 +8 +9 +10 + +-} +breakWhen :: Monad m => (x -> a -> x) -> x -> (x -> b) -> (b -> Bool) -> Stream (Of a) m r -> Stream (Of a) m (Stream (Of a) m r) +breakWhen step begin done pred = loop0 begin + where + loop0 x stream = case stream of + Return r -> return (return r) + Effect mn -> Effect $ liftM (loop0 x) mn + Step (a :> rest) -> loop a (step x a) rest + loop a !x stream = do + if pred (done x) + then return (yield a >> stream) + else case stream of + Return r -> yield a >> return (return r) + Effect mn -> Effect $ liftM (loop a x) mn + Step (a' :> rest) -> do + yield a + loop a' (step x a') rest +{-# INLINABLE breakWhen #-} + +-- -- Break during periods where the predicate is not satisfied, grouping the periods when it is. +-- +-- >>> S.print $ mapped S.toList $ S.breaks not $ S.each [False,True,True,False,True,True,False] +-- [True,True] +-- [True,True] +-- >>> S.print $ mapped S.toList $ S.breaks id $ S.each [False,True,True,False,True,True,False] +-- [False] +-- [False] +-- [False] +-- +-- -} +-- breaks +-- :: Monad m => +-- (a -> Bool) -> Stream (Of a) m r -> Stream (Stream (Of a) m) m r +-- breaks thus = loop where +-- loop stream = Effect $ do +-- e <- next stream +-- return $ case e of +-- Left r -> Return r +-- Right (a, p') -> +-- if not (thus a) +-- then Step $ fmap loop (yield a >> break thus p') +-- else loop p' +-- {-#INLINABLE breaks #-} + +{-| Apply an action to all values, re-yielding each + +>>> S.product $ S.chain Prelude.print $ S.each [1..5] +1 +2 +3 +4 +5 +120 :> () +-} + +chain :: Monad m => (a -> m ()) -> Stream (Of a) m r -> Stream (Of a) m r +chain f = loop where + loop str = case str of + Return r -> return r + Effect mn -> Effect (liftM loop mn) + Step (a :> rest) -> Effect $ do + f a + return (Step (a :> loop rest)) +{-# INLINABLE chain #-} + +{-| Make a stream of traversable containers into a stream of their separate elements. + This is just + +> concat = for str each + +>>> S.print $ S.concat (each ["xy","z"]) +'x' +'y' +'z' + + Note that it also has the effect of 'Data.Maybe.catMaybes', 'Data.Either.rights' + 'map snd' and such-like operations. + +>>> S.print $ S.concat $ S.each [Just 1, Nothing, Just 2] +1 +2 +>>> S.print $ S.concat $ S.each [Right 1, Left "Error!", Right 2] +1 +2 +>>> S.print $ S.concat $ S.each [('A',1), ('B',2)] +1 +2 + +-} + +concat :: (Monad m, Foldable.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: + +> Data.Text.foldr S.cons (return ()) :: Text -> Stream (Of Char) m () +> Lazy.foldrChunks S.cons (return ()) :: Lazy.ByteString -> Stream (Of Strict.ByteString) m () + + and so on. +-} + +cons :: (Monad m) => a -> Stream (Of a) m r -> Stream (Of a) m r +cons a str = Step (a :> str) +{-# INLINE cons #-} + +{- | Cycle repeatedly through the layers of a stream, /ad inf./ This + function is functor-general + +> cycle = forever + +>>> rest <- S.print $ S.splitAt 3 $ S.cycle (yield 0 >> yield 1) +True +False +True +>>> S.print $ S.take 3 rest +False +True +False + +-} + +cycle :: (Monad m, Functor f) => Stream f m r -> Stream f m s +cycle str = loop where loop = str >> loop +{-#INLINABLE cycle #-} + + +{-| A + + +-} +delay :: MonadIO m => Double -> Stream (Of a) m r -> Stream (Of a) m r +delay seconds = loop where + pico = truncate (seconds * 1000000) + loop str = do + e <- lift $ next str + case e of + Left r -> Return r + Right (a,rest) -> do + yield a + liftIO $ threadDelay pico + loop rest +{-#INLINABLE delay #-} + +-- --------------- +-- effects +-- --------------- + +{- | Reduce a stream, performing its actions but ignoring its elements. + +>>> rest <- S.effects $ S.splitAt 2 $ each [1..5] +>>> S.print rest +3 +4 +5 + + +-} +effects :: Monad m => Stream (Of a) m r -> m r +effects = loop where + loop stream = case stream of + Return r -> return r + Effect m -> m >>= loop + Step (_ :> rest) -> loop rest +{-#INLINABLE effects #-} + +{-| Where a transformer returns a stream, run the effects of the stream, keeping + the return value. This is usually used at the type + +> drained :: Monad m => Stream (Of a) m (Stream (Of b) m r) -> Stream (Of a) m r +> drained = join . fmap (lift . effects) + + Here, for example, we split a stream in two places and throw out the middle segment: + +>>> rest <- S.print $ S.drained $ S.splitAt 2 $ S.splitAt 5 $ each [1..7] +1 +2 +>>> S.print rest +6 +7 + + In particular, we can define versions of @take@ and @takeWhile@ which + retrieve the return value of the rest of the stream - and which can + thus be used with 'maps': + +> take' n = S.drained . S.splitAt n +> takeWhile' thus = S.drained . S.span thus + +-} +drained :: (Monad m, Monad (t m), Functor (t m), MonadTrans t) => t m (Stream (Of a) m r) -> t m r +drained = join . fmap (lift . effects) +{-#INLINE drained #-} + +-- --------------- +-- drop +-- --------------- +{-| Ignore the first n elements of a stream, but carry out the actions + +>>> S.toList $ S.drop 2 $ S.replicateM 5 getLine +a +b +c +d +e +["c","d","e"] :> () + + Because it retains the final return value, @drop n@ is a suitable argument + for @maps@: + +>>> S.toList $ concats $ maps (S.drop 4) $ chunksOf 5 $ each [1..20] +[5,10,15,20] :> () + + + + -} + +drop :: (Monad m) => Int -> Stream (Of a) m r -> Stream (Of a) m r +drop = loop where + loop 0 stream = stream + loop n stream = case stream of + Return r -> Return r + Effect ma -> Effect (liftM (loop n) ma) + Step (a :> as) -> loop (n-1) as +{-# INLINABLE drop #-} + +-- --------------- +-- dropWhile +-- --------------- + +{- | Ignore elements of a stream until a test succeeds, retaining the rest. + +>>> S.print $ S.dropWhile ((< 5) . length) S.stdinLn +one +two +three +"three" +four +"four" +^CInterrupted. + + +-} +dropWhile :: Monad m => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m r +dropWhile pred = loop where + loop stream = case stream of + Return r -> Return r + Effect ma -> Effect (liftM loop ma) + Step (a :> as) -> if pred a + then loop as + else Step (a :> as) +{-# INLINABLE dropWhile #-} + +-- --------------- +-- each +-- --------------- + +{- | Stream the elements of a pure, foldable container. + +>>> each [1..3] & S.print +1 +2 +3 +>>> S.replicateM 5 getLine & chunksOf 3 & mapped S.toList & S.print +s +t +u +["s","t","u"] +v +w +["v","w"] + +-} +each :: (Monad m, Foldable.Foldable f) => f a -> Stream (Of a) m () +each = Foldable.foldr (\a p -> (Step (a :> p))) (Return ()) +{-# INLINABLE each #-} + +{-| Exhaust a stream remembering only whether @a@ was an element. + +-} + +elem :: (Monad m, Eq a) => a -> Stream (Of a) m r -> m (Of Bool r) +elem a = fold op False id where + op True _ = True + op False a' | a == a' = True + op _ _ = False +{-#INLINABLE elem #-} + +elem_ :: (Monad m, Eq a) => a -> Stream (Of a) m r -> m Bool +elem_ a = fold_ op False id where + op True _ = True + op False a' | a == a' = True + op _ _ = False +{-#INLINABLE elem_ #-} + +-- ----- +-- enumFrom +-- ------ + +{-| An infinite stream of enumerable values, starting from a given value. + It is the same as `S.iterate succ`. + Because their return type is polymorphic, @enumFrom@ and @enumFromThen@ + (and @iterate@ are useful for example with @zip@ + and @zipWith@, which require the same return type in the zipped streams. + With @each [1..]@ the following bit of connect-and-resume would be impossible: + +>>> rest <- S.print $ S.zip (S.enumFrom 'a') $ S.splitAt 3 $ S.enumFrom 1 +('a',1) +('b',2) +('c',3) +>>> S.print $ S.take 3 rest +4 +5 +6 + +-} +enumFrom :: (Monad m, Enum n) => n -> Stream (Of n) m r +enumFrom = loop where + loop !n = Step (n :> loop (succ n)) +{-# INLINABLE enumFrom #-} + + +{-| An infinite sequence of enumerable values at a fixed distance, determined + by the first and second values. See the discussion of 'Streaming.enumFrom' + +>>> S.print $ S.take 3 $ S.enumFromThen 100 200 +100 +200 +300 + +-} +enumFromThen:: (Monad m, Enum a) => a -> a -> Stream (Of a) m r +enumFromThen first second = Streaming.Prelude.map toEnum (loop _first) + where + _first = fromEnum first + _second = fromEnum second + diff = _second - _first + loop !s = Step (s :> loop (s+diff)) +{-# INLINABLE enumFromThen #-} + +-- --------------- +-- filter +-- --------------- + +-- | Skip elements of a stream that fail a predicate +filter :: (Monad m) => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m r +filter pred = loop where + loop str = case str of + Return r -> Return r + Effect m -> Effect (liftM loop m) + Step (a :> as) -> if pred a + then Step (a :> loop as) + else loop as +{-# INLINABLE filter #-} + +-- --------------- +-- filterM +-- --------------- + +-- | Skip elements of a stream that fail a monadic test +filterM :: (Monad m) => (a -> m Bool) -> Stream (Of a) m r -> Stream (Of a) m r +filterM pred = loop where + loop str = case str of + Return r -> Return r + Effect m -> Effect $ liftM loop m + Step (a :> as) -> Effect $ do + bool <- pred a + if bool + then return $ Step (a :> loop as) + else return $ loop as +{-# INLINABLE filterM #-} + +-- --------------- +-- fold +-- --------------- + +{- $folds + Use these to fold the elements of a 'Stream'. + +>>> S.fold_ (+) 0 id $ S.each [1..0] +50 + + The general folds 'fold', fold_', 'foldM' and 'foldM_' are arranged + for use with 'Control.Foldl' + +>>> L.purely fold_ L.sum $ each [1..10] +55 +>>> L.purely fold_ (liftA3 (,,) L.sum L.product L.list) $ each [1..10] +(55,3628800,[1,2,3,4,5,6,7,8,9,10]) + + All functions marked with an underscore omit + (e.g. @fold_@, @sum_@) the stream's return value in a left-strict pair. + They are good for exiting streaming completely, + but when you are, e.g. @mapped@-ing over a @Stream (Stream (Of a) m) m r@, + which is to be compared with @[[a]]@. Specializing, we have e.g. + +> mapped sum :: (Monad m, Num n) => Stream (Stream (Of Int)) IO () -> Stream (Of n) IO () +> mapped (fold mappend mempty id) :: Stream (Stream (Of Int)) IO () -> Stream (Of Int) IO () + +>>> S.print $ mapped S.sum $ chunksOf 3 $ S.each [1..10] +6 +15 +24 +10 + +>>> let three_folds = L.purely S.fold (liftA3 (,,) L.sum L.product L.list) +>>> S.print $ mapped three_folds $ chunksOf 3 (each [1..10]) +(6,6,[1,2,3]) +(15,120,[4,5,6]) +(24,504,[7,8,9]) +(10,10,[10]) +-} + +{-| Strict fold of a 'Stream' of elements, preserving only the result of the fold, not + the return value of the stream. The third parameter will often be 'id' where a fold + is written by hand: + +>>> S.fold_ (+) 0 id $ each [1..10] +55 + + It can be used to replace a standard Haskell type with one more suited to + writing a strict accumulation function. It is also crucial to the + Applicative instance for @Control.Foldl.Fold@ + +> Control.Foldl.purely fold :: Monad m => Fold a b -> Stream (Of a) m () -> m b +-} +fold_ :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> m b +fold_ step begin done = liftM (\(a:>rest) -> a) . fold step begin done +{-#INLINE fold_ #-} + +{-| Strict fold of a 'Stream' of elements that preserves the return value. + The third parameter will often be 'id' where a fold is written by hand: + +>>> S.fold (+) 0 id $ each [1..10] +55 :> () + +>>> S.fold (*) 1 id $ S.fold (+) 0 id $ S.duplicate $ each [1..10] +3628800 :> (55 :> ()) + + + It can be used to replace a standard Haskell type with one more suited to + writing a strict accumulation function. It is also crucial to the + Applicative instance for @Control.Foldl.Fold@ We can apply such a fold + @purely@ + +> Control.Foldl.purely S.fold :: Monad m => Fold a b -> Stream (Of a) m r -> m (Of b r) + + Thus, specializing a bit: + +> L.purely S.fold L.sum :: Stream (Of Int) Int r -> m (Of Int r) +> maps (L.purely S.fold L.sum) :: Stream (Stream (Of Int)) IO r -> Stream (Of Int) IO r + + Here we use the Applicative instance for @Control.Foldl.Fold@ to + stream three-item segments of a stream together with their sums and products. + +>>> S.print $ mapped (L.purely S.fold (liftA3 (,,) L.list L.product L.sum)) $ chunksOf 3 $ each [1..10] +([1,2,3],6,6) +([4,5,6],120,15) +([7,8,9],504,24) +([10],10,10) + +-} + +fold :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> m (Of b r) +fold step begin done str = fold_loop str begin + where + fold_loop stream !x = case stream of + Return r -> return (done x :> r) + Effect m -> m >>= \str' -> fold_loop str' x + Step (a :> rest) -> fold_loop rest $! step x a +{-# INLINABLE fold #-} + + +{-| Strict, monadic fold of the elements of a 'Stream (Of a)' + +> Control.Foldl.impurely foldM :: Monad m => FoldM a b -> Stream (Of a) m () -> m b +-} +foldM_ + :: Monad m + => (x -> a -> m x) -> m x -> (x -> m b) -> Stream (Of a) m r -> m b +foldM_ step begin done = liftM (\(a:>rest) -> a) . foldM step begin done +{-#INLINE foldM_ #-} + +{-| Strict, monadic fold of the elements of a 'Stream (Of a)' + +> Control.Foldl.impurely foldM' :: Monad m => FoldM a b -> Stream (Of a) m r -> m (b, r) + + Thus to accumulate the elements of a stream as a vector, together with a random + element we might write: + +>>> L.impurely S.foldM (liftA2 (,) L.vector L.random) $ each [1..10::Int] :: IO (Of (U.Vector Int,Maybe Int) ()) +([1,2,3,4,5,6,7,8,9,10],Just 9) :> () + +-} +foldM + :: Monad m + => (x -> a -> m x) -> m x -> (x -> m b) -> Stream (Of a) m r ->m (Of b r) +foldM step begin done str = do + x0 <- begin + loop str x0 + where + loop stream !x = case stream of + Return r -> done x >>= \b -> return (b :> r) + Effect m -> m >>= \s -> loop s x + Step (a :> rest) -> do + x' <- step x a + loop rest x' +{-# INLINABLE foldM #-} + + + +{-| A natural right fold for consuming a stream of elements. + See also the more general 'iterTM' in the 'Streaming' module + and the still more general 'destroy' + +> foldrT (\a p -> Pipes.yield a >> p) :: Monad m => Stream (Of a) m r -> Producer a m r +> foldrT (\a p -> Conduit.yield a >> p) :: Monad m => Stream (Of a) m r -> Conduit a m r + +-} + +foldrT :: (Monad m, MonadTrans t, Monad (t m)) + => (a -> t m r -> t m r) -> Stream (Of a) m r -> t m r +foldrT step = loop where + loop stream = case stream of + Return r -> return r + Effect m -> lift m >>= loop + Step (a :> as) -> step a (loop as) +{-# INLINABLE foldrT #-} + +{-| A natural right fold for consuming a stream of elements. + See also the more general 'iterT' in the 'Streaming' module and the + still more general 'destroy' +-} +foldrM :: Monad m + => (a -> m r -> m r) -> Stream (Of a) m r -> m r +foldrM step = loop where + loop stream = case stream of + Return r -> return r + Effect m -> m >>= loop + Step (a :> as) -> step a (loop as) +{-# INLINABLE foldrM #-} + +-- --------------- +-- for +-- --------------- + +-- | @for@ replaces each element of a stream with an associated stream. Note that the +-- associated stream may layer any functor. +for :: (Monad m, Functor f) => Stream (Of a) m r -> (a -> Stream f m x) -> Stream f m r +for str0 act = loop str0 where + loop str = case str of + Return r -> Return r + Effect m -> Effect $ liftM loop m + Step (a :> rest) -> do + act a + loop rest +{-# INLINABLE for #-} + +-- -| Group layers of any functor by comparisons on a preliminary annotation + +-- groupedBy +-- :: (Monad m, Functor f) => +-- (a -> a -> Bool) +-- -> Stream (Compose (Of a) f) m r +-- -> Stream (Stream (Compose (Of a) f) m) m r +-- groupedBy equals = loop where +-- loop stream = Effect $ do +-- e <- inspect stream +-- return $ case e of +-- Left r -> Return r +-- Right s@(Compose (a :> p')) -> Step $ +-- fmap loop (Step $ Compose (a :> fmap (span' (equals a)) p')) +-- span' :: (Monad m, Functor f) => (a -> Bool) -> Stream (Compose (Of a) f) m r +-- -> Stream (Compose (Of a) f) m (Stream (Compose (Of a) f) m r) +-- span' pred = loop where +-- loop str = case str of +-- Return r -> Return (Return r) +-- Effect m -> Effect $ liftM loop m +-- Step s@(Compose (a :> rest)) -> case pred a of +-- True -> Step (Compose (a :> fmap loop rest)) +-- False -> Return (Step s) +-- {-# INLINABLE groupedBy #-} + +{-| Group elements of a stream in accordance with the supplied comparison. + + +>>> S.print $ mapped S.toList $ S.groupBy (>=) $ each [1,2,3,1,2,3,4,3,2,4,5,6,7,6,5] +[1] +[2] +[3,1,2,3] +[4,3,2,4] +[5] +[6] +[7,6,5] + +-} +groupBy :: Monad m + => (a -> a -> Bool) + -> Stream (Of a) m r + -> Stream (Stream (Of a) m) m r +groupBy equals = loop where + loop stream = Effect $ do + e <- next stream + return $ case e of + Left r -> Return r + Right (a, p') -> Step $ + fmap loop (yield a >> span (equals a) p') +{-# INLINABLE groupBy #-} + + +{-| Group successive equal items together + +>>> S.toList $ mapped S.toList $ S.group $ each "baaaaad" +["b","aaaaa","d"] :> () + +>>> S.toList $ concats $ maps (S.drained . S.splitAt 1) $ S.group $ each "baaaaaaad" +"bad" :> () + +-} +group :: (Monad m, Eq a) => Stream (Of a) m r -> Stream (Stream (Of a) m) m r +group = groupBy (==) +{-#INLINE group #-} + + +head :: Monad m => Stream (Of a) m r -> m (Of (Maybe a) r) +head str = case str of + Return r -> return (Nothing :> r) + Effect m -> m >>= head + Step (a :> rest) -> effects rest >>= \r -> return (Just a :> r) +{-#INLINABLE head #-} + +head_ :: Monad m => Stream (Of a) m r -> m (Maybe a) +head_ str = case str of + Return r -> return Nothing + Effect m -> m >>= head_ + Step (a :> rest) -> effects rest >> return (Just a) +{-#INLINABLE head_ #-} + +intersperse :: Monad m => a -> Stream (Of a) m r -> Stream (Of a) m r +intersperse x str = case str of + Return r -> Return r + Effect m -> Effect (liftM (intersperse x) m) + Step (a :> rest) -> loop a rest + where + loop !a str = case str of + Return r -> Step (a :> Return r) + Effect m -> Effect (liftM (loop a) m) + Step (b :> rest) -> Step (a :> Step (x :> loop b rest)) +{-#INLINABLE intersperse #-} + + + + +-- --------------- +-- iterate +-- --------------- + +{-| Iterate a pure function from a seed value, streaming the results forever + + + +-} +iterate :: (a -> a) -> a -> Stream (Of a) m r +iterate f = loop where + loop a' = Step (a' :> loop (f a')) +{-# INLINABLE iterate #-} + +-- | Iterate a monadic function from a seed value, streaming the results forever +iterateM :: Monad m => (a -> m a) -> m a -> Stream (Of a) m r +iterateM f = loop where + loop ma = Effect $ do + a <- ma + return (Step (a :> loop (f a))) +{-# INLINABLE iterateM #-} + + + +last :: Monad m => Stream (Of a) m r -> m (Of (Maybe a) r) +last = loop Nothing_ where + loop m str = case str of + Return r -> case m of + Nothing_ -> return (Nothing :> r) + Just_ a -> return (Just a :> r) + Effect m -> m >>= last + Step (a :> rest) -> loop (Just_ a) rest +{-#INLINABLE last #-} + + + +last_ :: Monad m => Stream (Of a) m r -> m (Maybe a) +last_ = loop Nothing_ where + loop m str = case str of + Return r -> case m of + Nothing_ -> return Nothing + Just_ a -> return (Just a) + Effect m -> m >>= last_ + Step (a :> rest) -> loop (Just_ a) rest +{-#INLINABLE last_ #-} + +-- --------------- +-- length +-- --------------- + +{-| Run a stream, remembering only its length: + +>>> S.length $ S.each [1..10] +10 + +-} +length_ :: Monad m => Stream (Of a) m r -> m Int +length_ = fold_ (\n _ -> n + 1) 0 id +{-#INLINE length_#-} + +{-| Run a stream, keeping its length and its return value. + +>>> S.print $ mapped S.length $ chunksOf 3 $ S.each [1..10] +3 +3 +3 +1 + +-} + +length :: Monad m => Stream (Of a) m r -> m (Of Int r) +length = fold (\n _ -> n + 1) 0 id +{-#INLINE length #-} +-- --------------- +-- map +-- --------------- + +{-| Standard map on the elements of a stream. + +>>> S.stdoutLn $ S.map reverse $ each (words "alpha beta") +ahpla +ateb +-} +map :: Monad m => (a -> b) -> Stream (Of a) m r -> Stream (Of b) m r +map f = maps (\(x :> rest) -> f x :> rest) + -- loop where + -- loop stream = case stream of + -- Return r -> Return r + -- Effect m -> Effect (liftM loop m) + -- Step (a :> as) -> Step (f a :> loop as) +{-# INLINABLE map #-} + + +{-| Replace each element of a stream with the result of a monadic action + +>>> S.print $ S.mapM readIORef $ S.chain (\ior -> modifyIORef ior (*100)) $ S.mapM newIORef $ each [1..6] +100 +200 +300 +400 +500 +600 +-} +mapM :: Monad m => (a -> m b) -> Stream (Of a) m r -> Stream (Of b) m r +mapM f = loop where + loop str = case str of + Return r -> Return r + Effect m -> Effect (liftM loop m) + Step (a :> as) -> Effect $ do + a' <- f a + return (Step (a' :> loop as) ) +{-# INLINABLE mapM #-} + + + +{-| Reduce a stream to its return value with a monadic action. + +>>> S.mapM_ Prelude.print $ each [1..5] +1 +2 +3 +4 +5 +>>> rest <- S.mapM_ Prelude.print $ S.splitAt 3 $ each [1..10] +1 +2 +3 +>>> S.sum rest +49 :> () + +-} +mapM_ :: Monad m => (a -> m b) -> Stream (Of a) m r -> m r +mapM_ f = loop where + loop str = case str of + Return r -> return r + Effect m -> m >>= loop + Step (a :> as) -> do + f a + loop as +{-# INLINABLE mapM_ #-} + +{-| Fold streamed items into their monoidal sum + +>>> S.mconcat $ S.take 2 $ S.map (Data.Monoid.Last . Just) (S.stdinLn) +first +last +Last {getLast = Just "last"} :> () + + -} +mconcat :: (Monad m, Monoid w) => Stream (Of w) m r -> m (Of w r) +mconcat = fold mappend mempty id +{-#INLINE mconcat #-} + +data Maybe_ a = Just_ !a | Nothing_ +mconcat_ :: (Monad m, Monoid w) => Stream (Of w) m r -> m w +mconcat_ = fold_ mappend mempty id + +minimum :: (Monad m, Ord a) => Stream (Of a) m r -> m (Of (Maybe a) r) +minimum = fold (\m a -> case m of Nothing_ -> Just_ a ; Just_ a' -> Just_ (min a a')) + Nothing_ + (\m -> case m of Nothing_ -> Nothing; Just_ r -> Just r) +{-#INLINE minimum #-} + +minimum_ :: (Monad m, Ord a) => Stream (Of a) m r -> m (Maybe a) +minimum_ = fold_ (\m a -> case m of Nothing_ -> Just_ a ; Just_ a' -> Just_ (min a a')) + Nothing_ + (\m -> case m of Nothing_ -> Nothing; Just_ r -> Just r) +{-#INLINE minimum_ #-} + +maximum :: (Monad m, Ord a) => Stream (Of a) m r -> m (Of (Maybe a) r) +maximum = fold (\m a -> case m of Nothing_ -> Just_ a ; Just_ a' -> Just_ (max a a')) + Nothing_ + (\m -> case m of Nothing_ -> Nothing; Just_ r -> Just r) +{-#INLINE maximum #-} + +maximum_ :: (Monad m, Ord a) => Stream (Of a) m r -> m (Maybe a) +maximum_ = fold_ (\m a -> case m of Nothing_ -> Just_ a ; Just_ a' -> Just_ (max a a')) + Nothing_ + (\m -> case m of Nothing_ -> Nothing; Just_ r -> Just r) +{-#INLINE maximum_ #-} + +{-| The standard way of inspecting the first item in a stream of elements, if the + stream is still \'running\'. The @Right@ case contains a + Haskell pair, where the more general @inspect@ would return a left-strict pair. + There is no reason to prefer @inspect@ since, if the @Right@ case is exposed, + the first element in the pair will have been evaluated to whnf. + +> next :: Monad m => Stream (Of a) m r -> m (Either r (a, Stream (Of a) m r)) +> inspect :: Monad m => Stream (Of a) m r -> m (Either r (Of a (Stream (Of a) m r))) + + Interoperate with @pipes@ producers thus: + +> Pipes.unfoldr Stream.next :: Stream (Of a) m r -> Producer a m r +> Stream.unfoldr Pipes.next :: Producer a m r -> Stream (Of a) m r + + Similarly: + +> IOStreams.unfoldM (liftM (either (const Nothing) Just) . next) :: Stream (Of a) IO b -> IO (InputStream a) +> Conduit.unfoldM (liftM (either (const Nothing) Just) . next) :: Stream (Of a) m r -> Source a m r + + But see 'uncons', which is better fitted to these @unfoldM@s +-} +next :: Monad m => Stream (Of a) m r -> m (Either r (a, Stream (Of a) m r)) +next = loop where + loop stream = case stream of + Return r -> return (Left r) + Effect m -> m >>= loop + Step (a :> rest) -> return (Right (a,rest)) +{-# INLINABLE next #-} + +{-| Remove repeated elements from a Stream. 'nub' of course accumulates a 'Data.Set.Set' of + elements that have already been seen and should thus be used with care. + +>>> S.toList_ $ S.nub $ S.take 5 S.readLn :: IO ([Int]) +1 +2 +3 +1 +2 +[1,2,3] + +-} +nub :: (Monad m, Ord a) => Stream (Of a) m r -> Stream (Of a) m r +nub = loop Set.empty where + loop !set stream = case stream of + Return r -> Return r + Effect m -> Effect (liftM (loop set) m) + Step (a :> rest) -> if Set.member a set + then loop set rest + else Step (a :> loop (Set.insert a set) rest) + +-- | Fold a 'Stream' of numbers into their product +product_ :: (Monad m, Num a) => Stream (Of a) m () -> m a +product_ = fold_ (*) 1 id +{-# INLINE product_ #-} + +{-| Fold a 'Stream' of numbers into their product with the return value + +> maps' product' :: Stream (Stream (Of Int)) m r -> Stream (Of Int) m r +-} +product :: (Monad m, Num a) => Stream (Of a) m r -> m (Of a r) +product = fold (*) 1 id +{-# INLINE product #-} + + +-- --------------- +-- read +-- --------------- + +{- | Make a stream of strings into a stream of parsed values, skipping bad cases + +>>> S.sum_ $ S.read $ S.takeWhile (/= "total") S.stdinLn :: IO Int +1000 +2000 +total +3000 + + +-} +read :: (Monad m, Read a) => Stream (Of String) m r -> Stream (Of a) m r +read stream = for stream $ \str -> case readMaybe str of + Nothing -> return () + Just r -> yield r +{-# INLINE read #-} + +-- --------------- +-- 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. + +>>> S.toList $ S.take 2 $ repeatM getLine +one +two +["one","two"] +-} + +repeatM :: Monad m => m a -> Stream (Of a) m r +repeatM ma = loop where + loop = do + a <- lift ma + yield a + loop +{-# INLINABLE repeatM #-} + +-- --------------- +-- replicate +-- --------------- + +-- | Repeat an element several times +replicate :: Monad m => Int -> a -> Stream (Of a) m () +replicate n a = loop n where + loop 0 = Return () + loop m = Step (a :> loop (m-1)) +{-# INLINABLE replicate #-} + +{-| Repeat an action several times, streaming the results. + +>>> S.print $ S.replicateM 2 getCurrentTime +2015-08-18 00:57:36.124508 UTC +2015-08-18 00:57:36.124785 UTC + +-} +replicateM :: Monad m => Int -> m a -> Stream (Of a) m () +replicateM n ma = loop n where + loop 0 = Return () + loop n = Effect $ do + a <- ma + return (Step $ a :> loop (n-1)) +{-# INLINABLE 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 = Effect $ do + m <- step s + case m of + Nothing -> return (Return ()) + Just a -> return (Step (a :> loop)) +{-# INLINABLE reread #-} + +{-| Strict left scan, streaming, e.g. successive partial results. + + +>>> S.print $ S.scan (++) "" id $ each (words "a b c d") +"" +"a" +"ab" +"abc" +"abcd" + + 'scan' is fitted for use with @Control.Foldl@, thus: + +>>> S.print $ L.purely S.scan L.list $ each [3..5] +[] +[3] +[3,4] +[3,4,5] + +-} +scan :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> Stream (Of b) m r +scan step begin done = loop begin + where + loop !x stream = Step $ done x :> + case stream of + Return r -> Return r + Effect m -> Effect $ liftM (loop x) m + Step (a :> rest) -> loop (step x a) rest +{-# INLINABLE scan #-} + +{-| Strict left scan, accepting a monadic function. It can be used with + 'FoldM's from @Control.Foldl@ using 'impurely'. Here we yield + a succession of vectors each recording + +>>> 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 + x <- lift begin + loop x str + where + loop !x stream = do + b <- lift (done x) + yield b + case stream of + Return r -> Return r + Effect m -> Effect (do + stream' <- m + return (loop x stream') + ) + Step (a :> rest) -> Effect (do + x' <- step x a + return (loop x' rest) + ) +{-# INLINABLE scanM #-} + +{- Label each element in a stream with a value accumulated according to a fold. + + +>>> S.print $ S.scanned (*) 1 id $ S.each [100,200,300] +(100,100) +(200,20000) +(300,6000000) + +>>> S.print $ L.purely S.scanned L.product $ S.each [100,200,300] +(100,100) +(200,20000) +(300,6000000) + +-} + +data Maybe' a = Just' a | Nothing' + +scanned :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> Stream (Of (a,b)) m r +scanned step begin done = loop Nothing' begin + where + loop !m !x stream = do + case stream of + Return r -> return r + Effect mn -> Effect $ liftM (loop m x) mn + Step (a :> rest) -> do + case m of + Nothing' -> do + let !acc = step x a + yield (a, done acc) + loop (Just' a) acc rest + Just' _ -> do + let !acc = done (step x a) + yield (a, acc) + loop (Just' a) (step x a) rest +{-# INLINABLE scanned #-} + + +{-| Streams the number of seconds from the beginning of action + + Thus, to mark times of user input we might write something like: + +>>> S.toList $ S.take 3 $ S.zip S.seconds S.stdinLn +a +b +c +[(0.0,"a"),(1.088711,"b"),(3.7289649999999996,"c")] :> () + + To restrict user input to some number of seconds, we might write: + +>>> S.toList $ S.zipWith (flip const) (S.takeWhile (< 5) S.seconds) S.stdinLn +one +two +three +four +five +["one","two","three","four","five"] :> () + + -} + +seconds :: Stream (Of Double) IO r +seconds = do + e <- lift $ next preseconds + case e of + Left r -> return r + Right (t, rest) -> do + yield 0 + map (subtract t) rest + where + preseconds :: Stream (Of Double) IO r + preseconds = do + utc <- liftIO getCurrentTime + map ((/1000000000) . nice utc) (repeatM getCurrentTime) + where + nice u u' = fromIntegral $ truncate (1000000000 * diffUTCTime u' u) + +-- --------------- +-- sequence +-- --------------- + +{-| Like the 'Data.List.sequence' but streaming. The result type is a + stream of a\'s, /but is not accumulated/; the effects of the elements + of the original stream are interleaved in the resulting stream. Compare: + +> sequence :: Monad m => [m a] -> m [a] +> sequence :: Monad m => Stream (Of (m a)) m r -> Stream (Of a) m r + + This obeys the rule + +-} +sequence :: Monad m => Stream (Of (m a)) m r -> Stream (Of a) m r +sequence = loop where + loop stream = case stream of + Return r -> Return r + Effect m -> Effect $ liftM loop m + Step (ma :> rest) -> Effect $ do + a <- ma + return (Step (a :> loop rest)) +{-# INLINABLE sequence #-} + +-- --------------- +-- show +-- --------------- + +show :: (Monad m, Show a) => Stream (Of a) m r -> Stream (Of String) m r +show = map Prelude.show +{-# INLINE show #-} +-- --------------- +-- sum +-- --------------- + +-- | Fold a 'Stream' of numbers into their sum +sum_ :: (Monad m, Num a) => Stream (Of a) m () -> m a +sum_ = fold_ (+) 0 id +{-# INLINE sum_ #-} + +{-| Fold a 'Stream' of numbers into their sum with the return value + +> mapped S.sum :: Stream (Stream (Of Int)) m r -> Stream (Of Int) m r + + +>>> S.sum $ each [1..10] +55 :> () + +>>> (n :> rest) <- S.sum $ S.splitAt 3 $ each [1..10] +>>> print n +6 +>>> (m :> rest') <- S.sum $ S.splitAt 3 rest +>>> print m +15 +>>> S.print rest' +7 +8 +9 + +-} +sum :: (Monad m, Num a) => Stream (Of a) m r -> m (Of a r) +sum = fold (+) 0 id +{-# INLINABLE sum #-} + +-- --------------- +-- span +-- --------------- + +-- | Stream elements until one fails the condition, return the rest. +span :: Monad m => (a -> Bool) -> Stream (Of a) m r + -> Stream (Of a) m (Stream (Of a) m r) +span pred = loop where + loop str = case str of + Return r -> Return (Return r) + Effect m -> Effect $ liftM loop m + Step (a :> rest) -> if pred a + then Step (a :> loop rest) + else Return (Step (a :> rest)) +{-# INLINABLE span #-} + + +{-| Split a stream of elements wherever a given element arises. + The action is like that of 'Prelude.words'. + +>>> S.stdoutLn $ mapped S.toList $ S.split ' ' $ each "hello world " +hello +world + +-} + +split :: (Eq a, Monad m) => + a -> Stream (Of a) m r -> Stream (Stream (Of a) m) m r +split t = loop where + loop stream = case stream of + Return r -> Return r + Effect m -> Effect (liftM loop m) + Step (a :> rest) -> + if a /= t + then Step (fmap loop (yield a >> break (== t) rest)) + else loop rest +{-#INLINABLE split #-} + +{-| 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 #-} + + +-- --------------- +-- take +-- --------------- + +{-| End a stream after n elements; the original return value is thus lost. + 'splitAt' preserves this information. Note that, like @splitAt@, this + function is functor-general, so that, for example, you can @take@ not + just a number of items from a stream of elements, but a number + of substreams and the like. + +>>> S.toList $ S.take 3 $ each "pennsylvania" +"pen" :> () + +>>> total <- S.sum_ $ S.take 3 S.readLn :: IO Int +1 +10 +100 +>>> print total +111 + +-} + +take :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m () +take = loop where + loop 0 p = return () + loop n p = + case p of Step fas -> Step (fmap (loop (n-1)) fas) + Effect m -> Effect (liftM (loop n) m) + Return r -> Return () +{-# INLINABLE take #-} + +-- --------------- +-- takeWhile +-- --------------- + +{-| End stream when an element fails a condition; the original return value is lost. + By contrast 'span' preserves this information. + + +-} +takeWhile :: Monad m => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m () +takeWhile pred = loop where + loop str = case str of + Step (a :> as) -> when (pred a) (Step (a :> loop as)) + Effect m -> Effect (liftM loop m) + Return r -> Return () +{-# INLINE takeWhile #-} + + +{-| Convert an effectful 'Stream (Of a)' into a list of @as@ + + 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. + +-} +toList_ :: Monad m => Stream (Of a) m () -> m [a] +toList_ = fold_ (\diff a ls -> diff (a: ls)) id (\diff -> diff []) +{-# INLINE toList_ #-} + + +{-| Convert an effectful 'Stream' into a list alongside the return value + +> mapped toListM :: Stream (Stream (Of a)) m r -> Stream (Of [a]) m +-} +toList :: Monad m => Stream (Of a) m r -> m (Of [a] r) +toList = fold (\diff a ls -> diff (a: ls)) id (\diff -> diff []) +{-# INLINE toList #-} + + +{-| Inspect the first item in a stream of elements, without a return value. + @uncons@ provides convenient exit into another streaming type: + +> IOStreams.unfoldM uncons :: Stream (Of a) IO b -> IO (InputStream a) +> 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 ())) +uncons = loop where + loop stream = case stream of + Return () -> return Nothing + Effect m -> m >>= loop + Step (a :> rest) -> return (Just (a,rest)) +{-# INLINABLE uncons #-} + + +{-| Build a @Stream@ by unfolding steps starting from a seed. + + The seed can of course be anything, but this is one natural way + to consume a @pipes@ 'Pipes.Producer'. Consider: + +>>> S.stdoutLn $ S.take 2 $ S.unfoldr P.next P.stdinLn +hello +hello +goodbye +goodbye + +>>> S.stdoutLn $ S.unfoldr P.next (P.stdinLn P.>-> P.take 2) +hello +hello +goodbye +goodbye + +>>> S.effects $ S.unfoldr P.next (P.stdinLn P.>-> P.take 2 P.>-> P.stdoutLn) +hello +hello +goodbye +goodbye + +-} +unfoldr :: Monad m + => (s -> m (Either r (a, s))) -> s -> Stream (Of a) m r +unfoldr step = loop where + loop s0 = Effect (do + e <- step s0 + case e of + Left r -> return (Return r) + Right (a,s) -> return (Step (a :> loop s))) +{-# INLINABLE unfoldr #-} + +-- --------------------------------------- +-- with +-- --------------------------------------- + +{-| Replace each element in a stream of individual Haskell values (a @Stream (Of a) m r@) with an associated 'functorial' step. + +> for str f = concats (with str f) +> with str f = for str (yields . f) +> with str f = maps (\(a:>r) -> r <$ f a) str + +>>> with (each [1..3]) (yield . show) & intercalates (yield "--") & S.stdoutLn +1 +-- +2 +-- +3 + -} +with :: (Monad m, Functor f) => Stream (Of a) m r -> (a -> f x) -> Stream f m r +with s f = loop s where + loop str = case str of + Return r -> Return r + Effect m -> Effect (liftM loop m) + Step (a :> rest) -> Step (loop rest <$ f a) +{-#INLINABLE with #-} + +-- --------------------------------------- +-- yield +-- --------------------------------------- + +{-| A singleton stream + +>>> stdoutLn $ yield "hello" +hello + +>>> S.sum $ do {yield 1; yield 2} +3 + +>>> let prompt = putStrLn "Enter a number:" +>>> let number = lift (prompt >> readLn) >>= yield :: Stream (Of Int) IO () +>>> S.toList $ do {number; number; number} +Enter a number: +1 +Enter a number: +2 +Enter a number: +3 +[1,2,3] :> () + +-} + +yield :: Monad m => a -> Stream (Of a) m () +yield a = Step (a :> Return ()) +{-# INLINE yield #-} + +-- | Zip two 'Streams's +zip :: Monad m + => (Stream (Of a) m r) + -> (Stream (Of b) m r) + -> (Stream (Of (a,b)) m r) +zip = zipWith (,) +{-# INLINE zip #-} + +-- | Zip two 'Streams's using the provided combining function +zipWith :: Monad m + => (a -> b -> c) + -> (Stream (Of a) m r) + -> (Stream (Of b) m r) + -> (Stream (Of c) m r) +zipWith f = loop + where + loop str0 str1 = case str0 of + Return r -> Return r + Effect m -> Effect $ liftM (\str -> loop str str1) m + Step (a :> rest0) -> case str1 of + Return r -> Return r + Effect m -> Effect $ liftM (loop str0) m + Step (b :> rest1) -> Step (f a b :>loop rest0 rest1) +{-# INLINABLE zipWith #-} + + +-- | Zip three 'Stream's with a combining function +zipWith3 :: Monad m => + (a -> b -> c -> d) + -> Stream (Of a) m r + -> Stream (Of b) m r + -> Stream (Of c) m r + -> Stream (Of d) m r +zipWith3 op = loop where + loop str0 str1 str2 = do + e0 <- lift (next str0) + case e0 of + Left r0 -> return r0 + Right (a0,rest0) -> do + e1 <- lift (next str1) + case e1 of + Left r1 -> return r1 + Right (a1,rest1) -> do + e2 <- lift (next str2) + case e2 of + Left r2 -> return r2 + Right (a2,rest2) -> do + yield (op a0 a1 a2) + loop rest0 rest1 rest2 +{-# INLINABLE zipWith3 #-} + + +-- | Zip three streams together +zip3 :: Monad m + => (Stream (Of a) m r) + -> (Stream (Of b) m r) + -> (Stream (Of c) m r) + -> (Stream (Of (a,b,c)) m r) +zip3 = zipWith3 (,,) +{-# INLINABLE zip3 #-} + +-- -------------- +-- IO fripperies +-- -------------- + +{-| View standard input as a 'Stream (Of String) m r'. 'stdoutLn', by + contrast, renders a 'Stream (Of String) m r' to standard output. The names + follow @Pipes.Prelude@ + +>>> stdoutLn stdinLn +hello +hello +world +world +^CInterrupted. + + +>>> stdoutLn $ S.map reverse stdinLn +hello +olleh +world +dlrow +^CInterrupted. + +-} +stdinLn :: MonadIO m => Stream (Of String) m () +stdinLn = fromHandle IO.stdin +{-# INLINABLE stdinLn #-} + +{-| Read values from 'IO.stdin', ignoring failed parses + +>>> S.sum_ $ S.take 2 S.readLn :: IO Int +10 +12 +22 + +>>> S.toList $ S.take 3 (S.readLn :: Stream (Of Int) IO ()) +1 +2 +1@#$%^&*\ +3 +[1,2,3] :> () + + +-} + +readLn :: (MonadIO m, Read a) => Stream (Of a) m () +readLn = for stdinLn $ \str -> case readMaybe str of + Nothing -> return () + Just n -> yield n +{-# INLINABLE readLn #-} + + +{-| Read 'String's from a 'IO.Handle' using 'IO.hGetLine' + + Terminates on end of input + +>>> IO.withFile "/usr/share/dict/words" IO.ReadMode $ S.stdoutLn . S.take 3 . S.drop 50000 . S.fromHandle +deflagrator +deflate +deflation + +-} +fromHandle :: MonadIO m => IO.Handle -> Stream (Of String) m () +fromHandle h = go + where + go = do + eof <- liftIO $ IO.hIsEOF h + unless eof $ do + str <- liftIO $ IO.hGetLine h + yield str + go +{-# INLINABLE fromHandle #-} + +{-| Write a succession of strings to a handle as separate lines. + +>>> S.toHandle IO.stdout $ each $ words "one two three" +one +two +three +-} +toHandle :: MonadIO m => IO.Handle -> Stream (Of String) m r -> m r +toHandle handle = loop where + loop str = case str of + Return r -> return r + Effect m -> m >>= loop + Step (s :> rest) -> do + liftIO (IO.hPutStrLn handle s) + loop rest +{-# INLINABLE toHandle #-} + +{-| Print the elements of a stream as they arise. + +>>> S.print $ S.take 2 S.stdinLn +hello +"hello" +world +"world" +>>> + +-} +print :: (MonadIO m, Show a) => Stream (Of a) m r -> m r +print = loop where + loop stream = case stream of + Return r -> return r + Effect m -> m >>= loop + Step (a :> rest) -> do + liftIO (Prelude.print a) + loop rest + + +{-| Write 'String's to 'IO.stdout' using 'putStrLn'; terminates on a broken output pipe + (This operation is modelled on 'Pipes.Prelude.stdoutLn'). + +>>> S.stdoutLn $ S.take 3 $ S.each $ words "one two three four five" +one +two +three +-} +stdoutLn :: MonadIO m => Stream (Of String) m () -> m () +stdoutLn = loop + where + loop stream = case stream of + Return _ -> return () + Effect m -> m >>= loop + Step (s :> rest) -> do + x <- liftIO $ try (putStrLn s) + case x of + Left (G.IOError { G.ioe_type = G.ResourceVanished + , G.ioe_errno = Just ioe }) + | Errno ioe == ePIPE + -> return () + Left e -> liftIO (throwIO e) + Right () -> loop rest +{-# INLINABLE stdoutLn #-} + + + + +{-| Write 'String's to 'IO.stdout' using 'putStrLn' + + This does not handle a broken output pipe, but has a polymorphic return + value, which makes this possible: + +>>> rest <- S.stdoutLn' $ S.show $ S.splitAt 3 (each [1..5]) +1 +2 +3 +>>> S.print rest +4 +5 + +-} + +stdoutLn' :: MonadIO m => Stream (Of String) m r -> m r +stdoutLn' = loop where + loop stream = case stream of + Return r -> return r + Effect m -> m >>= loop + Step (s :> rest) -> liftIO (putStrLn s) >> loop rest +{-# INLINE stdoutLn' #-} + +{-| Read a series of strings as lines to a file. + +>>> runResourceT $ S.writeFile "lines.txt" $ S.take 2 S.stdinLn +hello +world +>>> runResourceT $ S.print $ S.readFile "lines.txt" +"hello" +"world" + + 'runResourceT', as it is used here, means something like 'closing_handles'; + it makes it possible to write convenient, fairly sensible versions of + 'readFile', 'writeFile' and 'appendFile'. Its use is explained + . + +-} + +readFile :: MonadResource m => FilePath -> Stream (Of String) m () +readFile f = bracketStream (IO.openFile f IO.ReadMode) (IO.hClose) fromHandle + +{-| Write a series of strings as lines to a file. The handle is crudely + managed with 'ResourceT': + +>>> runResourceT $ S.writeFile "lines.txt" $ S.take 2 S.stdinLn +hello +world +>>> runResourceT $ S.print $ S.readFile "lines.txt" +"hello" +"world" + +-} +writeFile :: MonadResource m => FilePath -> Stream (Of String) m r -> m r +writeFile f str = do + (key, handle) <- allocate (IO.openFile f IO.WriteMode) (IO.hClose) + r <- toHandle handle str + release key + return r + +-- -- * Producers +-- -- $producers +-- stdinLn -- +-- , readLn -- +-- , fromHandle -- +-- , repeatM -- +-- , replicateM -- +-- +-- -- * Consumers +-- -- $consumers +-- , stdoutLn -- +-- , stdoutLn' -- +-- , mapM_ -- +-- , print -- +-- , toHandle -- +-- , effects -- +-- +-- -- * Pipes +-- -- $pipes +-- , map -- +-- , mapM -- +-- , sequence -- +-- , mapFoldable -- +-- , filter -- +-- , filterM -- +-- , take -- +-- , takeWhile -- +-- , takeWhile' -- +-- , drop -- +-- , dropWhile -- +-- , concat -- +-- , elemIndices +-- , findIndices +-- , scan -- +-- , scanM -- +-- , chain -- +-- , read -- +-- , show -- +-- , seq -- +-- +-- -- * Folds +-- -- $folds +-- , fold -- +-- , fold' -- +-- , foldM -- +-- , foldM' -- +-- , all +-- , any +-- , and +-- , or +-- , elem +-- , notElem +-- , find +-- , findIndex +-- , head +-- , index +-- , last +-- , length +-- , maximum +-- , minimum +-- , null +-- , sum -- +-- , product -- +-- , toList -- +-- , toListM -- +-- , toListM' -- +-- +-- -- * Zips +-- , zip -- +-- , zipWith -- +-- + +distinguish :: (a -> Bool) -> Of a r -> Sum (Of a) (Of a) r +distinguish predicate (a :> b) = if predicate a then InR (a :> b) else InL (a :> b) +{-#INLINE distinguish #-} + +sumToEither ::Sum (Of a) (Of b) r -> Of (Either a b) r +sumToEither s = case s of + InL (a :> r) -> Left a :> r + InR (b :> r) -> Right b :> r +{-#INLINE sumToEither #-} + +eitherToSum :: Of (Either a b) r -> Sum (Of a) (Of b) r +eitherToSum s = case s of + Left a :> r -> InL (a :> r) + Right b :> r -> InR (b :> r) +{-#INLINE eitherToSum #-} + +composeToSum :: Compose (Of Bool) f r -> Sum f f r +composeToSum x = case x of + Compose (True :> f) -> InR f + Compose (False :> f) -> InL f +{-#INLINE composeToSum #-} + +sumToCompose :: Sum f f r -> Compose (Of Bool) f r +sumToCompose x = case x of + InR f -> Compose (True :> f) + InL f -> Compose (False :> f) +{-#INLINE sumToCompose #-} + +{-| Store the result of any suitable fold over a stream, keeping the stream for + further manipulation. @store f = f . duplicate@ : + +>>> S.print $ S.store S.product $ each [1..4] +1 +2 +3 +4 +24 :> () + +>>> S.print $ S.store S.sum $ S.store S.product $ each [1..4] +1 +2 +3 +4 +10 :> (24 :> ()) + + Here the sum (10) and the product (24) have been \'stored\' for use when + finally we have traversed the stream with 'print' . Needless to say, + a second 'pass' is excluded conceptually, so the + folds that you apply successively with @store@ are performed + simultaneously, and in constant memory -- as they would be if, + say, you linked them together with @Control.Fold@: + +>>> L.impurely S.foldM (liftA3 (\a b c -> (b,c)) (L.sink print) (L.generalize L.sum) (L.generalize L.product)) $ each [1..4] +1 +2 +3 +4 +(10,24) :> () + + Fusing folds after the fashion of @Control.Foldl@ will generally be a bit faster + than the corresponding succession of uses of 'store', but by + constant factor that will be completely dwarfed when any IO is at issue. + + But 'store' / 'duplicate' is /much/ more powerful, as you can see by reflecting on + uses like this: + +>>> S.sum $ S.store (S.sum . mapped S.product . chunksOf 2) $ S.store (S.product . mapped S.sum . chunksOf 2 )$ each [1..6] +21 :> (44 :> (231 :> ())) + + It will be clear that this cannot be reproduced with any combination of lenses, + @Control.Fold@ folds, or the like. (See also the discussion of 'duplicate'.) + + 'store' is intended to be used at types like these + +> storeM :: (Monad m => Stream (Of a) m r -> m (Of b r)) +> -> (Monad n => Stream (Of a) n r -> Stream (Of a) n (Of b r)) +> storeM = store +> +> storeMIO :: (MonadIO m => Stream (Of a) m r -> m (Of b r)) +> -> ( MonadIO n => Stream (Of a) n r -> Stream (Of a) n (Of b r) +> storeMIO = store + + And similarly for other constraints that @Stream (Of a)@ inherits, + like 'MonadResource'. Thus I can filter and write to one file, but + nub and write to another: + +>>> runResourceT $ (S.writeFile "hello2.txt" . S.nub) $ store (S.writeFile "hello.txt" . S.filter (/= "world")) $ each ["hello", "world", "goodbye", "world"] +>>> :! cat hello.txt +hello +goodbye +>>> :! cat hello2.txt +hello +world +goodbye + + +-} +store + :: Monad m => + (Stream (Of a) (Stream (Of a) m) r -> t) -> Stream (Of a) m r -> t +store f x = f (duplicate x) +{-#INLINE store #-} + +{-| Duplicate the content of stream, so that it can be acted on twice in different ways, + but without breaking streaming. Thus, given: + +>>> S.print $ each ["one","two"] +"one" +"two" +>>> S.stdoutLn $ each ["one","two"] +one +two + + I can as well do: + +>>> S.print $ S.stdoutLn $ S.duplicate $ each ["one","two"] +one +"one" +two +"two" + + Where the actions you are contemplating are each simple folds over + the elements, or a selection of elements, then the coupling of the + folds is often more straightforwardly effected with `Control.Foldl`, + e.g. + +>>> L.purely S.fold (liftA2 (,) L.sum L.product) $ each [1..10] +(55,3628800) :> () + + rather than + +>>> S.sum $ S.product . S.duplicate $ each [1..10] +55 :> (3628800 :> ()) + + A @Control.Foldl@ fold can be altered to act on a selection of elements by + using 'Control.Foldl.handles' on an appropriate lens. Some such + manipulations are simpler and more 'Data.List'-like, using 'duplicate': + +>>> L.purely S.fold (liftA2 (,) (L.handles (filtered odd) L.sum) (L.handles (filtered even) L.product)) $ each [1..10] +(25,3840) :> () + + becomes + +>>> S.sum $ S.filter odd $ S.product $ S.filter even $ S.duplicate $ each [1..10] +25 :> (3840 :> ()) + + or using 'store' + +>>> S.sum $ S.filter odd $ S.store (S.product . S.filter even) $ each [1..10] +25 :> (3840 :> ()) + + But anything that fold of a @Stream (Of a) m r@ into e.g. an @m (Of b r)@ + that has a constraint on @m@ that is carried over into @Stream f m@ - + e.g. @Monad@, @MonadIO@, @MonadResource@, etc. can be used on the stream. + Thus, I can fold over different groupings of the original stream: + +>>> (S.toList . mapped S.toList . chunksOf 5) $ (S.toList . mapped S.toList . chunksOf 3) $ S.duplicate $ each [1..10] +[[1,2,3,4,5],[6,7,8,9,10]] :> ([[1,2,3],[4,5,6],[7,8,9],[10]] :> ()) + + The procedure can be iterated as one pleases, as one can see from this (otherwise unadvisable!) example: + +>>> (S.toList . mapped S.toList . chunksOf 4) $ (S.toList . mapped S.toList . chunksOf 3) $ S.duplicate $ (S.toList . mapped S.toList . chunksOf 2) $ S.duplicate $ each [1..12] +[[1,2,3,4],[5,6,7,8],[9,10,11,12]] :> ([[1,2,3],[4,5,6],[7,8,9],[10,11,12]] :> ([[1,2],[3,4],[5,6],[7,8],[9,10],[11,12]] :> ())) + +-} +duplicate + :: Monad m => + Stream (Of a) m r -> Stream (Of a) (Stream (Of a) m) r +duplicate = loop where + loop str = case str of + Return r -> Return r + Effect m -> Effect (liftM loop (lift m)) + Step (a :> rest) -> Step (a :> Effect (Step (a :> Return (loop rest)))) +{-#INLINABLE duplicate#-} + +{-| The type + +> Data.List.unzip :: [(a,b)] -> ([a],[b]) + + might lead us to expect + +> Streaming.unzip :: Stream (Of (a,b)) m r -> Stream (Of a) m (Stream (Of b) m r) + + which would not stream. Of course, neither does 'Data.List.unzip' + +-} +unzip :: Monad m => Stream (Of (a,b)) m r -> Stream (Of a) (Stream (Of b) m) r +unzip = loop where + loop str = case str of + Return r -> Return r + Effect m -> Effect (liftM loop (lift m)) + Step ((a,b):> rest) -> Step (a :> Effect (Step (b :> Return (loop rest)))) +{-#INLINABLE unzip #-} + + +-- "fold/map" forall step begin done f str . +-- fold step begin done (map f str) = fold (\x a -> step x $! f a) begin done str; +-- +-- "fold/filter" forall step begin done pred str . +-- fold step begin done (filter pred str) = fold (\x a -> if pred a then step x a else x) begin done str; +-- +-- "scan/map" forall step begin done f str . +-- scan step begin done (map f str) = scan (\x a -> step x $! f a) begin done str +-- diff --git a/streaming.cabal b/streaming.cabal --- a/streaming.cabal +++ b/streaming.cabal @@ -1,16 +1,45 @@ name: streaming -version: 0.1.2.2 +version: 0.1.3.0 cabal-version: >=1.10 build-type: Simple synopsis: an elementary streaming prelude and a general monad transformer for streaming applications. -description: @Streaming.Prelude@ exports an elementary streaming prelude; @Streaming@ exports a free monad transformer - optimized for streaming applications and replacing @FreeT@. See the +description: @Streaming.Prelude@ exports an elementary streaming prelude relating to + an elementary source\/generator\/producer type, @Stream (Of a) m r@. + @Streaming@ exports a much more general type, @Stream f m r@, which + can be used to 'stream' successive distinct steps characterized by any + functor @f@, though we are here interested only in a limited range of + cases. + . + The streaming-io libraries have various devices for dealing + with effectful variants of @[a]@ or @([a],r)@. But it is only with + the general type @Stream f m r@, or some equivalent, + that one can hope to stream streams, as one makes lists of + lists in the Haskell @Prelude@ and @Data.List@. Once one sees + the necessity of some such type if we are to + express a properly streaming equivalent of e.g. + + > groups :: Ord a => [a] -> [[a]] + > chunksOf :: Int -> [a] -> [[a]] + + and the like, then one will also see that, with it, + one is already in possession of a complete + elementary streaming library. The present @Streaming.Prelude@ is the + simplest streaming library that can replicate anything like the + API of the @Prelude@ and @Data.List@. + . + The emphasis of the library is on interoperation; for + the rest its advantages are: extreme simplicity and re-use of + intuitions the user has gathered from mastery of @Prelude@ and + @Data.List@. The two conceptual pre-requisites are some + comprehension of monad transformers and some familiarity + with \'rank 2 types\'. + . + See the below - for an explanation. Elementary usage can be divined from the ghci examples in - @Streaming.Prelude@ and from the remarks somewhat theoretical - - below, including the examples linked there. Note also the + for an explanation, including the examples linked there. Elementary usage can be divined from the ghci examples in + @Streaming.Prelude@ and perhaps from this rough beginning of a + Note also the and @@ -36,15 +65,15 @@ below. . - Here are some results for an - - of the little + Here are the results of some + + based on the included in the machines package: . <> . - + license: BSD3 license-file: LICENSE @@ -77,9 +106,13 @@ , mtl >=2.1 && <2.3 , mmorph >=1.0 && <1.2 , transformers >=0.4 && <0.5 + , transformers-base , bytestring , time - + , resourcet + , exceptions + , containers + default-language: Haskell2010