streaming 0.1.0.6 → 0.1.0.7
raw patch · 4 files changed
+351/−118 lines, 4 filesdep −ghc-prim
Dependencies removed: ghc-prim
Files
- Streaming.hs +28/−8
- Streaming/Internal.hs +87/−17
- Streaming/Prelude.hs +226/−86
- streaming.cabal +10/−7
Streaming.hs view
@@ -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+-}
Streaming/Internal.hs view
@@ -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 #-}
Streaming/Prelude.hs view
@@ -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<Enter>+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<Enter>+world<Enter> ["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<Enter>+hello+goodbye<Enter>+goodbye++>>> S.stdoutLn $ S.unfoldr P.next (P.stdinLn P.>-> P.take 2)+hello<Enter>+hello+goodbye<Enter>+goodbye++>>> S.drain $ S.unfoldr P.next (P.stdinLn P.>-> P.take 2 P.>-> P.stdoutLn)+hello<Enter>+hello+goodbye<Enter>+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>+Enter a number:+20<Enter>+Enter a number:+100<Enter>+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<Enter>+hello+world<Enter>+world+^CInterrupted.+++>>> stdoutLn $ S.map reverse stdinLn +hello<Enter>+olleh+world<Enter>+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<Enter>+#$%^&\^?<Enter>+1000<Enter> 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
streaming.cabal view
@@ -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 <http://hackage.haskell.org/package/conduit conduit> 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