streaming 0.1.1.1 → 0.1.2.0
raw patch · 5 files changed
+474/−217 lines, 5 filesdep −random
Dependencies removed: random
Files
- README.md +73/−10
- Streaming.hs +5/−2
- Streaming/Internal.hs +201/−50
- Streaming/Prelude.hs +194/−153
- streaming.cabal +1/−2
README.md view
@@ -3,18 +3,82 @@ -`Stream` can be used wherever [FreeT](https://hackage.haskell.org/package/free-4.12.1/docs/Control-Monad-Trans-Free.html) is used. The compiler's standard range of optimizations work better for operations written in terms of `Stream`. `FreeT f m r` and `Stream f m r` are of course extremely general, and many functor-general combinators are exported by the general module `Streaming`.+`Stream` can be used wherever [FreeT](https://hackage.haskell.org/package/free-4.12.1/docs/Control-Monad-Trans-Free.html) is used. The compiler's standard range of optimizations work better for operations written in terms of `Stream`. `FreeT f m r` and `Stream f m r` are of course extremely general, and many functor-general combinators are exported by the general module `Streaming`. +But the library is focused on uses of `Stream f m r` where `f` is itself in some sense "streaming". This means, very crudely, that it is possible to make strict left folds over it. Where `f` is complex, and has the form `t m`, these folds will have the form `t m r -> m (a, r)`, polymorphic in `r`. In particular, it will be possible, for example, to write the trivial left fold - a `drain` or `runEffects` function - `t m r -> m r` or `f r -> m r` - polymorphically. `Stream f m r` preserves this property. In particular, branching and failure are excluded; the latter is always handled in the monad `m`. ++The abstraction is inevitable, though there are many ways of writing it. Once one possesses it, though, one is already in possession of an elementary streaming library, since `Stream ((,)a) m r` or its equivalent is the type of a producer, generator or source. I try to argue for this more elaborately below, bringing it into connection with the standard streaming io libraries. + The freely generated stream on a streamable functor ---------------------------------------------------- As soon as you consider the idea of an effectful stream of any kind whatsoever, for example, a stream of bytes from a handle, however constituted, you will inevitably be forced to contemplate the idea of a streaming *succession* of *just such streams*. Thus, for example, however you imagine your bytes streaming from a handle, you will want to consider a *succession* of *such streams* divided on newlines. -Similarly, suppose you have the idea the unfolding of some sort of stream from a Haskell value, a seed - a file name, as it might be. And suppose you *also* have some idea of a stream of individual Haskell values - maybe a stream of file names coming from something like `du`, subjected to some filter. Then you will also have the idea of a streaming *succession* of *such unfoldings* linked together end to end in accordance with the initial succession of seed values.+This is closely related to the fact that, as soon as you contemplate a complex streaming phenomenon, you will want to consider a break in the stream, a function that divides the stream into parts according to some internal characteristic, and allows us to handle the parts separately, making it possible to do one thing with the first part and another with the second. Such a function will not have the form: -Call the thoughts in that paragraph the ABCs of streaming. If you understood these ABCs you have a total comprehension of `Stream f m r`:+ splitter :: S -> (S, S) +like the splitting operations we find with lists and the like, e.g. ++ splitAt 3 :: [a] -> ([a],[a])++Since we can assume an underlying monad m, which may be implicit (in `io-streams`, for example, `IO` is implicit in the types of `InputStream` and `Generator`), we can write the candidate type thus:++ splitter :: S m -> (S m, S m)++These types use ordinary "pure" pairing, and cannot express the fundamental point that I cannot get to the 'second' stream without passing through the 'first'; the features of the 'second half' may depend causally on events in the first half. We do not repair this, but just make it worse, by complicating the type thus++ splitter :: S m -> m (S m, S m)+ ++since the effects I must pass through to get to pair, and thus the second element, are precisely the effects putatively contained in the first element in the result type. My idea was to do "one thing with the first half" and "another thing with the second half"; in this type I somehow do the effects of the first half to get the pair, and still have the first half before me, coupled with the second half. If I am not proposing to repeat the action of the first part and I have not lost information, my type must secretly be something like ++ splitter :: S m -> m (S Identity, S m)+ +or++ splitAccum :: S m -> m ([z], S m)+ +as we see, e.g. [here](http://hackage.haskell.org/package/list-t-0.4.5.1/docs/src/ListT.html#splitAt) or, more obscurely in functions like [these](http://hackage.haskell.org/package/conduit-combinators-1.0.3/docs/src/Data-Conduit-Combinators.html#splitOnUnboundedE). (I will return to this difficulty below.)++This point makes it inevitable that *a rational stream type will have a return value*. It will have the form ++ S r+ +or ++ S m r+ +and the dividing functions will have the form+++ splitter :: S r -> S (S r) ++or, where the underlying form of effect is explicit++ splitter :: S m r -> S m (S m r)+ +Now we can express what we meant by 'doing one thing with the first half and another with the second', we were thinking of applying some sort of polymorphic folds, maybe with types like++ folder :: S m x -> m (a, x)+ +then we have++ folder . splitter :: S m r -> m (a, S m r)+ +and can contemplate applying this or another folding operation to the 'second half', e.g.++ liftM (fmap folder) . folder . splitter :: S m x -> m (a, m (a,x))+ +and can reshuffle to get a function `S m x -> m ((a,a), x)`. This function has the form of our original folder function, since it is polymorphic in `x`. ++That folds over streaming types should be polymorphic in their return type is written already into this simple material: we want to 'do one thing with the first half and something else - or the same thing - with the second half'. The thing we 'do with the first half' will have to be something we could do even if the second half doesn't exist, and it must preserve it if it does. In the simplest case, 'what we do with the first half' might be simply to throw it out, or drain it. ++Now, to return to the first point, suppose you have the idea the unfolding of some sort of stream from an individual Haskell value, a seed - a file name, as it might be. And suppose you *also* have some idea of a stream *of* individual Haskell values - maybe a stream of file names coming from something like `du`, subjected to some filter. Then you will also have the idea of a streaming *succession* of *such unfoldings* linked together end to end in accordance with the initial succession of seed values.++Call the thoughts above the ABCs of streaming. If you understood these ABCs you have a total comprehension of `Stream f m r`:+ - `Stream` expresses what the word "succession" meant in the ABCs - The general parameter `f` expresses what was meant by "such streams" - `m` expresses the relevant form of "effect".@@ -39,7 +103,7 @@ conduit: ConduitM () o m r streaming: Stream (Of a) m r -The only difference is that in `streaming` the simple Generator or Producer concept is formulated explicitly in terms of the *general* concept of successive connection. But this is a concept you need and already possess anyway, as your comprehension of the four sentences above showed.+The only difference is that in `streaming` the simple generator or producer concept is formulated explicitly in terms of the *general* concept of successive connection. But *this is a concept you need and already possess anyway*, as your comprehension of the streaming ABCs showed. The special case of a *stream of individual Haskell values* that simply *comes to an end without a special result* is variously expressed thus: @@ -135,7 +199,7 @@ accumulate . Streaming.Prelude.replicateM :: Int -> m a -> m (Stream (Of a) Identity ()) -which is what we find in our diseased base libraries. But once you know how to operate with a stream directly you will see less and less point in what is called *extracting the (structured) value from IO*. The distinction between+which is what we find in our diseased base libraries. But once you know how to operate with a stream directly you will see less and less point in what is called *extracting the (structured) value from IO*. Consider the apparently innocent distinction between "getContents" :: String @@ -143,13 +207,13 @@ getContents :: IO String -but, omitting consideration of eof, we might define `getContents` thus+Omitting consideration of eof, we might define `getContents` thus getContents = sequence $ repeat getChar There it is again! The very devil! By contrast there is no distinction between - "getContents" :: Stream (Of Char) m ()+ "getContents" :: Stream (Of Char) m () -- the IsString instance is monad-general and @@ -167,10 +231,9 @@ splitAt 20 $ "getLine" >> getLine :: String IO (String IO ()) length $ "getLine" >> getLine :: IO Int -and can dispense with half the advice they will give you on `#haskell`. It is only a slight exaggeration to say that a stream should never be "extracted from IO".+and can dispense with half the advice they will give you on `#haskell`. It is only a slight exaggeration to say that a stream should never be "extracted from IO". -With `sequence` and `traverse`, we accumulate a pure succession of pure values from a pure succession of monadic values.\-Why bother if you have intrinsically monadic conception of succession or traversal? `Stream f m r` gives you an immense body of such structures and a simple discipline for working with them. Spinkle `id` freely though your program if you get homesick.+With `sequence` and `traverse`, we accumulate a pure succession of pure values from a pure succession of monadic values. Why bother if you have intrinsically monadic conception of succession or traversal? `Stream f m r` gives you an immense body of such structures and a simple discipline for working with them. Spinkle `id` freely though your program, under various names, if you get homesick for `sequence` and company. Interoperation with the streaming-io libraries ----------------------------------------------
Streaming.hs view
@@ -8,7 +8,7 @@ unfold, construct, for,- elevate,+ yields, replicates, repeats, repeatsM,@@ -20,7 +20,9 @@ maps, mapsM, distribute,- eithers,+ separate,+ unseparate, + groups, -- * Inspecting a stream inspect,@@ -48,6 +50,7 @@ Of (..), lazily, strictly,+ -- * re-exports MFunctor(..),
Streaming/Internal.hs view
@@ -13,7 +13,7 @@ , repeatsM , mwrap , wrap- , elevate+ , yields -- * Eliminating a stream , intercalates @@ -21,7 +21,6 @@ , iterT , iterTM , destroy - , destroyWith -- * Inspecting a stream wrap by wrap , inspect @@ -31,9 +30,12 @@ , mapsM , decompose , mapsM_- , eithers , run , distribute+ , separate+ , unseparate+ , groups+-- , groupInL -- * Splitting streams , chunksOf @@ -45,6 +47,10 @@ , zips , interleaves + -- * Assorted Data.Functor.x help+ + , switch+ -- * For use in implementation , unexposed , hoistExposed@@ -165,19 +171,19 @@ {-# INLINE pure #-} streamf <*> streamx = do {f <- streamf; x <- streamx; return (f x)} {-# INLINABLE (<*>) #-} - stra0 *> strb = loop stra0 where- loop stra = case stra of- Return _ -> strb- Delay m -> Delay (do {stra' <- m ; return (stra' *> strb)})- Step fstr -> Step (fmap (*> strb) fstr)- {-# INLINABLE (*>) #-} - stra <* strb0 = loop strb0 where- loop strb = case strb of- Return _ -> stra- Delay m -> Delay (do {strb' <- m ; return (stra <* strb')})- Step fstr -> Step (fmap (stra <*) fstr)- {-# INLINABLE (<*) #-} - + -- stra0 *> strb = loop stra0 where+ -- loop stra = case stra of+ -- Return _ -> strb+ -- Delay m -> Delay (do {stra' <- m ; return (stra' *> strb)})+ -- Step fstr -> Step (fmap (*> strb) fstr)+ -- {-# INLINABLE (*>) #-}+ -- stra <* strb0 = loop strb0 where+ -- loop strb = case strb of+ -- Return _ -> stra+ -- Delay m -> Delay (do {strb' <- m ; return (stra <* strb')})+ -- Step fstr -> Step (fmap (stra <*) fstr)+ -- {-# INLINABLE (<*) #-}+ -- instance Functor f => MonadTrans (Stream f) where lift = Delay . liftM Return {-# INLINE lift #-}@@ -220,10 +226,10 @@ destroy :: (Functor f, Monad m) => Stream f m r -> (f b -> b) -> (m b -> b) -> (r -> b) -> b-destroy stream0 construct mwrap done = loop (unexposed stream0) where+destroy stream0 construct effect done = loop (unexposed stream0) where loop stream = case stream of Return r -> done r- Delay m -> mwrap (liftM loop m)+ Delay m -> effect (liftM loop m) Step fs -> construct (fmap loop fs) {-# INLINABLE destroy #-} @@ -238,26 +244,26 @@ >>> :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 mwrap return+>>> :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 mwrap return (wrap . lazily)+>>> :t destroyWith effect return (wrap . lazily) Monad m => Stream (Of a) m r -> Stream ((,) a) m r->>> :t destroyWith mwrap return (wrap . strictly)+>>> :t destroyWith effect return (wrap . strictly) Monad m => Stream ((,) a) m r -> Stream (Of a) m r->>> :t destroyWith Data.ByteString.Streaming.mwrap return +>>> :t destroyWith Data.ByteString.Streaming.effect return (Monad m, Functor f) => (f (ByteString m r) -> ByteString m r) -> Stream f m r -> ByteString m r->>> :t destroyWith Data.ByteString.Streaming.mwrap return (\(a:>b) -> consChunk a b) +>>> :t destroyWith Data.ByteString.Streaming.effect 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 mwrap done construct stream = destroy stream construct mwrap done+destroyWith effect done construct stream = destroy stream construct effect done -- | Reflect a church-encoded stream; cp. @GHC.Exts.build@ construct@@ -312,6 +318,28 @@ Step f -> Step (phi (fmap loop f)) {-# 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+-- Delay m -> Delay (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:@@ -348,7 +376,7 @@ {-| Run the effects in a stream that merely layers effects. -}-run :: Monad m => Stream m m r -> m r+run :: Monad m => Stream m m r -> m r run = loop where loop stream = case stream of Return r -> return r@@ -364,16 +392,6 @@ {-# INLINABLE mapsM_ #-} -{-| Lift for items in the base functor. Makes a singleton or- one-layer succession. It is named by similarity to lift: --> lift :: (Monad m, Functor f) => m r -> Stream f m r-> elevate :: (Monad m, Functor f) => f r -> Stream f m r--}--elevate :: (Monad m, Functor f) => f r -> Stream f m r-elevate 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@@ -405,7 +423,7 @@ (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+iterTM out stream = destroyExposed stream out (join . lift) return {-# INLINE iterTM #-} {-| Specialized fold@@ -414,7 +432,7 @@ -} iterT :: (Functor f, Monad m) => (f (m a) -> m a) -> Stream f m a -> m a-iterT out stream = destroy stream out join return+iterT out stream = destroyExposed stream out join return {-# INLINE iterT #-} {-| Dissolves the segmentation into layers of @Stream f m@ layers.@@ -557,15 +575,15 @@ -- See Atkey "Reasoning about Stream Processing with Effects" -destroyExposed stream0 construct mwrap done = loop stream0 where+destroyExposed stream0 construct effect done = loop stream0 where loop stream = case stream of Return r -> done r- Delay m -> mwrap (liftM loop m)+ Delay m -> effect (liftM loop m) Step fs -> construct (fmap loop fs) {-# INLINABLE destroyExposed #-} -{-| This is akin to the @observe@ of @Pipes.Internal@ . It remwraps the layering+{-| This is akin to the @observe@ of @Pipes.Internal@ . It reeffects the layering in instances of @Stream f m r@ so that it replicates that of @FreeT@. @@ -587,6 +605,17 @@ wrap = Step +{-| Lift for items in the base functor. Makes a singleton or+ one-layer succession. It is named by similarity to lift: ++> lift :: (Monad m, Functor f) => m r -> Stream f m r+> yields :: (Monad m, Functor f) => f r -> Stream f m r+-}++yields :: (Monad m, Functor f) => f r -> Stream f m r+yields fr = Step (fmap Return fr)++ 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@@ -625,16 +654,138 @@ {-# INLINE interleaves #-} -eithers :: (Monad m, Applicative h) => - (forall x . f x -> h x) -> (forall x . g x -> h x) -> Stream (Sum f g) m r -> Stream h m r-eithers f g = loop where- loop str = case str of - Return r -> Return r- Delay m -> Delay (liftM loop m)- Step str' -> case str' of - InL s -> Step (fmap loop (f s))- InR t -> Step (fmap loop (g t))- +{-| Swap the order of functors in a sum of functors. +++>>> S.toListM' $ S.print $ separate $ maps S.switch $ maps (S.distinguish (=='a')) $ S.each "banana"+'a'+'a'+'a'+"bnn" :> ()+>>> S.toListM' $ S.print $ separate $ maps (S.distinguish (=='a')) $ S.each "banana"+'b'+'n'+'n'+"aaa" :> ()+-}+switch :: Sum f g r -> Sum g f r+switch s = case s of InL a -> InR a; InR a -> InL a+{-#INLINE switch #-}++ +{-| Given a stream on a sum of functors, make it a stream on the left functor,+ 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++ Now, for example, it is convenient to fold on the left and right values separately:++>>> toListM' $ 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++-}++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) + return +{-#INLINE separate #-}++unseparate :: (Monad m, Functor f, Functor g) => Stream f (Stream g m) r -> Stream (Sum f g) m r+unseparate str = destroyExposed + str + (wrap . InL) + (join . maps InR) + return +{-#INLINE unseparate #-}+ +{-| 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 + -> Stream (Sum (Stream f m) (Stream g m)) m r+groups = loop + where+ loop str = do+ e <- lift $ inspect str+ case e of+ Left r -> return r+ Right ostr -> case ostr of+ InR gstr -> wrap $ InR (fmap loop (cleanR (wrap (InR gstr))))+ InL fstr -> wrap $ InL (fmap loop (cleanL (wrap (InL fstr))))++ cleanL :: (Monad m, Functor f, Functor g) =>+ Stream (Sum f g) m r -> Stream f m (Stream (Sum f g) m r)+ cleanL = loop where+ loop s = do+ e <- lift $ inspect s+ case e of+ Left r -> return (return r)+ Right (InL fstr) -> wrap (fmap loop fstr)+ Right (InR gstr) -> return (wrap (InR gstr))++ cleanR :: (Monad m, Functor f, Functor g) =>+ Stream (Sum f g) m r -> Stream g m (Stream (Sum f g) m r)+-- cleanR = fmap (maps switch) . cleanL . maps switch+ cleanR = loop where+ loop s = do+ e <- lift $ inspect s+ case e of+ Left r -> return (return r)+ Right (InL fstr) -> return (wrap (InL fstr))+ Right (InR gstr) -> wrap (fmap loop gstr)++-- groupInL :: (Monad m, Functor f, Functor g)+-- => Stream (Sum f g) m r+-- -> Stream (Sum (Stream f m) g) m r+-- groupInL = loop+-- where+-- loop str = do+-- e <- lift $ inspect str+-- case e of+-- Left r -> return r+-- Right ostr -> case ostr of+-- InR gstr -> wrap $ InR (fmap loop gstr)+-- InL fstr -> wrap $ InL (fmap loop (cleanL (wrap (InL fstr))))+-- cleanL :: (Monad m, Functor f, Functor g) =>+-- Stream (Sum f g) m r -> Stream f m (Stream (Sum f g) m r)+-- cleanL = loop where+-- loop s = dos+-- e <- lift $ inspect s+-- case e of+-- Left r -> return (return r)+-- Right (InL fstr) -> wrap (fmap loop fstr)+-- Right (InR gstr) -> return (wrap (InR gstr))
Streaming/Prelude.hs view
@@ -12,6 +12,9 @@ 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@@ -52,13 +55,12 @@ , fromHandle , iterate , repeat+ , replicate , cycle , repeatM , replicateM , enumFrom , enumFromThen- , randomRs- , randoms -- * Consuming streams of elements -- $consumers@@ -67,7 +69,7 @@ , mapM_ , print , toHandle- , drain+ , effects , drained -- * Stream transformers@@ -108,30 +110,37 @@ , span , group , groupBy+ , groupedBy , timed -- , split - -- * Pair manipulation- , lazily- , strictly- , fst'- , snd'++ -- * Sum and Compose manipulation + , distinguish + , switch+ , separate+ , unseparate+ , eitherToSum+ , sumToCompose+ , composeToSum+ -- * Folds -- $folds , fold- , fold'+ , fold_ , foldM- , foldM'+ , foldM_ , sum- , sum'+ , sum_ , product- , product'+ , product_ , length- , length'+ , length_ , toList- , toListM- , toListM'+ , toList_+ , mconcat+ , mconcat_ , foldrM , foldrT @@ -156,6 +165,12 @@ , zip , zipWith + -- * Pair manipulation+ , lazily+ , strictly+ , fst'+ , snd'+ -- * Interoperation , reread @@ -165,9 +180,10 @@ ) where import Streaming.Internal -import Control.Monad hiding (filterM, mapM, mapM_, foldM, replicateM, sequence)+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 (..), (<$))@@ -177,7 +193,7 @@ import Data.Traversable (Traversable) import qualified Data.Foldable as Foldable import Text.Read (readMaybe)-import Prelude hiding (map, mapM, mapM_, filter, drop, dropWhile, take, sum, product+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, seq, show, read@@ -187,11 +203,12 @@ import qualified System.IO as IO import Foreign.C.Error (Errno(Errno), ePIPE) import Control.Exception (throwIO, try)-import Data.Monoid (Monoid (..))+import Data.Monoid (Monoid (mappend, mempty)) import Data.String (IsString (..))-import qualified System.Random as R 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,@@ -203,8 +220,6 @@ {-#INLINE mempty #-} mappend (m :> w) (m' :> w') = mappend m m' :> mappend w w' {-#INLINE mappend #-}- mconcat = foldr mappend mempty- {-#INLINE mconcat #-} instance Functor (Of a) where fmap f (a :> x) = a :> f x@@ -233,6 +248,11 @@ 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 @@ -329,12 +349,15 @@ loop a' (step x a') rest {-# INLINABLE breakWhen #-} -{- Break during periods where the predicate is not satisfied. -->>> S.print $ mapsM S.toListM' $ breaks even $ S.each [2,2,1,1,2,2,2,1,1]-[1,1]-[1,1]+{- 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 =>@@ -357,14 +380,19 @@ >>> S.product (S.chain Prelude.print (S.each [2..4])) >>= Prelude.print 2 3-4-24+4f+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)- yield a+chain f = loop where + loop str = case str of + Return r -> return r+ Delay mn -> Delay (liftM loop mn)+ Step (a :> rest) -> Delay $ do+ f a+ return (Step (a :> loop rest)) {-# INLINE chain #-} {-| Make a stream of traversable containers into a stream of their separate elements.@@ -453,7 +481,7 @@ delay seconds = mapM go where go a = liftIO (threadDelay (truncate (seconds * 1000000))) >> return a -- ------------------ drain+-- effects -- --------------- {- | Reduce a stream, performing its actions but ignoring its elements. @@ -462,29 +490,29 @@ >>> let effect = lift (putStrLn "Effect!") >>> let stream = do {yield 1; effect; yield 2; effect; return (2^100)} ->>> S.drain stream+>>> S.effects stream Effect! Effect! 1267650600228229401496703205376 ->>> S.drain $ S.takeWhile (<2) stream+>>> S.effects $ S.takeWhile (<2) stream Effect! -}-drain :: Monad m => Stream (Of a) m r -> m r-drain = loop where+effects :: Monad m => Stream (Of a) m r -> m r+effects = loop where loop stream = case stream of Return r -> return r Delay m -> m >>= loop Step (_ :> rest) -> loop rest-{-#INLINABLE drain #-}+{-#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 . drain)+> drained = join . fmap (lift . effects) >>> let take' n = S.drained . S.splitAt n >>> S.print $ concats $ maps (take' 1) $ S.group $ S.each "wwwwarrrrr"@@ -495,7 +523,7 @@ -} 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 . drain)+drained = join . fmap (lift . effects) {-#INLINE drained #-} -- ---------------@@ -651,32 +679,33 @@ {- $folds Use these to fold the elements of a 'Stream'. ->>> S.fold (+) 0 id $ S.each [1..0]+>>> S.fold_ (+) 0 id $ S.each [1..0] 50 - The general folds 'fold', fold\'', 'foldM' and 'foldM\'' are arranged + The general folds 'fold', fold_', 'foldM' and 'foldM_' are arranged for use with 'Control.Foldl' ->>> L.purely fold L.sum $ each [1..10]+>>> L.purely fold_ L.sum $ each [1..10] 55->>> L.purely fold (liftA3 (,,) L.sum L.product L.list) $ each [1..10]+>>> 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 a single quote - (e.g. @fold'@, @sum'@ carry the stream's return value in a left-strict pair.- These are convenient for @mapsM@-ing over a @Stream (Stream (Of a) m) m r@, + 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 ()+> 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 sum' $ chunksOf 3 $ each [1..10]+>>> 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)+>>> 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])@@ -688,24 +717,24 @@ > 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 () -> m b-fold step begin done stream0 = loop stream0 begin+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 + loop !stream !x = case stream of Return r -> return (done x) Delay m -> m >>= \s -> loop s x Step (a :> rest) -> loop rest (step x a)-{-# INLINABLE fold #-}+{-# INLINABLE fold_ #-} {-| Strict fold of a 'Stream' of elements that preserves the return value. ->>> S.sum' $ each [1..10]+>>> S.sum $ each [1..10] 55 :> () ->>> (n :> rest) <- sum' $ S.splitAt 3 (each [1..10])+>>> (n :> rest) <- S.sum $ S.splitAt 3 (each [1..10]) >>> print n 6->>> (m :> rest') <- sum' $ S.splitAt 3 rest+>>> (m :> rest') <- S.sum $ S.splitAt 3 rest >>> print m 15 >>> S.print rest'@@ -715,38 +744,38 @@ 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)+> 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+> 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]+>>> 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+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) Delay m -> m >>= \s -> loop s x Step (a :> rest) -> loop rest (step x a)-{-# INLINABLE fold' #-}+{-# 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+foldM_ :: Monad m- => (x -> a -> m x) -> m x -> (x -> m b) -> Stream (Of a) m () -> m b-foldM step begin done s0 = do+ => (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@@ -756,16 +785,16 @@ Step (a :> rest) -> do x' <- step x a loop rest x'-{-# INLINABLE foldM #-}+{-# 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'+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+foldM step begin done str = do x0 <- begin loop str x0 where@@ -775,8 +804,10 @@ Step (a :> rest) -> do x' <- step x a loop rest x'-{-# INLINABLE foldM' #-}+{-# 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'@@ -824,7 +855,35 @@ 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 = Delay $ 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)+ Delay m -> Delay $ 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 @@ -837,6 +896,8 @@ 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 (==) @@ -870,17 +931,13 @@ 10 -}-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 Int+length_ = fold_ (\n _ -> n + 1) 0 id+{-#INLINE length_#-} -{-| Run a stream, keeping its length and return value. As with all folds- this permits more complex mappings.+{-| Run a stream, keeping its length and its return value. ->>> S.length' $ S.each [1..10]-10 :> ()->>> fmap S.fst' $ S.length' $ S.each [1..10]-10->>> S.print $ mapsM S.length' $ chunksOf 3 $ S.each [1..10]+>>> S.print $ mapsM S.length $ chunksOf 3 $ S.each [1..10] 3 3 3@@ -888,8 +945,9 @@ -} -length' :: Monad m => Stream (Of a) m r -> m (Of Int r)-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+{-#INLINE length #-} -- --------------- -- map -- ---------------@@ -925,13 +983,14 @@ mapM f = loop where loop str = case str of Return r -> Return r - Delay m -> Delay $ liftM loop m+ Delay m -> Delay (liftM loop m) Step (a :> as) -> Delay $ do a' <- f a - return $ Step (a' :> loop as) + 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@@ -952,6 +1011,13 @@ {-# 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. @@ -999,52 +1065,20 @@ -- | 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 #-}+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' #-}+product :: (Monad m, Num a) => Stream (Of a) m r -> m (Of a r)+product = fold (*) 1 id+{-# INLINE product #-} -- ------------------ random--- -----------------{-| A crude infinite stream of random items, using @System.Random@--> randoms = liftIO Random.newStdGen >>= unfoldr (return . Right . Random.random)-->>> S.print $ S.take 4 (S.randoms :: Stream (Of Bool) IO ())-True-False-True-True--}-randoms :: (R.Random a, MonadIO m) => Stream (Of a) m r-randoms = do - g <- liftIO $ R.newStdGen- unfoldr (return . Right . R.random) g--{-| A crude infinite stream of random items between some bounds, using @System.Random@-->>> S.print $ S.take 4 $ S.randomRs (0,10^10::Int)-6489666022-3984407086-4271461383-3632382535--}-randomRs :: (R.Random a, MonadIO m) => (a, a) -> Stream (Of a) m r-randomRs limits = do - g <- liftIO $ R.getStdGen- unfoldr (return . Right . R.randomR limits) g---- --------------- -- read -- --------------- @@ -1261,17 +1295,17 @@ -- --------------- -- | 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 #-}+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' #-}+sum :: (Monad m, Num a) => Stream (Of a) m r -> m (Of a r)+sum = fold (+) 0 id+{-# INLINE sum #-} -- --------------- -- span@@ -1293,7 +1327,7 @@ {-| Split a stream of elements wherever a given element arises. The action is like that of 'Prelude.words'. ->>> S.stdoutLn $ mapsM S.toListM' $ split ' ' "hello world "+>>> S.stdoutLn $ mapsM S.toList $ split ' ' "hello world " hello world >>> Prelude.mapM_ Prelude.putStrLn (Prelude.words "hello world ")@@ -1338,14 +1372,9 @@ just a number of items from a stream of elements, but a number of substreams and the like. ->>> S.print $ mapsM sum' $ S.take 2 $ chunksOf 3 $ each [1..]+>>> 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->>> S.print $ mapsM S.sum' $ S.take 2 $ chunksOf 3 $ S.each [1..4] >> S.readLn-6 -- sum of first group of 3, which is already in [1..4]-100 -- user input-10000 -- user input-10104 -- sum of second group of 3 -} @@ -1404,15 +1433,6 @@ Step (a:>rest) -> yield a >> loop utc rest --- | Convert a pure @Stream (Of a)@ into a list of @as@-toList :: Stream (Of a) Identity () -> [a]-toList = loop- where- loop stream = case stream of- Return _ -> []- Delay (Identity stream') -> loop stream'- Step (a :> rest) -> a : loop rest-{-# INLINABLE toList #-} {-| Convert an effectful 'Stream (Of a)' into a list of @as@ @@ -1422,18 +1442,18 @@ is a leading cause of space leaks. -}-toListM :: Monad m => Stream (Of a) m () -> m [a]-toListM = fold (\diff a ls -> diff (a: ls)) id (\diff -> diff [])-{-# INLINE toListM #-}+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 -> maps' toListM' :: Stream (Stream (Of a)) m r -> Stream (Of [a]) m +> mapsM toListM :: Stream (Stream (Of a)) m r -> Stream (Of [a]) m -}-toListM' :: Monad m => Stream (Of a) m r -> m (Of [a] r)-toListM' = fold' (\diff a ls -> diff (a: ls)) id (\diff -> diff [])-{-# INLINE toListM' #-}+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. @@ -1452,7 +1472,7 @@ goodbye<Enter> goodbye ->>> S.drain $ S.unfoldr P.next (P.stdinLn P.>-> P.take 2 P.>-> P.stdoutLn)+>>> S.effects $ S.unfoldr P.next (P.stdinLn P.>-> P.take 2 P.>-> P.stdoutLn) hello<Enter> hello goodbye<Enter>@@ -1696,7 +1716,7 @@ -- , mapM_ -- -- , print -- -- , toHandle ----- , drain --+-- , effects -- -- -- -- * Pipes -- -- $pipes@@ -1752,3 +1772,24 @@ -- , 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)+ +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++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)+
streaming.cabal view
@@ -1,5 +1,5 @@ name: streaming-version: 0.1.1.1+version: 0.1.2.0 cabal-version: >=1.10 build-type: Simple synopsis: an elementary streaming prelude and a general monad transformer for streaming applications.@@ -68,7 +68,6 @@ , mmorph >=1.0 && <1.2 , transformers >=0.4 && <0.5 , bytestring- , random , time default-language: Haskell2010