diff --git a/Streaming.hs b/Streaming.hs --- a/Streaming.hs +++ b/Streaming.hs @@ -6,8 +6,8 @@ Stream, -- * Constructing a 'Stream' on a base functor unfold, - for, construct, + for, replicates, repeats, repeatsM, @@ -21,11 +21,12 @@ inspect, -- * Eliminating a 'Stream' - destroy, intercalates, concats, iterTM, iterT, + destroy, + -- * Splitting and joining 'Stream's splitsAt, @@ -39,15 +40,23 @@ -- * re-exports MFunctor(..), - MonadTrans(..) + MMonad(..), + MonadTrans(..), + MonadIO(..), + Compose(..), + join, + liftA2, + liftA3, + void ) where -import Streaming.Internal +import Streaming.Internal import Streaming.Prelude -import Control.Monad.Morph (MFunctor(..)) +import Control.Monad.Morph import Control.Monad +import Control.Applicative import Control.Monad.Trans - +import Data.Functor.Compose {- $stream @@ -82,11 +91,22 @@ To avoid breaking reasoning principles, the constructors should not be used directly. A pattern-match should go by way of 'inspect' - \- or, in the producer case, 'Streaming.Prelude.next' - The constructors are exported by the 'Internal' module. + \- or, in the producer case, 'Streaming.Prelude.next'. These mirror + the type of @runFreeT@. The constructors are exported by the 'Internal' module. -} +{-| Map a stream to its church encoding; compare @Data.List.foldr@ + This is the @safe_destroy@ exported by the @Internal@ module. + Typical @FreeT@ operators can be defined in terms of @destroy@ + e.g. +> iterT :: (Functor f, Monad m) => (f (m a) -> m a) -> Stream f m a -> m a +> iterT out stream = destroy stream out join return +> iterTM :: (Functor f, Monad m, MonadTrans t, Monad (t m)) => (f (t m a) -> t m a) -> Stream f m a -> t m a +> iterTM out stream = destroy stream out (join . lift) return +> concats :: (Monad m, MonadTrans t, Monad (t m)) => Stream (t m) m a -> t m a +> concats stream = destroy stream join (join . lift) return +-} diff --git a/Streaming/Internal.hs b/Streaming/Internal.hs --- a/Streaming/Internal.hs +++ b/Streaming/Internal.hs @@ -30,6 +30,14 @@ -- * Splitting streams , chunksOf , splitsAt + + -- * For internal use + , unexposed + , hoistExposed + , mapsExposed + , mapsMExposed + , destroyExposed + ) where import Control.Monad @@ -113,36 +121,52 @@ {-# INLINE lift #-} instance Functor f => MFunctor (Stream f) where - hoist trans = loop where + hoist trans = loop . unexposed where loop stream = case stream of Return r -> Return r Delay m -> Delay (trans (liftM loop m)) Step f -> Step (fmap loop f) {-# INLINABLE hoist #-} + instance Functor f => MMonad (Stream f) where embed phi = loop where loop stream = case stream of Return r -> Return r - Delay m -> phi m >>= loop - Step f -> Step (fmap loop f) + Delay m -> phi m >>= loop + Step f -> Step (fmap loop f) {-# INLINABLE embed #-} - + instance (MonadIO m, Functor f) => MonadIO (Stream f m) where liftIO = Delay . liftM Return . liftIO {-# INLINE liftIO #-} --- | Map a stream to its church encoding; compare @Data.List.foldr@ -destroy + +{-| 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) => Stream f m r -> (f b -> b) -> (m b -> b) -> (r -> b) -> b -destroy stream0 construct wrap done = loop stream0 where +destroy stream0 construct wrap done = loop (unexposed stream0) where loop stream = case stream of Return r -> done r Delay m -> wrap (liftM loop m) Step fs -> construct (fmap loop fs) {-# INLINABLE destroy #-} + -- | Reflect a church-encoded stream; cp. @GHC.Exts.build@ construct :: (forall b . (f b -> b) -> (m b -> b) -> (r -> b) -> b) -> Stream f m r @@ -186,7 +210,7 @@ {-# INLINABLE unfold #-} --- | Map layers of one functor to another with a natural transformation +-- | Map layers of one functor to another with a transformation maps :: (Monad m, Functor f) => (forall x . f x -> g x) -> Stream f m r -> Stream g m r maps phi = loop where @@ -249,11 +273,7 @@ iterT out stream = destroy stream out join return {-# INLINE iterT #-} -{-| This specializes to the more transparent case: - -> concats :: (Monad m, Functor f) => Stream (Stream f m) m r -> Stream f m r - - Thus dissolving the segmentation into @Stream f m@ layers. +{-| Dissolves the segmentation into layers of @Stream f m@ layers. > concats stream = destroy stream join (join . lift) return @@ -268,10 +288,12 @@ 5 -} -concats :: - (MonadTrans t, Monad (t m), Monad m) => - Stream (t m) m a -> t m a -concats stream = destroy stream join (join . lift) return +concats :: (Monad m, Functor f) => Stream (Stream f m) m r -> Stream f m r +concats = loop where + loop stream = case stream of + Return r -> return r + Delay m -> join $ lift (liftM loop m) + Step fs -> join (fmap loop fs) {-# INLINE concats #-} {-| Split a succession of layers after some number, returning a streaming or @@ -375,3 +397,51 @@ cycles :: (Monad m, Functor f) => Stream f m () -> Stream f m r cycles = forever + + +hoistExposed trans = loop where + loop stream = case stream of + Return r -> Return r + Delay m -> Delay (trans (liftM loop m)) + Step f -> Step (fmap loop f) + +mapsExposed :: (Monad m, Functor f) + => (forall x . f x -> g x) -> Stream f m r -> Stream g m r +mapsExposed phi = loop where + loop stream = case stream of + Return r -> Return r + Delay m -> Delay (liftM loop m) + Step f -> Step (phi (fmap loop f)) +{-# INLINABLE mapsExposed #-} + +mapsMExposed phi = loop where + loop stream = case stream of + Return r -> Return r + Delay m -> Delay (liftM loop m) + Step f -> Delay (liftM Step (phi (fmap loop f))) +{-# INLINABLE mapsMExposed #-} +-- 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" + + +destroyExposed stream0 construct wrap done = loop stream0 where + loop stream = case stream of + Return r -> done r + Delay m -> wrap (liftM loop m) + Step fs -> construct (fmap loop fs) +{-# INLINABLE destroyExposed #-} + +unexposed :: (Functor f, Monad m) => Stream f m r -> Stream f m r +unexposed = Delay . loop where + loop stream = case stream of + Return r -> return (Return r) + Delay m -> m >>= loop + Step f -> return (Step (fmap (Delay . loop) f)) +{-# INLINABLE unexposed #-} diff --git a/Streaming/Prelude.hs b/Streaming/Prelude.hs --- a/Streaming/Prelude.hs +++ b/Streaming/Prelude.hs @@ -1,35 +1,37 @@ -{-| This module is very closely modeled on Pipes.Prelude. - - Import qualified thus: - -> import Streaming -> import qualified Streaming as S - - The @Streaming@ module exports types, functor-general operations and some other kit; - it may clash with @free@ and @pipes-group@, but not with standard base modules. +{-| 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. - Interoperation with @pipes@ is accomplished with this isomorphism, which - uses @Pipes.Prelude.unfoldr@ from @HEAD@: + Some interoperation incantations would be e.g. > Pipes.unfoldr Streaming.next :: Stream (Of a) m r -> Producer a m r > Streaming.unfoldr Pipes.next :: Producer a m r -> Stream (Of a) m r - - Interoperation with `iostreams` is thus: - > Streaming.reread IOStreams.read :: InputStream a -> Stream (Of a) IO () > IOStreams.unfoldM Streaming.uncons :: Stream (Of a) IO () -> IO (InputStream a) +> Conduit.unfoldM Streaming.uncons :: Stream (Of a) m () -> Source m a - A simple exit to conduit would be, for example: + Import qualified thus: -> Conduit.unfoldM Streaming.uncons :: Stream (Of a) m () -> Source m a +> import Streaming +> import qualified Streaming.Prelude as S + + For the examples below, one sometimes needs + +> import Streaming.Prelude (each, yield, stdoutLn, stdinLn) +> 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 + -} {-# LANGUAGE RankNTypes, BangPatterns, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveTraversable #-} module Streaming.Prelude ( -- * Types - Stream - , Of (..) + Of (..) , lazily , strictly @@ -41,9 +43,13 @@ , stdinLn , readLn , fromHandle + , iterate , repeat + , cycle , repeatM , replicateM + , enumFrom + , enumFromThen -- * Consuming streams of elements -- $consumers @@ -97,13 +103,15 @@ , sum' , product , product' + , length + , length' , toList , toListM , toListM' , foldrM , foldrT - -- * Short circuiting folds + -- , all -- , any -- , and @@ -127,6 +135,8 @@ -- * Interoperation , reread + -- * Basic Type + , Stream ) where import Streaming.Internal @@ -139,8 +149,8 @@ import qualified Data.Foldable as Foldable import Text.Read (readMaybe) import Prelude hiding (map, mapM, mapM_, filter, drop, dropWhile, take, sum, product - , iterate, repeat, replicate, splitAt - , takeWhile, enumFrom, enumFromTo + , iterate, repeat, cycle, replicate, splitAt + , takeWhile, enumFrom, enumFromTo, enumFromThen, length , print, zipWith, zip, seq, show, read , readLn, sequence, concat, span, break) @@ -190,14 +200,14 @@ {-| Apply an action to all values flowing downstream ->>> let debug str = chain print str ->>> S.product (debug (S.each [2..4])) >>= print + +>>> S.product (chain print (S.each [2..4])) >>= print 2 3 4 24 - -} + chain :: Monad m => (a -> m ()) -> Stream (Of a) m r -> Stream (Of a) m r chain f str = for str $ \a -> do lift (f a) @@ -206,25 +216,39 @@ {-| Make a stream of traversable containers into a stream of their separate elements ->>> Streaming.print $ concat (each ["hi","ho"]) -'h' -'i' -'h' -'o' - ->>> S.print $ S.concat (S.each [Just 1, Nothing, Just 2, Nothing]) +>>> S.print $ S.concat (each ["xy","z"]) +'x' +'y' +'z' +>>> 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 [Right 1, Left "error!", Right 2]) + 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 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 @@ -237,12 +261,42 @@ 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 + -- --------------- -- drain -- --------------- --- | Reduce a stream, performing its actions but ignoring its elements. +{- | Reduce a stream, performing its actions but ignoring its elements. + +>>> let stream = do {yield 1; lift (putStrLn "Effect!"); yield 2; lift (putStrLn "Effect!"); return (2^100)} + +>>> S.drain stream +Effect! +Effect! +1267650600228229401496703205376 + +>>> S.drain $ S.takeWhile (<2) stream +Effect! + +-} drain :: Monad m => Stream (Of a) m r -> m r drain = loop where loop stream = case stream of @@ -269,7 +323,14 @@ -- dropWhile -- --------------- --- | Ignore elements of a stream until a test succeeds. +{- | 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 @@ -286,10 +347,19 @@ {- | Stream the elements of a foldable container. ->>> S.print $ S.each [1..3] +>>> S.print $ each [1..3] >> yield 4 +0 1 2 3 +4 + +> 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 ()) @@ -299,24 +369,27 @@ -- enumFrom -- ------ -enumFrom :: (Monad m, Num n) => n -> Stream (Of n) m () +enumFrom :: (Monad m, Enum n) => n -> Stream (Of n) m r enumFrom = loop where - loop !n = Step (n :> loop (n+1)) + loop !n = Step (n :> loop (succ n)) {-# INLINEABLE enumFrom #-} - -enumFromTo :: (Monad m, Num n, Ord n) => n -> n -> Stream (Of n) m () -enumFromTo = loop where - loop !n m = if n <= m - then Step (n :> loop (n+1) m) - else Return () -{-# INLINEABLE enumFromTo #-} +-- +-- enumFromTo :: (Monad m, Num n, Ord n) => n -> n -> Stream (Of n) m () +-- enumFromTo = loop where +-- loop !n m = if n <= m +-- then Step (n :> loop (n+1) m) +-- else Return () +-- {-# INLINEABLE enumFromTo #-} +-- enumFromThen x y = map toEnum [fromEnum x, fromEnum y ..] -enumFromStepN :: (Monad m, Num a) => a -> a -> Int -> Stream (Of a) m () -enumFromStepN start step = loop start where - loop !s m = case m of - 0 -> Return () - _ -> Step (s :> loop (s+step) (m-1)) -{-# INLINEABLE enumFromStepN #-} +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 @@ -547,7 +620,16 @@ return (Step (a :> loop (f a))) {-# INLINEABLE iterateM #-} + -- --------------- +-- length +-- --------------- +length :: Monad m => Stream (Of a) m () -> m Int +length = fold (\n _ -> n + 1) 0 id + +length' :: Monad m => Stream (Of a) m r -> m (Of Int r) +length' = fold' (\n _ -> n + 1) 0 id +-- --------------- -- map -- --------------- @@ -566,7 +648,7 @@ {-| For each element of a stream, stream a foldable container of elements instead ->>> D.print $ D.mapFoldable show $ D.yield 12 +>>> S.print $ S.mapFoldable show $ yield 12 '1' '2' @@ -694,11 +776,10 @@ {-| Repeat a monadic action /ad inf./, streaming its results. ->>> L.purely fold L.list $ S.take 2 $ repeatM getLine -hello -world +>>> S.toListM $ S.take 2 (repeatM getLine) +hello +world ["hello","world"] - -} repeatM :: Monad m => m a -> Stream (Of a) m r @@ -719,7 +800,13 @@ loop m = Step (a :> loop (m-1)) {-# INLINEABLE replicate #-} --- | Repeat an action several times, streaming the results. +{-| 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 () @@ -969,13 +1056,27 @@ {-# INLINE toListM' #-} {-| Build a @Stream@ by unfolding steps starting from a seed. - This is one natural way to consume a 'Pipes.Producer'. It is worth - adding it to the functor-general 'unfold' to avoid dealing with - the left-strict pairing we are using in place of Haskell pairing. -> unfoldr Pipes.next :: Monad m => Producer a m r -> Stream (Of a) m r -> unfold (curry (:>) . Pipes.next) :: Monad m => Producer a m r -> Stream (Of a) m r + 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.drain $ 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 @@ -983,7 +1084,7 @@ loop s0 = Delay $ do e <- step s0 case e of - Left r -> return (Return r) + Left r -> return (Return r) Right (a,s) -> return (Step (a :> loop s)) {-# INLINABLE unfoldr #-} @@ -993,26 +1094,26 @@ {-| A singleton stream ->>> S.sum $ do {yield 1; lift (putStrLn "hello"); yield 2; lift (putStrLn "goodbye"); S.yield 3} +>>> stdoutLn $ yield "hello" hello -goodbye -6 ->>> S.sum $ S.take 3 $ forever $ do {lift (putStrLn "enter a number") ; n <- lift readLn; S.yield n } -enter a number -100 -enter a number -200 -enter a number -300 -600 - -enter a number -1 -enter a number -1000 -1001 +>>> 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 #-} @@ -1048,11 +1149,26 @@ {-| repeatedly stream lines as 'String' from stdin ->>> S.stdoutLn $ S.show (S.each [1..3]) +>>> 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 @@ -1060,10 +1176,10 @@ {-| Read values from 'IO.stdin', ignoring failed parses ->>> S.sum $ S.take 2 $ forever S.readLn :: IO Int -3 -#$%^&\^? -1000 +>>> S.sum $ S.take 2 S.readLn :: IO Int +3 +#$%^&\^? +1000 1003 -} @@ -1076,6 +1192,12 @@ {-| 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 @@ -1140,7 +1262,25 @@ {-| Write 'String's to 'IO.stdout' using 'putStrLn' This does not handle a broken output pipe, but has a polymorphic return - value + value, which makes this possible: + +>>> rest <- stdoutLn' $ S.splitAt 3 $ S.show (each [1..5]) +1 +2 +3 +>>> stdoutLn' rest +4 +5 + + Or indeed: + +>>> 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 diff --git a/streaming.cabal b/streaming.cabal --- a/streaming.cabal +++ b/streaming.cabal @@ -1,5 +1,5 @@ name: streaming -version: 0.1.0.6 +version: 0.1.0.7 cabal-version: >=1.10 build-type: Simple synopsis: A free monad transformer optimized for streaming applications. @@ -11,8 +11,8 @@ . @Streaming.Prelude@ closely follows @Pipes.Prelude@, but cleverly /omits the pipes/. It is focused - on employment with a base functors which generate - effectful sequences: e.g. + on employment with base functors which generate + effectful sequences. These appear elsewhere under titles like . > pipes: Producer a m r, Producer a m (Producer a m r), FreeT (Producer a m) m r > io-streams: InputStream a, Generator a r @@ -36,16 +36,20 @@ > Streaming.reread IOStreams.read :: InputStream a -> Stream (Of a) IO () > IOStreams.unfoldM Streaming.uncons :: Stream (Of a) IO () -> IO (InputStream a) . - The purposes of the separate @Generator a r@ type can as well be met with - @Stream (Of a) m r@, which admits more complex manipulations and should - be somewhat friendlier to the compiler. + The separate @Generator a r@ type in @io-streams@ is intended to permit + construction of an @InputStream@ with more possibilities + (such as the @yield@ statement). This purpose can as well be met with + @Stream (Of a) m r@, which may be friendlier to the compiler. . A simple exit to would be, e.g.: . > Conduit.unfoldM Streaming.uncons :: Stream (Of a) m () -> Source m a . These conversions should never be more expensive than a single @>->@ or @=$=@. + . + + license: BSD3 license-file: LICENSE author: michaelt @@ -75,7 +79,6 @@ , mtl >=2.1 && <2.3 , mmorph >=1.0 && <1.2 , transformers >=0.3 && <0.5 - , ghc-prim default-language: Haskell2010