diff --git a/Streaming.hs b/Streaming.hs --- a/Streaming.hs +++ b/Streaming.hs @@ -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' diff --git a/Streaming/Internal.hs b/Streaming/Internal.hs --- a/Streaming/Internal.hs +++ b/Streaming/Internal.hs @@ -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 #-} + + + + + + diff --git a/Streaming/Prelude.hs b/Streaming/Prelude.hs --- a/Streaming/Prelude.hs +++ b/Streaming/Prelude.hs @@ -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 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 diff --git a/streaming.cabal b/streaming.cabal --- a/streaming.cabal +++ b/streaming.cabal @@ -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 .) - . Interoperation with 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 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