streaming 0.1.0.7 → 0.1.0.8
raw patch · 4 files changed
+201/−41 lines, 4 files
Files
- Streaming.hs +34/−19
- Streaming/Internal.hs +131/−6
- Streaming/Prelude.hs +34/−6
- streaming.cabal +2/−10
Streaming.hs view
@@ -8,9 +8,13 @@ unfold, construct, for,+ layer,+ layers, replicates, repeats, repeatsM,+ wrap,+ step, -- * Transforming streams maps,@@ -20,13 +24,19 @@ -- * Inspecting a stream inspect, + -- * Zipping streams+ zips,+ zipsWith,+ interleaves,+ -- * Eliminating a 'Stream' intercalates, concats, iterTM, iterT, destroy,- + mapsM_,+ runEffect, -- * Splitting and joining 'Stream's splitsAt,@@ -47,7 +57,9 @@ join, liftA2, liftA3,- void+ void,+ (&),+ (-->) ) where import Streaming.Internal @@ -57,37 +69,40 @@ import Control.Applicative import Control.Monad.Trans import Data.Functor.Compose +import Data.Function ((&))+infixl 6 -->+(-->) = flip (.) {- $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. + specified by the first (functor) parameter. The (hidden) implementation is > data Stream f m r = Step !(f (Stream f m r)) | Delay (m (Stream f m r)) | Return r 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. a @yield@ statement. In 'Streaming.Prelude', @(a,b)@ is- replaced by the left-strict pair @Of a b@. Various operations are- defined for types like+ 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)@ + In it, various operations are defined for fundamental streaming types like -> Stream (Of a) m r -- a producer in the pipes sense -> -- i.e. an effectful, streaming [a], or rather ([a],r) +> Stream (Of a) m r -- a generator or producer (in the pipes sense) +> -- compare [a], or rather ([a],r) > Stream (Of a) m (Stream (Of a) m r) -- the effectful splitting of a producer-> -- i.e. an effectful ([a],[a]) or rather ([a],([a],r))-> Stream (Stream (Of a) m) m r -- successive, segmentation of a producer-> -- i.e. [[a]], or ([a],([a],([a]... r)))-- and so on. But of course any functor can be used. So, for example, +> -- compare ([a],[a]) or rather ([a],([a],r))+> Stream (Stream (Of a) m) m r -- segmentation of a producer+> -- cp. [[a]], or rather ([a],([a],([a],(...,r)))) -> Stream ((->) input) m result+ and so on. But of course any functor can be used, and this is part of + the point of this prelude - as we already see from + the type of the segmented stream, @Stream (Stream (Of a) m) m r@ - is a simple @Consumer input m result@ or @Parser input m result@ type. And so on.- See e.g. http://www.haskellforall.com/2012/07/purify-code-using-free-monads.html ,- http://www.haskellforall.com/2012/07/free-monad-transformers.html and similar- literature.+and operations like e.g. +> 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 To avoid breaking reasoning principles, the constructors should not be used directly. A pattern-match should go by way of 'inspect'
Streaming/Internal.hs view
@@ -1,5 +1,6 @@ {-# LANGUAGE RankNTypes, StandaloneDeriving,DeriveDataTypeable, BangPatterns #-} {-# LANGUAGE UndecidableInstances #-} -- for show, data instances+{-#LANGUAGE MultiParamTypeClasses, FunctionalDependencies #-} module Streaming.Internal ( -- * The free monad transformer -- $stream@@ -11,27 +12,38 @@ , replicates , repeats , repeatsM+ , wrap+ , step+ , layer -- * Eliminating a stream- , destroy - , concats , intercalates + , concats , iterT , iterTM -+ , destroy + , destroyWith+ -- * Inspecting a stream step by step , inspect -- * Transforming streams , maps , mapsM + , mapsM_+ , runEffect , distribute -- * Splitting streams , chunksOf , splitsAt - -- * For internal use+ -- * Zipping streams+ , zipsWith+ , zips+ , interleaves+ + -- * For use in implementation , unexposed , hoistExposed , mapsExposed@@ -52,7 +64,7 @@ import GHC.Exts ( build ) import Data.Data ( Data, Typeable ) import Prelude hiding (splitAt)-+import Data.Functor.Compose {- $stream The 'Stream' data type is equivalent to @FreeT@ and can represent any effectful@@ -128,7 +140,6 @@ Step f -> Step (fmap loop f) {-# INLINABLE hoist #-} - instance Functor f => MMonad (Stream f) where embed phi = loop where loop stream = case stream of@@ -167,6 +178,37 @@ {-# INLINABLE destroy #-} +{-| 'destroyWith' 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.++>>> :t destroyWith join return+(Monad m, Functor f) => + (f (m a) -> m a) -> Stream f m a -> m a -- iterT+>>> :t destroyWith (join . lift) return+(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 wrap 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 wrap return (step . lazily)+Monad m => + Stream (Of a) m r -> Stream ((,) a) m r+>>> :t destroyWith wrap return (step . strictly)+Monad m => + Stream ((,) a) m r -> Stream (Of a) m r+>>> :t destroyWith Data.ByteString.Streaming.wrap return +(Monad m, Functor f) =>+ (f (ByteString m r) -> ByteString m r) -> Stream f m r -> ByteString m r+>>> :t destroyWith Data.ByteString.Streaming.wrap return (\(a:>b) -> consChunk a b) +Monad m => + Stream (Of B.ByteString) m r -> ByteString m r -- fromChunks+-}+destroyWith+ :: (Functor f, Monad m) =>+ (m b -> b) -> (r -> b) -> (f b -> b) -> Stream f m r -> b+destroyWith wrap done construct stream = destroy stream construct wrap done+ -- | 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@@ -230,6 +272,30 @@ {-# INLINABLE mapsM #-} +{-| Run the effects in a stream that merely layers effects.+-}+runEffect :: Monad m => Stream m m r -> m r+runEffect = loop where+ loop stream = case stream of+ Return r -> return r+ Delay m -> m >>= loop+ Step mrest -> mrest >>= loop+{-# INLINABLE runEffect #-}+++{-| Map each layer to an effect in the base monad, and run them all.+-}+mapsM_ :: (Functor f, Monad m) => (forall x . f x -> m x) -> Stream f m r -> m r+mapsM_ f str = runEffect (maps f str)+{-# INLINABLE mapsM_ #-}+++{-| Lift for items in the base functor. Makes a singleton or+ one-layer succession.`+-}+layer :: (Monad m, Functor f) => f r -> Stream f m r+layer fr = Step (fmap Return fr)+ {-| Interpolate a layer at each segment. This specializes to e.g. > intercalates :: (Monad m, Functor f) => Stream f m () -> Stream (Stream f m) m r -> Stream f m r@@ -420,6 +486,7 @@ 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.@@ -438,6 +505,12 @@ Step fs -> construct (fmap loop fs) {-# INLINABLE destroyExposed #-} ++{-| This is akin to the @observe@ of @Pipes.Internal@ . It rewraps the layering+ in instances of @Stream f m r@ so that it replicates that of + @FreeT@. ++-} unexposed :: (Functor f, Monad m) => Stream f m r -> Stream f m r unexposed = Delay . loop where loop stream = case stream of @@ -445,3 +518,55 @@ Delay m -> m >>= loop Step f -> return (Step (fmap (Delay . loop) f)) {-# INLINABLE unexposed #-} ++++wrap :: (Monad m, Functor f ) => m (Stream f m r) -> Stream f m r+wrap = Delay++step :: (Monad m, Functor f ) => f (Stream f m r) -> Stream f m r+step = Step+++zipsWith :: (Monad m, Functor h) + => (forall x y . f x -> g y -> h (x,y)) + -> Stream f m r -> Stream g m r -> Stream h m r+zipsWith phi = curry loop where+ loop (s1, s2) = Delay $ go s1 s2+ go (Return r) p = return $ Return r+ go q (Return s) = return $ Return s+ go (Delay m) p = m >>= \s -> go s p+ go q (Delay m) = m >>= go q+ go (Step f) (Step g) = return $ Step $ fmap loop (phi f g)+{-# INLINABLE zipsWith #-} + +zips :: (Monad m, Functor f, Functor g) + => Stream f m r -> Stream g m r -> Stream (Compose f g) m r +zips = zipsWith go where+ go fx gy = Compose (fmap (\x -> fmap (\y -> (x,y)) gy) fx)+{-# INLINE zips #-} +++{-| Interleave functor layers, with the effects of the first preceding+ the effects of the second.++> interleaves = zipsWith (liftA2 (,))++>>> let paste = \a b -> interleaves (Q.lines a) (maps (Q.cons' '\t') (Q.lines b))+>>> Q.stdout $ Q.unlines $ paste "hello\nworld\n" "goodbye\nworld\n"+hello goodbye+world world++-}+ +interleaves+ :: (Monad m, Applicative h) =>+ Stream h m r -> Stream h m r -> Stream h m r+interleaves = zipsWith (liftA2 (,))+{-# INLINE interleaves #-} ++++ + +
Streaming/Prelude.hs view
@@ -34,11 +34,14 @@ Of (..) , lazily , strictly+ , fst'+ , snd' -- * Introducing streams of elements -- $producers , yield , each+ , layers , unfoldr , stdinLn , readLn@@ -174,6 +177,11 @@ 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 {-| Break a sequence when a element falls under a predicate, keeping the rest of the stream as the return value. @@ -347,14 +355,14 @@ {- | Stream the elements of a foldable container. ->>> S.print $ each [1..3] >> yield 4+>>> S.print $ S.map (*100) $ each [1..3] >> yield 4 0-1-2-3-4+100+200+300+400 -> S.print $ S.map (*100) $ each [1..3] >> lift readLn >>= yield+>>> S.print $ S.map (*100) $ each [1..3] >> lift readLn >>= yield 100 200 300@@ -621,6 +629,11 @@ {-# INLINEABLE iterateM #-} +layers+ :: (Monad m, Functor f) =>+ Stream (Of a) m r -> (a -> f x) -> Stream f m r+layers stream f = for stream $ layer . f+{-# INLINE layers #-} -- --------------- -- length -- ---------------@@ -1077,6 +1090,18 @@ hello goodbye<Enter> 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@@ -1096,6 +1121,9 @@ >>> 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
streaming.cabal view
@@ -1,5 +1,5 @@ name: streaming-version: 0.1.0.7+version: 0.1.0.8 cabal-version: >=1.10 build-type: Simple synopsis: A free monad transformer optimized for streaming applications.@@ -27,8 +27,6 @@ > 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 .- (If you don't have @pipes-HEAD@, inline the definition of <https://github.com/Gabriel439/Haskell-Pipes-Library/blob/master/src/Pipes/Prelude.hs#L909 unfoldr>.) - . Interoperation with <http://hackage.haskell.org/package/io-streams io-streams> is thus:@@ -36,19 +34,13 @@ > Streaming.reread IOStreams.read :: InputStream a -> Stream (Of a) IO () > IOStreams.unfoldM Streaming.uncons :: Stream (Of a) IO () -> IO (InputStream a) .- 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