streaming 0.1.4.0 → 0.1.4.1
raw patch · 5 files changed
+555/−303 lines, 5 filesdep +ghc-primdep −bytestringdep −containersdep ~exceptionsdep ~resourcetdep ~transformers
Dependencies added: ghc-prim
Dependencies removed: bytestring, containers
Dependency ranges changed: exceptions, resourcet, transformers, transformers-base
Files
- README.md +49/−92
- Streaming.hs +39/−22
- Streaming/Internal.hs +276/−67
- Streaming/Prelude.hs +144/−86
- streaming.cabal +47/−36
README.md view
@@ -1,106 +1,61 @@ streaming ========= -1. The freely generated stream on a streamable functor-2. A freely generated stream of individual Haskell values is a Producer, Generator or Source-3. `Streaming.Prelude`-4. Mother's `Prelude` v. `Streaming.Prelude`-5. How come there's not one of those fancy "ListT done right" implementations in here?-6. Didn't I hear that free monads are a dog from the point of view of efficiency?-7. Interoperation with the streaming-io libraries-8. Where can I find examples of use?-9. Problems-10. Implementation and benchmarking notes+Contents+-------- +§ 1. The freely generated stream on a streamable functor -`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`. +§ 2. A freely generated stream of individual Haskell values is a Producer, Generator or Source -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`. +§ 3. `Streaming.Prelude` -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. +§ 4. Mother's `Prelude` v. `Streaming.Prelude` -1. The freely generated stream on a streamable functor--------------------------------------------------------+§ 5. How come there's not one of those fancy "ListT done right" implementations in here? -(This section is a rather abstract defense of the inevitability of the leading type we are discussing, `Stream f m r` ; it may be well to skip to the next section.)+§ 6. Didn't I hear that free monads are a dog from the point of view of efficiency? -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. +§ 7. Interoperation with the streaming-io libraries -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:+§ 8. Where can I find examples of use? - splitter :: S -> (S, S)+§ 9. Problems -like the splitting operations we find with lists and the like, e.g. +§ 10. Implementation and benchmarking notes - 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:+§ 1. The freely generated stream on a streamable functor - splitter :: S m -> (S m, S m)+`Stream` can be used wherever [`FreeT`](https://hackage.haskell.org/package/free-4.12.1/docs/Control-Monad-Trans-Free.html) or [`Coroutine`](http://hackage.haskell.org/package/monad-coroutine-0.9.0.2/docs/Control-Monad-Coroutine.html#t:Coroutine) are used. The compiler's standard range of optimizations work better for operations written in terms of `Stream`. `Stream f m r`, like `FreeT f m r` or `Couroutine f m r` - is of course extremely general, and many functor-general combinators are exported by the general module `Streaming`. -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+In the applications we are thinking of, the general type `Stream f m r` expresses a succession of steps arising in a monad `m`, with a shape determined by the 'functor' parameter `f`, and resulting in a final value `r`. In the first instance you might read `Stream` as `Repeatedly`, with the understanding that one way of doing something some number of times, is to do it no times at all. - splitter :: S m -> m (S m, S m)- +Readings of `f` can be wildly various. Thus, for example, -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 + Stream Identity IO r+ +is the type of an indefinitely delayed `IO r`, or an extended `IO` process broken into stages marked by the `Identity` constructor. This is the `Trampoline` type of the ["Coroutine Pipelines" tutorial](https://themonadreader.files.wordpress.com/2011/10/issue19.pdf), and the [`IterT` of the `free` library](http://hackage.haskell.org/package/free-4.12.4/docs/Control-Monad-Trans-Iter.html) (which is mysteriously not identified with `FreeT Identity` - all of the associated combinators are found within the general `Streaming` module.) - splitter :: S m -> m (S Identity, S m)- -or+In particular, though, given readings of `f` and `m` we can, for example, always consider the type `Stream (Stream f m) m r`, in which steps of the form `Stream f m` are joined end to end. Such a stream-of-streams might arise in any number of ways; a crude (because hyper-general) way would be with - splitAccum :: S m -> m ([z], S m)+ chunksOf :: Monad m, Functor f => Int -> Stream f m r -> Stream (Stream f m) m r -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.)+and we can always rejoin such a stream with -This point makes it inevitable that *a rational stream type will have a return value*. It will have the form + concats :: Monad m, Functor f => Stream (Stream f m) m r -> Stream f m r - S r- -or +But other things can be chunked and concatenated in that sense; they need not themselves be explicitly represented in terms of `Stream`; indeed `chunksOf` and `concats` are modeled on those in `pipes-group`. In our [variant of `pipes-group`](https://hackage.haskell.org/package/streaming-utils-0.1.4.0/docs/Streaming-Pipes.html#v:concats), these have the types - S m r+ chunksOf :: Monad m => Int -> Producer a m r -> Stream (Producer a m) m r+ concats :: Monad m => Stream (Producer a m) m r -> Producer a m r -and the dividing functions will have the form -- splitter :: S r -> S (S r) --or-- 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 would 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".--General combinators for working with this idea of succession __irrespective of the form of succession__ are contained in the module `Stream`. They can be used, or example, to organize a succession of io-streams `Generator`s or pipes `Producer`s or the effectful bytestreams of the [streaming-bytestring](https://hackage.haskell.org/package/streaming-bytestring) library, or whatever stream-form you can express in a Haskell functor.--2. A freely generated stream of individual Haskell values is a Producer, Generator or Source-------------------------------------------------------------------------------------------------------+§ 2. A freely generated stream of individual Haskell values is a Producer, Generator or Source+--------------------------------------------------------------------------------------------------------- -But, of course, as soon as you grasp the general form of *succession*, you are already in possession of the most basic concrete form: a simple *succession of individual Haskell values* one after another. This is just `Stream ((,) a) m r`. Here we prefer `Stream (Of a) m r`, strictifying the left element of the pair with +Of course, as soon as you grasp the general form of *succession* you are already in possession of the most basic concrete form: a simple *succession of individual Haskell values* one after another, the effectful list or sequence. This is just `Stream ((,) a) m r`. Here we prefer to write `Stream (Of a) m r`, strictifying the left element of the pair with data Of a r = !a :> r deriving Functor @@ -125,9 +80,11 @@ machines: SourceT m a (= forall k. MachineT m k a) streaming: Stream (Of a) m () -3. `Streaming.Prelude`-----------------------+Note that the above libraries generally employ elaborate systems of type synonyms in order to intimate to the reader the meaning of specialized forms. `io-streams` is an exception. This libary is completely opposed to this tendency, and exports no synonyms. +§ 3. `Streaming.Prelude`+-------------------------+ `Streaming.Prelude` closely follows `Pipes.Prelude`. But since it restricts itself to use only of the general idea of streaming, it cleverly *omits the pipes*: ghci> S.stdoutLn $ S.take 2 S.stdinLn@@ -147,10 +104,10 @@ Somehow, we didn't even need a four-character operator for that, nor advice about best practices! - just ordinary Haskell common sense. -4. Mother's `Prelude` v. `Streaming.Prelude`---------------------------------------------+§ 4. Mother's `Prelude` v. `Streaming.Prelude`+------------------------------------------------ -The effort of `Streaming.Prelude` is to leverage the intuition the user has acquired in mastering `Prelude` and `Data.List` and to elevate her understanding into a general comprehension of effectful streaming transformations. Unsurprisingly, it takes longer to type out the signatures. It cannot be emphasized enough, thought, that *the transpositions are totally mechanical*:+The effort of `Streaming.Prelude` is to leverage the intuition the user has acquired in mastering `Prelude` and `Data.List` and to elevate her understanding into a general comprehension of effectful streaming transformations. Unsurprisingly, it takes longer to type out the signatures. It cannot be emphasized enough, though, that *the transpositions are totally mechanical*: Data.List.Split.chunksOf :: Int -> [a] -> [[a]] Streaming.chunksOf :: Int -> Stream f m r -> Stream (Stream f m) m r@@ -165,8 +122,8 @@ It is easy to prove that *resistance to these types is resistance to effectful streaming itself*. I will labor this point a bit more below, but you can also find it developed, with greater skill, in the documentation for the pipes libraries. -5. How come there's not one of those fancy "ListT done right" implementations in here?----------------------------------------------------------------------------------------+§ 5. How come there's not one of those fancy "ListT done right" implementations in here?+------------------------------------------------------------------------------------------ The use of the final return value appears to be a complication, but in fact it is essentially contained in the idea of effectful streaming. This is why this library does not export a \_ListT done right/, which would be simple enough - following `pipes`, as usual: @@ -181,8 +138,8 @@ Note similarly that you can write a certain kind of [take](http://hackage.haskell.org/package/machines-0.5.1/docs/Data-Machine-Process.html#v:taking) and [drop](http://hackage.haskell.org/package/machines-0.5.1/docs/Data-Machine-Process.html#v:dropping) with the `machines` library - as you can even with a "`ListT` done right". But I wish you luck writing `splitAt`! Similarly you can write a [getContents](http://hackage.haskell.org/package/machines-io-0.2.0.6/docs/System-IO-Machine.html); but I wish you luck dividing the resulting bytestream on its lines. This is - as usual! - because the library was not written with the general concept of effectful succession or streaming in view. Materials for sinking some elements of a stream in one way, and others in other ways - copying each line to a different file, as it might be, but without accumulation - are documented within. So are are myriad other elementary operations of streaming io. -6. Didn't I hear that free monads are a dog from the point of view of efficiency?-----------------------------------------------------------------------------------+§ 6. Didn't I hear that free monads are a dog from the point of view of efficiency?+------------------------------------------------------------------------------------- We noted above that if we instantiate `Stream f m r` to `Stream ((,) a) m r` or the like, we get the standard idea of a producer or generator. If it is instantiated to `Stream f Identity m r` then we have the standard \_free monad construction/. This construction is subject to certain familiar objections from an efficiency perspective; efforts have been made to substitute exotic cps-ed implementations and so forth. It is an interesting topic. @@ -247,8 +204,8 @@ 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. -7. Interoperation with the streaming-io libraries---------------------------------------------------+§ 7. Interoperation with the streaming-io libraries+----------------------------------------------------- The simplest form of interoperation with [pipes](http://hackage.haskell.org/package/pipes) is accomplished with this isomorphism: @@ -271,20 +228,20 @@ Free.iterTM Stream.wrap :: FreeT f m a -> Stream f m a Stream.iterTM Free.wrap :: Stream f m a -> FreeT f m a -8. Where can I find examples of use?--------------------------------------+§ 8. Where can I find examples of use?+----------------------------------------- For some simple ghci examples, see the commentary throughout the Prelude module. For slightly more advanced usage see the commentary in the haddocks of [streaming-bytestring](https://hackage.haskell.org/package/streaming-bytestring) and e.g. [these replicas](https://gist.github.com/michaelt/6c6843e6dd8030e95d58) of shell-like programs from the io-streams tutorial. Here's a simple [streaming GET request](https://gist.github.com/michaelt/2dcea1ba32562c091357) with intrinsically streaming byte streams. Here is a comically simple ['high - low' game](https://gist.github.com/michaelt/242f6a23267707ad29e9) -9. Problems-------------+§ 9. Problems+--------------- Questions about this library can be put as issues through the github site or on the [pipes mailing list](https://groups.google.com/forum/#!forum/haskell-pipes). (This library understands itself as part of the pipes "ecosystem.") -10. Implementation and benchmarking notes-------------------------------------------+§ 10. Implementation and benchmarking notes+---------------------------------------------- This library defines an optimized `FreeT` with an eye to use with streaming libraries, namely:
Streaming.hs view
@@ -5,17 +5,19 @@ -- $stream Stream, -- * Constructing a 'Stream' on a given functor- unfold, yields,+ effect,+ wrap, replicates, repeats, repeatsM,- effect,- wrap,+ unfold,+ never,+ untilJust, streamBuild,+ delays, -- * Transforming streams- decompose, maps, mapsM, mapped, @@ -25,32 +27,35 @@ -- * Inspecting a stream inspect, + -- * Splitting and joining 'Stream's + splitsAt,+ takes,+ chunksOf,+ concats,+ intercalates,+ cutoff, + -- period,+ -- periods, - -- * Zipping and unzipping streams+ + -- * Zipping, unzipping, separating and unseparating streams zipsWith, zips, unzips, interleaves, separate, unseparate,- + decompose,++ -- * Eliminating a 'Stream'+ mapsM_,+ run,+ streamFold, iterTM, iterT, destroy,- streamFold, - mapsM_,- run, - -- * Splitting and joining 'Stream's - splitsAt,- takes,- chunksOf,- concats,- intercalates,- -- period,- -- periods,- -- * Base functor for streams of individual items Of (..), lazily,@@ -66,24 +71,33 @@ MonadTrans(..), MonadIO(..), Compose(..),+ Sum(..),+ Identity(..),+ Alternative((<|>)), MonadThrow(..), MonadResource(..), MonadBase(..), ResourceT(..), runResourceT, join,+ liftM,+ liftM2, liftA2, liftA3, void,+ (<>) ) where import Streaming.Internal import Streaming.Prelude import Control.Monad.Morph import Control.Monad+import Data.Monoid ((<>)) import Control.Applicative import Control.Monad.Trans import Data.Functor.Compose +import Data.Functor.Sum+import Data.Functor.Identity import Control.Monad.Base import Control.Monad.Trans.Resource@@ -109,11 +123,14 @@ Others are surprisingly determinate in content: -> chunksOf :: Int -> Stream f m r -> Stream (Stream f m) m r-> splitsAt :: Int -> Stream f m r -> Stream f m (Stream f m r)-> zipsWith :: (forall x y. f x -> g y -> h (x, y)) -> Stream f m r -> Stream g m r -> Stream h m r+> chunksOf :: Int -> Stream f m r -> Stream (Stream f m) m r+> splitsAt :: Int -> Stream f m r -> Stream f m (Stream f m r)+> zipsWith :: (forall x y. f x -> g y -> h (x, y)) -> Stream f m r -> Stream g m r -> Stream h m r > intercalates :: Stream f m () -> Stream (Stream f m) m r -> Stream f m r-> groups: Stream (Sum f g) m r -> Stream (Sum (Stream f m) (Stream g m)) m r+> unzips :: Stream (Compose f g) m r -> Stream f (Stream g m) r +> separate :: Stream (Sum f g) m r -> Stream f (Stream g) m r -- cp. partitionEithers+> unseparate :: Stream f (Stream g) m r -> Stream (Sum f g) m r+> groups :: Stream (Sum f g) m r -> Stream (Sum (Stream f m) (Stream g m)) m r One way to see that /any/ streaming library needs some such general type is that it is required to represent the segmentation of a stream, and to
Streaming/Internal.hs view
@@ -15,6 +15,9 @@ , yields , streamBuild , cycles+ , delays+ , never+ , untilJust -- * Eliminating a stream , intercalates @@ -41,6 +44,7 @@ , chunksOf , splitsAt , takes+ , cutoff -- , period -- , periods @@ -73,6 +77,11 @@ import Control.Monad import Control.Monad.Trans import Control.Monad.Trans.Class+import Control.Monad.Reader.Class+import Control.Monad.Writer.Class+import Control.Monad.State.Class+import Control.Monad.Error.Class+import Control.Monad.Cont.Class import Control.Applicative import Data.Foldable ( Foldable(..) ) import Data.Traversable@@ -84,6 +93,7 @@ import Prelude hiding (splitAt) import Data.Functor.Compose import Data.Functor.Sum+import Control.Concurrent (threadDelay) -- import Data.Time (getCurrentTime, diffUTCTime, picosecondsToDiffTime, addUTCTime) import Control.Monad.Base@@ -183,33 +193,37 @@ "_bind (Return r) f" forall r f . _bind (Return r) f = f r; #-}- - instance (Functor f, Monad m) => Applicative (Stream f m) where pure = Return {-# INLINE pure #-} streamf <*> streamx = do {f <- streamf; x <- streamx; return (f x)} {-# INLINE (<*>) #-} - -- stra0 *> strb = loop stra0 where- -- loop stra = case stra of- -- Return _ -> strb- -- Effect m -> Effect (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- -- Effect m -> Effect (do {strb' <- m ; return (stra <* strb')})- -- Step fstr -> Step (fmap (stra <*) fstr)- -- {-# INLINABLE (<*) #-}- --++{- | The 'Alternative' instance glues streams together stepwise. ++> empty = never+> (<|>) = zipsWith (liftA2 (,))++ See also 'never', 'untilJust' and 'delays'+-}+instance (Applicative f, Monad m) => Alternative (Stream f m) where+ empty = never+ {-#INLINE empty #-}+ + str <|> str' = zipsWith (liftA2 (,)) str str'+ {-#INLINE (<|>) #-}++instance (Applicative f, Monad m) => MonadPlus (Stream f m) where+ mzero = empty+ mplus = (<|>)+ instance Functor f => MonadTrans (Stream f) where lift = Effect . liftM Return {-# INLINE lift #-} instance Functor f => MFunctor (Stream f) where- hoist trans = loop . unexposed where+ hoist trans = loop where loop stream = case stream of Return r -> Return r Effect m -> Effect (trans (liftM loop m))@@ -222,7 +236,7 @@ Return r -> Return r Effect m -> phi m >>= loop Step f -> Step (fmap loop f)- {-# INLINABLE embed #-} + {-# INLINABLE embed #-} instance (MonadIO m, Functor f) => MonadIO (Stream f m) where liftIO = Effect . liftM Return . liftIO@@ -235,9 +249,7 @@ instance (MonadThrow m, Functor f) => MonadThrow (Stream f m) where throwM = lift . throwM {-#INLINE throwM #-}- - instance (MonadCatch m, Functor f) => MonadCatch (Stream f m) where catch str f = go str where@@ -253,7 +265,52 @@ instance (MonadResource m, Functor f) => MonadResource (Stream f m) where liftResourceT = lift . liftResourceT {-#INLINE liftResourceT #-}- +++instance (Functor f, MonadReader r m) => MonadReader r (Stream f m) where+ ask = lift ask+ {-# INLINE ask #-}+ local f = hoist (local f)+ {-# INLINE local #-}++-- instance (Functor f, MonadWriter w m) => MonadWriter w (Stream f m) where+-- tell = lift . tell+-- {-# INLINE tell #-}+-- -- listen (FreeT m) = FreeT $ liftM concat' $ listen (fmap listen `liftM` m)+-- where+-- concat' (Pure x, w) = Pure (x, w)+-- concat' (Free y, w) = Free $ fmap (second (w <>)) <$> y+-- pass m = FreeT . pass' . runFreeT . hoist clean $ listen m+-- where+-- clean = pass . liftM (\x -> (x, const mempty))+-- pass' = join . liftM g+-- g (Pure ((x, f), w)) = tell (f w) >> return (Pure x)+-- g (Free f) = return . Free . fmap (FreeT . pass' . runFreeT) $ f+-- #if MIN_VERSION_mtl(2,1,1)+-- writer w = lift (writer w)+-- {-# INLINE writer #-}+-- #endif+--+instance (Functor f, MonadState s m) => MonadState s (Stream f m) where+ get = lift get+ {-# INLINE get #-}+ put = lift . put+ {-# INLINE put #-}+#if MIN_VERSION_mtl(2,1,1)+ state f = lift (state f)+ {-# INLINE state #-}+#endif++instance (Functor f, MonadError e m) => MonadError e (Stream f m) where+ throwError = lift . throwError+ {-# INLINE throwError #-}+ str `catchError` f = loop str where+ loop str = case str of+ Return r -> Return r+ Effect m -> Effect $ liftM loop m `catchError` (return . f)+ Step f -> Step (fmap loop f)+ {-# INLINABLE catchError #-}+ bracketStream :: (Functor f, MonadResource m) => IO a -> (a -> IO ()) -> (a -> Stream f m b) -> Stream f m b bracketStream alloc free inside = do@@ -266,13 +323,14 @@ Effect m -> Effect (liftM loop m) Step f -> Step (fmap loop f) {-#INLINABLE bracketStream #-}- ++ {-| Map a stream directly to its church encoding; compare @Data.List.foldr@ -} destroy :: (Functor f, Monad m) => Stream f m r -> (f b -> b) -> (m b -> b) -> (r -> b) -> b-destroy stream0 construct effect done = loop (unexposed stream0) where+destroy stream0 construct effect done = loop stream0 where loop stream = case stream of Return r -> done r Effect m -> effect (liftM loop m)@@ -306,17 +364,6 @@ (f (Stream g m a) -> m (g (Stream g m a))) -> Stream f m a -> Stream g m a -- mapped - So for example, when we realize that-->>> :t streamFold return Q.mwrap -(Monad m, Functor f) =>- (f (Q.ByteString m a) -> Q.ByteString m a)- -> Stream f m a -> Q.ByteString m a-- it is easy to see how to write @fromChunks@:-->>> streamFold return Q.mwrap (\(a:>b) -> Q.chunk a >> b)-Monad m => Stream (Of B.ByteString) m a -> Q.ByteString m a -- fromChunks -} streamFold :: (Functor f, Monad m) =>@@ -326,11 +373,11 @@ {- | Reflect a church-encoded stream; cp. @GHC.Exts.build@ -> destroy a b c (streamBuild psi) = +> streamFold return_ effect_ step_ (streamBuild psi) = psi return_ effect_ step_ -} streamBuild- :: (forall b . (f b -> b) -> (m b -> b) -> (r -> b) -> b) -> Stream f m r-streamBuild = \phi -> phi Step Effect Return+ :: (forall b . (r -> b) -> (m b -> b) -> (f b -> b) -> b) -> Stream f m r+streamBuild = \phi -> phi Return Effect Step {-# INLINE streamBuild #-} @@ -406,15 +453,22 @@ {-# INLINABLE mapsM #-} -{-| Resort a succession of layers of the form @m (f x)@. Though @mapsM@ - is best understood as:--> mapsM phi = decompose . maps (Compose . phi)+{-| Rearrange a succession of layers of the form @Compose m (f x)@. we could as well define @decompose@ by @mapsM@: -> decompose = mapsM getCompose+> decompose = mapped getCompose + but @mapped@ is best understood as:++> mapped phi = decompose . maps (Compose . phi)++ since @maps@ and @hoist@ are the really fundamental operations that preserve the+ shape of the stream:++> maps :: (Monad m, Functor f) => (forall x. f x -> g x) -> Stream f m r -> Stream g m r+> hoist :: (Monad m, Functor f) => (forall a. m a -> n a) -> Stream f m r -> Stream f n r+ -} decompose :: (Monad m, Functor f) => Stream (Compose m f) m r -> Stream f m r decompose = loop where@@ -522,7 +576,7 @@ -} splitsAt :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m (Stream f m r)-splitsAt = loop where+splitsAt = loop where loop !n stream | n <= 0 = Return stream | otherwise = case stream of@@ -584,23 +638,27 @@ => Stream f (t m) r -> t (Stream f m) r distribute = loop where loop stream = case stream of - Return r -> lift $ Return r- Effect tmstr -> hoist lift tmstr >>= distribute- Step fstr -> join $ lift (Step (fmap (Return . distribute) fstr))+ Return r -> lift (Return r)+ Effect tmstr -> hoist lift tmstr >>= loop+ Step fstr -> join (lift (Step (fmap (Return . loop) fstr)))+{-#INLINABLE distribute #-} --- | Repeat a functorial layer, command or instruction forever.+-- | Repeat a functorial layer (a \"command\" or \"instruction\") forever. repeats :: (Monad m, Functor f) => f () -> Stream f m r repeats f = loop where- loop = Step $ fmap (\_ -> loop) f+ loop = Effect (return (Step (fmap (\_ -> loop) f))) --- Repeat a functorial layer, command or instruction forever.+-- | Repeat an effect containing a functorial layer, command or instruction forever. repeatsM :: (Monad m, Functor f) => m (f ()) -> Stream f m r repeatsM mf = loop where loop = Effect $ do f <- mf return $ Step $ fmap (\_ -> loop) f --- | Repeat a functorial layer, command or instruct several times.+{- | Repeat a functorial layer, command or instruction a fixed number of times.++> replicates n = takes n . repeats +-} replicates :: (Monad m, Functor f) => Int -> f () -> Stream f m () replicates n f = splitsAt n (repeats f) >> return () @@ -671,21 +729,55 @@ {-# INLINABLE unexposed #-} +{- Wrap a new layer of a stream. So, e.g. -effect :: (Monad m, Functor f ) => m (Stream f m r) -> Stream f m r-effect = Effect-{-#INLINE effect #-}+> S.cons :: Monad m => a -> Stream (Of a) m r -> Stream (Of a) m r+> S.cons a str = wrap (a :> str) + and, recursively:++> S.each :: (Monad m, Foldable t) => t a -> Stream (Of a) m ()+> S.each = foldr (\a b -> wrap (a :> b)) (return ())++ The two operations ++> wrap :: (Monad m, Functor f ) => f (Stream f m r) -> Stream f m r+> effect :: (Monad m, Functor f ) => m (Stream f m r) -> Stream f m r++ are fundamental. We can define the parallel operations @yields@ and @lift@ in + terms of them++> yields :: (Monad m, Functor f ) => f r -> Stream f m r+> yields = wrap . fmap return+> lift :: (Monad m, Functor f ) => m r -> Stream f m r+> lift = effect . fmap return++-} wrap :: (Monad m, Functor f ) => f (Stream f m r) -> Stream f m r wrap = Step {-#INLINE wrap #-} -{-| Lift for items in the base functor. Makes a singleton or- one-layer succession. It is named by similarity to lift: +{- | Wrap an effect that returns a stream -> lift :: (Monad m, Functor f) => m r -> Stream f m r+> effect = join . lift ++-}+effect :: (Monad m, Functor f ) => m (Stream f m r) -> Stream f m r+effect = Effect+{-#INLINE effect #-}+++{-| @yields@ is like @lift@ for items in the streamed functor. + It makes a singleton or one-layer succession. ++> lift :: (Monad m, Functor f) => m r -> Stream f m r > yields :: (Monad m, Functor f) => f r -> Stream f m r++ Viewed in another light, it is like a functor-general version of @yield@:++> S.yield a = yields (a :> ())+ -} yields :: (Monad m, Functor f) => f r -> Stream f m r@@ -697,12 +789,12 @@ => (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) = Effect $ go s1 s2- go (Return r) p = return $ Return r- go q (Return s) = return $ Return s- go (Effect m) p = m >>= \s -> go s p- go q (Effect m) = m >>= go q- go (Step f) (Step g) = return $ Step $ fmap loop (phi f g)+ loop (s1, s2) = Effect (go s1 s2)+ go (Return r) p = return (Return r)+ go q (Return s) = return (Return s)+ go (Effect m) p = m >>= (\s -> go s p)+ go q (Effect m) = m >>= (\s -> go q s)+ go (Step f) (Step g) = return (Step (fmap loop (phi f g))) {-# INLINABLE zipsWith #-} zips :: (Monad m, Functor f, Functor g) @@ -712,6 +804,7 @@ {-# INLINE zips #-} + {-| Interleave functor layers, with the effects of the first preceding the effects of the second. @@ -753,8 +846,9 @@ {-| 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.- + useful for acting on one or the other functor with a fold. It generalizes+ 'Data.Either.partitionEithers' massively, but actually streams properly.+ >>> let odd_even = S.maps (S.distinguish even) $ S.each [1..10::Int] >>> :t separate odd_even separate odd_even@@ -762,13 +856,30 @@ Now, for example, it is convenient to fold on the left and right values separately: ->>> toList $ toList $ separate odd_even+>>> S.toList $ S.toList $ separate odd_even [2,4,6,8,10] :> ([1,3,5,7,9] :> ()) - We can achieve the above effect more simply- in the case of @Stream (Of a) m r@ by using 'Streaming.Prelude.duplicate' ->>> S.toList . S.filter even $ S.toList . S.filter odd $ S.duplicate $ each [1..10::Int]+ Or we can write them to separate files or whatever:++>>> runResourceT $ S.writeFile "even.txt" . S.show $ S.writeFile "odd.txt" . S.show $ S.separate odd_even +>>> :! cat even.txt +2+4+6+8+10+>>> :! cat odd.txt+1+3+5+7+9++ Of course, in the special case of @Stream (Of a) m r@, we can achieve the above + effects more simply by using 'Streaming.Prelude.copy'++>>> S.toList . S.filter even $ S.toList . S.filter odd $ S.copy $ each [1..10::Int] [2,4,6,8,10] :> ([1,3,5,7,9] :> ()) @@ -862,6 +973,82 @@ -- Right (InL fstr) -> wrap (fmap loop fstr) -- Right (InR gstr) -> return (wrap (InR gstr)) +{- | 'never' interleaves the pure applicative action with the return of the monad forever. + It is the 'empty' of the 'Alternative' instance, thus++> never <|> a = a+> a <|> never = a++ and so on. If w is a monoid then @never :: Stream (Of w) m r@ is+ the infinite sequence of 'mempty', and+ @str1 \<|\> str2@ appends the elements monoidally until one of streams ends.+ Thus we have, e.g.++>>> S.stdoutLn $ S.take 2 $ S.stdinLn <|> S.repeat " " <|> S.stdinLn <|> S.repeat " " <|> S.stdinLn +1<Enter> +2<Enter>+3<Enter>+1 2 3+4<Enter>+5<Enter>+6<Enter>+4 5 6++ This is equivalent to ++>>> S.stdoutLn $ S.take 2 $ foldr (<|>) never [S.stdinLn, S.repeat " ", S.stdinLn, S.repeat " ", S.stdinLn ]++ Where 'f' is a monad, @(\<|\>)@ sequences the conjoined streams stepwise. See the+ definition of @paste@ <https://gist.github.com/michaelt/6c6843e6dd8030e95d58 here>,+ where the separate steps are bytestreams corresponding to the lines of a file. ++ Given, say,++> data Branch r = Branch r r deriving Functor -- add obvious applicative instance++ then @never :: Stream Branch Identity r@ is the pure infinite binary tree with+ (inaccessible) @r@s in its leaves. Given two binary trees, @tree1 \<|\> tree2@+ intersects them, preserving the leaves that came first, + so @tree1 \<|\> never = tree1@++ @Stream Identity m r@ is an action in @m@ that is indefinitely delayed. Such an + action can be constructed with e.g. 'untilJust'.++> untilJust :: (Monad m, Applicative f) => m (Maybe r) -> Stream f m r++ Given two such items, @\<|\>@ instance races them. + It is thus the iterative monad transformer specially defined in + <https://hackage.haskell.org/package/free-4.12.1/docs/Control-Monad-Trans-Iter.html Control.Monad.Trans.Iter>++ So, for example, we might write++>>> let justFour str = if length str == 4 then Just str else Nothing+>>> let four = untilJust (liftM justFour getLine) +>>> run four+one<Enter>+two<Enter>+three<Enter>+four<Enter>+"four"+++ The 'Alternative' instance in + <https://hackage.haskell.org/package/free-4.12.1/docs/Control-Monad-Trans-Free.html Control.Monad.Trans.Free> + is avowedly wrong, though no explanation is given for this.+++-}+never :: (Monad m, Applicative f) => Stream f m r+never = let loop = Effect $ return $ Step $ pure loop in loop+{-#INLINABLE never #-} +++delays :: (MonadIO m, Applicative f) => Double -> Stream f m r+delays seconds = loop where+ loop = Effect $ liftIO (threadDelay delay) >> return (Step (pure loop))+ delay = fromInteger (truncate (1000000 * seconds)) +{-#INLINABLE delays #-}+ -- {-| Permit streamed actions to proceed unless the clock has run out. -- -- -}@@ -934,3 +1121,25 @@ -- Effect m -> Effect (liftM loop m) -- Step f -> Step (fmap loop f) -- loop str++{- | Repeat a ++-}++untilJust :: (Monad m, Applicative f) => m (Maybe r) -> Stream f m r+untilJust act = loop where+ loop = Effect $ do+ m <- act+ case m of + Nothing -> return $ Step $ pure loop+ Just a -> return $ Return a+ + +cutoff :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m (Maybe r)+cutoff = loop where+ loop 0 str = return Nothing+ loop n str = do + e <- lift $ inspect str+ case e of+ Left r -> return (Just r)+ Right (frest) -> Step $ fmap (loop (n-1)) frest
Streaming/Prelude.hs view
@@ -51,6 +51,7 @@ -- $producers , yield , each+ , each' , unfoldr , stdinLn , readLn@@ -97,7 +98,6 @@ , store , chain , sequence- , nub , filter , filterM , delay@@ -207,6 +207,11 @@ , partitionEithers , partition + -- * Maybes+ -- $maybes+ , catMaybes+ , mapMaybe+ -- * Pair manipulation , lazily , strictly@@ -254,10 +259,9 @@ import Data.Functor.Classes import Data.Functor.Compose import Control.Monad.Trans.Resource-import qualified Data.Set as Set import GHC.Exts ( SpecConstrAnnotation(..) )-+import GHC.Magic data SPEC = SPEC | SPEC2 @@ -301,10 +305,10 @@ 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+-- 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 @@ -317,18 +321,26 @@ > lazily :: Of a ~~> (,) a > Identity . fst' :: Of a ~~> Identity a - Manipulation of a @Stream f m r@ by mapping often turns on recognizing natural transformations of @f@,- thus @maps@ is far more general the the @map@ of the present module, which can be+ Manipulation of a @Stream f m r@ by mapping often turns on recognizing natural transformations of @f@.+ Thus @maps@ is far more general the the @map@ of the @Streaming.Prelude@, which can be defined thus: > S.map :: (a -> b) -> Stream (Of a) m r -> Stream (Of b) m r > S.map f = maps (mapOf f)++ i.e.++> S.map f = maps (\(a :> x) -> (f a :> x)) This rests on recognizing that @mapOf@ is a natural transformation; note though that it results in such a transformation as well: > S.map :: (a -> b) -> Stream (Of a) m ~> Stream (Of b) m + Thus we can @maps@ it in turn++> + -} lazily :: Of a b -> (a,b) lazily = \(a:>b) -> (a,b)@@ -340,13 +352,19 @@ fst' :: Of a b -> a fst' (a :> b) = a+{-#INLINE fst' #-} snd' :: Of a b -> b snd' (a :> b) = b+{-#INLINE snd' #-} mapOf :: (a -> b) -> Of a r -> Of b r mapOf f (a:> b) = (f a :> b)+{-#INLINE mapOf #-} +_first :: Functor f => (a -> f a1) -> Of a b -> f (Of a1 b)+_first afb (a:>b) = fmap (\c -> (c:>b)) (afb a)+{-# INLINE _first #-} all :: Monad m => (a -> Bool) -> Stream (Of a) m r -> m (Of Bool r) all thus = loop True where@@ -580,32 +598,7 @@ loop rest {-#INLINABLE delay #-} --- ------------------ effects--- --------------- -{- | Reduce a stream, performing its actions but ignoring its elements. - ->>> rest <- S.effects $ S.splitAt 2 $ each [1..5]->>> S.print rest-3-4-5- 'effects' should be understood together with 'copy' and is subject to the rules--> S.effects . S.copy = id-> hoist S.effects . S.copy = id-- The similar @effects@ and @copy@ operations in @Data.ByteString.Streaming@ obey the same rules. ---}-effects :: Monad m => Stream (Of a) m r -> m r-effects = loop where- loop stream = case stream of - Return r -> return r- Effect m -> m >>= loop - Step (_ :> rest) -> loop rest-{-#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@@ -715,9 +708,40 @@ -} each :: (Monad m, Foldable.Foldable f) => f a -> Stream (Of a) m ()-each = Foldable.foldr (\a p -> (Step (a :> p))) (Return ())+each = Foldable.foldr (\a p -> Step (a :> p)) (Return ()) {-# INLINABLE each #-} +each' :: (Monad m, Foldable.Foldable f) => f a -> Stream (Of a) m ()+each' = Foldable.foldr (\a p -> Effect (return (Step (a :> p)))) (Return ())+{-# INLINABLE each' #-}++-- ---------------+-- effects+-- ---------------++{- | Reduce a stream, performing its actions but ignoring its elements. + +>>> rest <- S.effects $ S.splitAt 2 $ each [1..5]+>>> S.print rest+3+4+5+ 'effects' should be understood together with 'copy' and is subject to the rules++> S.effects . S.copy = id+> hoist S.effects . S.copy = id++ The similar @effects@ and @copy@ operations in @Data.ByteString.Streaming@ obey the same rules. ++-}+effects :: Monad m => Stream (Of a) m r -> m r+effects = loop where+ loop stream = case stream of + Return r -> return r+ Effect m -> m >>= loop + Step (_ :> rest) -> loop rest+{-#INLINABLE effects #-}+ {-| Exhaust a stream remembering only whether @a@ was an element. -}@@ -769,7 +793,7 @@ -} enumFrom :: (Monad m, Enum n) => n -> Stream (Of n) m r enumFrom = loop where- loop !n = Step (n :> loop (succ n))+ loop !n = Effect (return (Step (n :> loop (succ n)))) {-# INLINABLE enumFrom #-} @@ -946,13 +970,13 @@ -} fold :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> m (Of b r)-fold step begin done str = fold_loop str begin+fold step begin done str = fold_loop str begin where- fold_loop stream !x = case stream of + fold_loop stream !x = case stream of Return r -> return (done x :> r) Effect m -> m >>= \str' -> fold_loop str' x Step (a :> rest) -> fold_loop rest $! step x a-{-# INLINABLE fold #-}+{-# INLINE fold #-} {-| Strict, monadic fold of the elements of a 'Stream (Of a)'@@ -979,11 +1003,12 @@ 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 x0 <- begin loop str x0 where- loop stream !x = case stream of + loop stream !x = case stream of Return r -> done x >>= \b -> return (b :> r) Effect m -> m >>= \s -> loop s x Step (a :> rest) -> do@@ -991,7 +1016,19 @@ loop rest x' {-# INLINABLE foldM #-} -+-- the following requires GHC.Magic.oneShot:+-- foldM step begin done str = do+-- x <- begin+-- (x' :> r) <- streamFold+-- (\r x -> return (x :> r))+-- (\mx2mx -> oneShot (\x -> x `seq` mx2mx >>= ($ x) ))+-- (\(a :> x2mx') -> oneShot (\x -> x `seq` (step x a >>= x2mx')) )+-- ( str)+-- x+-- b <- done x'+-- return (b :> r)+-- where seq = Prelude.seq+-- {-#INLINE foldM #-} {-| A natural right fold for consuming a stream of elements. See also the more general 'iterTM' in the 'Streaming' module @@ -1145,9 +1182,9 @@ -}-iterate :: (a -> a) -> a -> Stream (Of a) m r+iterate :: Monad m => (a -> a) -> a -> Stream (Of a) m r iterate f = loop where- loop a' = Step (a' :> loop (f a'))+ loop a' = Effect (return (Step (a' :> loop (f a')))) {-# INLINABLE iterate #-} -- | Iterate a monadic function from a seed value, streaming the results forever@@ -1284,20 +1321,29 @@ -{- | Map layers of one functor to another with a transformation involving the base monad+{- | Map layers of one functor to another with a transformation involving the base monad.+ This could be trivial, e.g.++> let noteBeginning text x = putStrLn text >> return text++ this puts the+ is completely functor-general ++ @maps@ and @mapped@ obey these rules:++> maps id = id+> mapped return = id+> maps f . maps g = maps (f . g)+> mapped f . mapped g = mapped (f <=< g)+> maps f . mapped g = mapped (liftM f . g)+> mapped f . maps g = mapped (f <=< liftM g)+ @maps@ is more fundamental than @mapped@, which is best understood as a convenience for effecting this frequent composition: -> mapped = mapsM -> mapsM phi = decompose . maps (Compose . phi) +> mapped phi = decompose . maps (Compose . phi) - @mapped@ obeys these rules: -> mapped return = id-> mapped f . mapped g = mapped (f <=< g)-> map f . mapped g = mapped (liftM f . g)-> mapped f . map g = mapped (f . g)- -} mapped :: (Monad m, Functor f) => (forall x . f x -> m (g x)) -> Stream f m r -> Stream g m r@@ -1403,28 +1449,7 @@ else loop True rest {-#INLINABLE notElem_ #-} -{-| Remove repeated elements from a Stream. 'nub' of course accumulates a 'Data.Set.Set' of- elements that have already been seen and should thus be used with care.- ->>> S.toList_ $ S.nub $ S.take 5 S.readLn :: IO ([Int])-1<Enter>-2<Enter>-3<Enter>-1<Enter>-2<Enter>-[1,2,3] --}--nub :: (Monad m, Ord a) => Stream (Of a) m r -> Stream (Of a) m r-nub = loop Set.empty where- loop !set stream = case stream of - Return r -> Return r- Effect m -> Effect (liftM (loop set) m)- Step (a :> rest) -> if Set.member a set - then loop set rest- else Step (a :> loop (Set.insert a set) rest)- {-| > filter p = hoist effects (partition p) @@ -1517,8 +1542,8 @@ 1 -} -repeat :: a -> Stream (Of a) m r-repeat a = loop where loop = Step (a :> loop)+repeat :: Monad m => a -> Stream (Of a) m r+repeat a = loop where loop = Effect (return (Step (a :> loop))) {-# INLINE repeat #-} @@ -1547,7 +1572,7 @@ replicate n a | n <= 0 = return () replicate n a = loop n where loop 0 = Return ()- loop m = Step (a :> loop (m-1))+ loop m = Effect (return (Step (a :> loop (m-1)))) {-# INLINABLE replicate #-} {-| Repeat an action several times, streaming the results.@@ -1605,12 +1630,9 @@ where loop !acc stream = do case stream of - Return r -> yield (done acc) >> return r+ Return r -> Step (done acc :> Return r) Effect m -> Effect (liftM (loop acc) m)- Step (a :> rest) -> do- yield (done acc) - let !acc' = step acc a- loop acc' rest+ Step (a :> rest) -> Step (done acc :> loop (step acc a) rest) {-#INLINABLE scan #-} {-| Strict left scan, accepting a monadic function. It can be used with@@ -1695,7 +1717,7 @@ To restrict user input to some number of seconds, we might write: ->>> S.toList $ S.zipWith (flip const) (S.takeWhile (< 5) S.seconds) S.stdinLn+>>> S.toList $ S.map fst $ S.zip S.stdinLn $ S.takeWhile (< 3) S.seconds one<Enter> two<Enter> three<Enter>@@ -1703,6 +1725,8 @@ five<Enter> ["one","two","three","four","five"] :> () + This is of course does not interrupt an action that has already begun.+ -} seconds :: Stream (Of Double) IO r@@ -2575,11 +2599,12 @@ copy :: Monad m => Stream (Of a) m r -> Stream (Of a) (Stream (Of a) m) r-copy = loop where+copy = Effect . return . loop where loop str = case str of Return r -> Return r Effect m -> Effect (liftM loop (lift m))- Step (a :> rest) -> Step (a :> Effect (Step (a :> Return (loop rest)))) + Step (a :> rest) -> Effect (Step (a :> Return (Step (a :> loop rest))))+ {-#INLINABLE copy#-} duplicate@@ -2596,11 +2621,11 @@ copy' :: Monad m => Stream (Of a) m r -> Stream (Of a) (Stream (Of a) m) r-copy' = loop where+copy' = Effect . return . loop where loop str = case str of Return r -> Return r Effect m -> Effect (liftM loop (lift m))- Step (a :> rest) -> Effect (Step (a :> Return (Step (a :> loop rest))))+ Step (a :> rest) -> Step (a :> Effect (Step (a :> Return (loop rest)))) {-#INLINABLE copy' #-} duplicate'@@ -2674,3 +2699,36 @@ -- "scan/map" forall step begin done f str . -- scan step begin done (map f str) = scan (\x a -> step x $! f a) begin done str --++{- $maybes+ These functions discard the 'Nothing's that they encounter. They are analogous + to the functions from @Data.Maybe@ that share their names.+-}++{-| The 'catMaybes' function takes a 'Stream' of 'Maybe's and returns+ a 'Stream' of all of the 'Just' values.+-}+catMaybes :: Monad m => Stream (Of (Maybe a)) m r -> Stream (Of a) m r +catMaybes = loop where + loop stream = case stream of + Return r -> Return r + Effect m -> Effect (liftM loop m) + Step (ma :> snext) -> case ma of + Nothing -> loop snext + Just a -> Step (a :> loop snext) +{-#INLINABLE catMaybes #-}++{-| The 'mapMaybe' function is a version of 'map' which can throw out elements. In particular, + the functional argument returns something of type @'Maybe' b@. If this is 'Nothing', no element + is added on to the result 'Stream'. If it is @'Just' b@, then @b@ is included in the result 'Stream'.+-}+mapMaybe :: Monad m => (a -> Maybe b) -> Stream (Of a) m r -> Stream (Of b) m r +mapMaybe phi = loop where + loop stream = case stream of + Return r -> Return r + Effect m -> Effect (liftM loop m) + Step (a :> snext) -> case phi a of + Nothing -> loop snext + Just b -> Step (b :> loop snext) +{-#INLINABLE mapMaybe #-}+
streaming.cabal view
@@ -1,10 +1,10 @@ name: streaming-version: 0.1.4.0+version: 0.1.4.1 cabal-version: >=1.10 build-type: Simple synopsis: an elementary streaming prelude and general stream type. -description: @Streaming.Prelude@ exports an elementary streaming prelude focussed on+description: @Streaming.Prelude@ exports an elementary streaming prelude focused on a simple \"source\" or \"producer\" type, namely @Stream (Of a) m r@. This is a sort of effectful version of @([a],r)@ in which monadic action is interleaved between successive elements.@@ -43,36 +43,50 @@ intuitions the user has gathered from mastery of @Prelude@ and @Data.List@. The two conceptual pre-requisites are some comprehension of monad transformers and some familiarity- with \'rank 2 types\'.+ with \'rank 2 types\'. . See the- <https://hackage.haskell.org/package/streaming#readme readme> below- for an explanation, including the examples linked there. Elementary usage can be divined from the ghci examples in+ <https://hackage.haskell.org/package/streaming#readme readme> + below for an explanation, including the examples linked there. + Elementary usage can be divined from the ghci examples in @Streaming.Prelude@ and perhaps from this rough beginning of a <https://github.com/michaelt/streaming-tutorial/blob/master/tutorial.md tutorial>. Note also the <https://hackage.haskell.org/package/streaming-bytestring streaming bytestring> and <https://hackage.haskell.org/package/streaming-utils streaming utils>- packages.+ packages. Questions about usage can be put+ raised on StackOverflow with the tag @[haskell-streaming]@, + or as an issue on Github, or on the + <https://groups.google.com/forum/#!forum/haskell-pipes pipes list>+ (the package understands itself as part of the pipes \'ecosystem\'.) .- The simplest form of interoperation with <http://hackage.haskell.org/package/pipes pipes>+ The simplest form of interoperation with + <http://hackage.haskell.org/package/pipes pipes> is accomplished with this isomorphism: . > 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 <http://hackage.haskell.org/package/io-streams io-streams> is thus:+ Interoperation with + <http://hackage.haskell.org/package/io-streams io-streams> + is thus: . > Streaming.reread IOStreams.read :: InputStream a -> Stream (Of a) IO () > IOStreams.unfoldM Streaming.uncons :: Stream (Of a) IO () -> IO (InputStream a) .- A simple exit to <http://hackage.haskell.org/package/conduit conduit> would be, e.g.:+ With + <http://hackage.haskell.org/package/conduit conduit> + one might use, e.g.: .- > Conduit.unfoldM Streaming.uncons :: Stream (Of a) m () -> Source m a+ > Conduit.unfoldM Streaming.uncons :: Stream (Of a) m () -> Source m a+ > Streaming.mapM_ Conduit.yield . hoist lift :: Stream (Of o) m r -> ConduitM i o m r+ > ($$ Conduit.mapM_ Streaming.yield) . hoist lift :: Source m a -> Stream (Of a) m () . These conversions should never be more expensive than a single @>->@ or @=$=@. - Here is a simple example that runs a single underlying stream with several+ .+ Here is a simple example (conceptually it is a bit advanced, maybe) + that runs a single underlying stream with several streaming-io libraries at once, superimposing their effects without any accumulation: .@@ -92,26 +106,25 @@ > mkIOStream = IOS.unfoldM S.uncons > > main = iostreamed where- > urstream = S.take 4 S.readLn :: Stream (Of Int) IO () - > streamed = S.copy urstream- > & S.map (\n -> "streaming says: " ++ show n) - > & S.stdoutLn + > urstream = S.take 3 S.readLn :: Stream (Of Int) IO () + + > streamed = S.copy urstream & S.map (\n -> "streaming says: " ++ show n) + > & S.stdoutLn > piped = runEffect $ - > mkPipe (S.copy streamed)- > >-> P.map (\n -> "pipes says: " ++ show n)- > >-> P.stdoutLn - > conduited = mkConduit (S.copy piped) - > $$ CL.map (\n -> "conduit says: " ++ show n)- > =$ CL.mapM_ (liftIO . putStrLn)+ > mkPipe (S.copy streamed) >-> P.map (\n -> "pipes says: " ++ show n) + > >-> P.stdoutLn + > conduited = + > mkConduit (S.copy piped) $$ CL.map (\n -> "conduit says: " ++ show n) + > =$ CL.mapM_ (liftIO . putStrLn) > iostreamed = do > str0 <- mkIOStream conduited > str1 <- IOS.map (\n -> pack $ "io-streams says: " ++ show n ++ "\n") str0 > IOS.supply str1 IOS.stdout .- This program successively parses four @Int@s from standard input, + This program successively parses three @Int@s from standard input, and /simulaneously/ passes them to (here trivial) stream-consuming processes from four different libraries, using the @copy@ function from- @Streaming.Prelude@. I mark my own input with @/<Enter/>@:+ @Streaming.Prelude@. I mark my own input with @/<Enter/>@ below: . > >>> main > 1 <Enter>@@ -129,14 +142,12 @@ > pipes says: 3 > conduit says: 3 > io-streams says: 3- > 4 <Enter>- > streaming says: 4- > pipes says: 4- > conduit says: 4- > io-streams says: 4+ > >>> . Of course, I could as well have passed the stream to several- independent conduits, for example. Further+ independent conduits; and I might have derived the original+ stream from a conduit @Source@ or pipes @Producer@ etc., using+ one of the \'conversion\' functions above. Further points of comparison with the going streaming-IO libraries are discussed in the <https://hackage.haskell.org/package/streaming#readme readme>@@ -151,7 +162,7 @@ <<http://i.imgur.com/sSG5MvH.png>> . Because these are microbenchmarks for individual functions, - they represent a sort of "worst case"; many other factors can influence+ they represent a sort of \"worst case\"; many other factors can influence the speed of a complex program. .@@ -187,13 +198,13 @@ build-depends: base >=4.6 && <5 , mtl >=2.1 && <2.3 , mmorph >=1.0 && <1.2- , transformers >=0.4 && <0.5- , transformers-base- , bytestring+ , transformers >=0.4 && <0.5.2+ , transformers-base < 0.5+ , resourcet > 1.1.0 && < 1.2+ , exceptions > 0.5 && < 0.9 , time- , resourcet- , exceptions- , containers+ , ghc-prim+ default-language: Haskell2010