streaming 0.1.1.0 → 0.1.1.1
raw patch · 2 files changed
+324/−72 lines, 2 filesdep +time
Dependencies added: time
Files
- Streaming/Prelude.hs +302/−71
- streaming.cabal +22/−1
Streaming/Prelude.hs view
@@ -41,10 +41,6 @@ module Streaming.Prelude ( -- * Types Of (..)- , lazily- , strictly- , fst'- , snd' -- * Introducing streams of elements -- $producers@@ -72,17 +68,20 @@ , print , toHandle , drain+ , drained -- * Stream transformers -- $pipes , map , mapM+ , chain , maps , sequence , mapFoldable , filter , filterM , for+ , delay , take , takeWhile -- , takeWhile'@@ -93,7 +92,7 @@ -- , findIndices , scan , scanM- , chain+ , scanned , read , show , cons@@ -102,11 +101,21 @@ , next , uncons , splitAt+ , split+ , breaks , break+ , breakWhen , span , group , groupBy+ , timed -- , split+ + -- * Pair manipulation+ , lazily+ , strictly+ , fst'+ , snd' -- * Folds -- $folds@@ -181,6 +190,8 @@ import Data.Monoid (Monoid (..)) import Data.String (IsString (..)) import qualified System.Random as R+import Control.Concurrent (threadDelay)+import Data.Time (getCurrentTime, diffUTCTime, picosecondsToDiffTime) -- | A left-strict pair; the base functor for streams of individual elements. data Of a b = !a :> b deriving (Data, Eq, Foldable, Ord,@@ -225,16 +236,17 @@ {-| Note that 'lazily', 'strictly', 'fst'', and 'mapOf' are all so-called /natural transformations/ on the primitive @Of a@ functor If we write -> type f ~> g = forall x . f x -> g x+> type f ~~> g = forall x . f x -> g x - then we have+ then we can restate some types as follows: -> mapOf :: (a -> b) -> Of a ~> Of b-> lazily :: Of a -> (,) a-> fst' :: Of a -> Identity a+> mapOf :: (a -> b) -> Of a ~~> Of b -- bifunctor lmap+> 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+ thus @maps@ is far more general the the @map@ of the present module, 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)@@ -243,7 +255,6 @@ that it results in such a transformation as well: > S.map :: (a -> b) -> Stream (Of a) m ~> Stream (Of b) m - -} lazily :: Of a b -> (a,b)@@ -262,6 +273,7 @@ mapOf :: (a -> b) -> Of a r -> Of b r mapOf f (a:> b) = (f a :> b)+ {-| Break a sequence when a element falls under a predicate, keeping the rest of the stream as the return value. @@ -286,10 +298,63 @@ else Step (a :> loop rest) {-# INLINEABLE break #-} +{-| Yield elements, using a fold to maintain state, until the accumulated + value satifies the supplied predicate. The fold will then be short-circuited + and the element that breaks it will be included with the stream returned.+ This function is easiest to use with 'Control.Foldl.purely'++>>> rest <- S.print $ L.purely S.breakWhen L.sum even $ S.each [1,2,3,4]+1+2+>>> S.print rest+3+4++-}+breakWhen :: Monad m => (x -> a -> x) -> x -> (x -> b) -> (b -> Bool) -> Stream (Of a) m r -> Stream (Of a) m (Stream (Of a) m r)+breakWhen step begin done pred = loop0 begin+ where+ loop0 x stream = case stream of + Return r -> return (return r)+ Delay mn -> Delay $ liftM (loop0 x) mn+ Step (a :> rest) -> loop a (step x a) rest+ loop a !x stream = do+ if pred (done x) + then return (yield a >> stream) + else case stream of + Return r -> yield a >> return (return r)+ Delay mn -> Delay $ liftM (loop a x) mn+ Step (a' :> rest) -> do+ yield a+ 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]++-}+breaks+ :: Monad m =>+ (a -> Bool) -> Stream (Of a) m r -> Stream (Stream (Of a) m) m r+breaks thus = loop where+ loop stream = Delay $ do+ e <- next stream+ return $ case e of+ Left r -> Return r+ Right (a, p') -> + if not (thus a)+ then Step $ fmap loop (yield a >> break thus p')+ else loop p'+{-#INLINABLE breaks #-}+ + {-| Apply an action to all values flowing downstream ->>> S.product (chain print (S.each [2..4])) >>= print+>>> S.product (S.chain Prelude.print (S.each [2..4])) >>= Prelude.print 2 3 4@@ -302,20 +367,27 @@ yield a {-# INLINE chain #-} -{-| Make a stream of traversable containers into a stream of their separate elements+{-| Make a stream of traversable containers into a stream of their separate elements.+ This is just +> concat = for str each+ >>> S.print $ S.concat (each ["xy","z"]) 'x' 'y' 'z'->>> S.print $ S.concat (S.each [Just 1, Nothing, Just 2])++ Note that it also has the effect of 'Data.Maybe.catMaybes' and 'Data.Either.rights'+++>>> S.print $ S.concat $ S.each [Just 1, Nothing, Just 2] 1 2->>> S.print $ S.concat (S.each [Right 1, Left "Error!", Right 2])+>>> S.print $ S.concat $ S.each [Right 1, Left "Error!", Right 2] 1 2 - Not to be confused with the functor-general + @concat@ is not to be confused with the functor-general > concats :: (Monad m, Functor f) => Stream (Stream f m) m r -> Stream f m r -- specializing @@ -372,13 +444,23 @@ cycle :: (Monad m, Functor f) => Stream f m r -> Stream f m s cycle = forever ++{-| Delay each element by the supplied number of seconds.+mapM :: Monad m => (a -> m b) -> Stream (Of a) m r -> Stream (Of b) m r++-}+delay :: MonadIO m => Double -> Stream (Of a) m r -> Stream (Of a) m r+delay seconds = mapM go where+ go a = liftIO (threadDelay (truncate (seconds * 1000000))) >> return a -- --------------- -- drain -- --------------- -{- | Reduce a stream, performing its actions but ignoring its elements.+{- | Reduce a stream, performing its actions but ignoring its elements. + This might just be called @effects@ or @runEffects@. ->>> let stream = do {yield 1; lift (putStrLn "Effect!"); yield 2; lift (putStrLn "Effect!"); return (2^100)} +>>> let effect = lift (putStrLn "Effect!")+>>> let stream = do {yield 1; effect; yield 2; effect; return (2^100)} >>> S.drain stream Effect!@@ -395,7 +477,27 @@ Return r -> return r Delay m -> m >>= loop Step (_ :> rest) -> loop rest+{-#INLINABLE drain #-}+ +{-| 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)++>>> let take' n = S.drained . S.splitAt n+>>> S.print $ concats $ maps (take' 1) $ S.group $ S.each "wwwwarrrrr"+'w'+'a'+'r'++ +-}+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)+{-#INLINE drained #-}+ -- --------------- -- drop -- ---------------@@ -459,19 +561,48 @@ -- enumFrom -- ------ +{-| An infinite stream of enumerable values, starting from a given value.+ @Streaming.Prelude.enumFrom@ is more desirable that @each [x..]@ for + the infinite case, because it has a polymorphic return type.+ +>>> S.print $ S.take 3 $ S.enumFrom 'a'+'a'+'b'+'c'++ Because their return type is polymorphic, @enumFrom@ and @enumFromThen@+ are useful for example with @zip@+ and @zipWith@, which require the same return type in the zipped streams. + With @each [1..]@ the following would be impossible.++>>> rest <- S.print $ S.zip (S.enumFrom 'a') $ S.splitAt 3 $ S.enumFrom 1+('a',1)+('b',2)+('c',3)+>>> S.print $ S.take 3 rest+4+5+6++ Where a final element is specified, as in @each [1..10]@ a special combinator+ is unneeded, since the return type would be @()@ anyway.++-} enumFrom :: (Monad m, Enum n) => n -> Stream (Of n) m r enumFrom = loop where loop !n = Step (n :> loop (succ n)) {-# INLINEABLE enumFrom #-}------ enumFromTo :: (Monad m, Num n, Ord n) => n -> n -> Stream (Of n) m ()--- enumFromTo = loop where--- loop !n m = if n <= m--- then Step (n :> loop (n+1) m)--- else Return ()--- {-# INLINEABLE enumFromTo #-}--- enumFromThen x y = map toEnum [fromEnum x, fromEnum y ..] ++{-| An infinite sequence of enumerable values at a fixed distance, determined+ by the first and second values. See the discussion of 'Streaming.enumFrom'++>>> S.print $ S.take 3 $ S.enumFromThen 100 200+100+200+300++-} enumFromThen:: (Monad m, Enum a) => a -> a -> Stream (Of a) m r enumFromThen first second = Streaming.Prelude.map toEnum (loop _first) where@@ -512,6 +643,7 @@ then return $ Step (a :> loop as) else return $ loop as {-# INLINEABLE filterM #-}+ -- --------------- -- fold -- ---------------@@ -707,8 +839,8 @@ group :: (Monad m, Eq a) => Stream (Of a) m r -> Stream (Stream (Of a) m) m r group = groupBy (==)- + -- --------------- -- iterate -- ---------------@@ -731,9 +863,31 @@ -- --------------- -- length -- ---------------++{-| Run a stream, remembering only its length:++>>> S.length $ S.each [1..10]+10++-} length :: Monad m => Stream (Of a) m () -> m Int length = fold (\n _ -> n + 1) 0 id +{-| Run a stream, keeping its length and return value. As with all folds+ this permits more complex mappings.++>>> 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]+3+3+3+1++-}+ length' :: Monad m => Stream (Of a) m r -> m (Of Int r) length' = fold' (\n _ -> n + 1) 0 id -- ---------------@@ -753,8 +907,11 @@ -- mapFoldable -- --------------- -{-| For each element of a stream, stream a foldable container of elements instead+{-| For each element of a stream, stream a foldable container of elements instead; compare+ 'Pipes.Prelude.mapFoldable'. +> mapFoldable f str = for str (\a -> each (f a))+ >>> S.print $ S.mapFoldable show $ yield 12 '1' '2'@@ -814,7 +971,7 @@ > IOStreams.unfoldM (liftM (either (const Nothing) Just) . next) :: Stream (Of a) IO b -> IO (InputStream a) > Conduit.unfoldM (liftM (either (const Nothing) Just) . next) :: Stream (Of a) m r -> Source a m r - But see 'uncons'+ But see 'uncons', which is better fitted to these @unfoldM@s -} next :: Monad m => Stream (Of a) m r -> m (Either r (a, Stream (Of a) m r)) next = loop where@@ -859,9 +1016,9 @@ -- random -- --------------- -{- An infinite stream of random items +{-| A crude infinite stream of random items, using @System.Random@ -> randoms = liftIO Random.getStdGen >>= unfoldr (return . Right . Random.random)+> randoms = liftIO Random.newStdGen >>= unfoldr (return . Right . Random.random) >>> S.print $ S.take 4 (S.randoms :: Stream (Of Bool) IO ()) True@@ -871,12 +1028,10 @@ -} randoms :: (R.Random a, MonadIO m) => Stream (Of a) m r randoms = do - g <- liftIO $ R.getStdGen+ g <- liftIO $ R.newStdGen unfoldr (return . Right . R.random) g -{- An infinite stream of random items between some bounds--> randomRs limits = liftIO Random.getStdGen >>= unfoldr (return . Right . Random.randomR limits)+{-| 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@@ -926,9 +1081,10 @@ repeatM :: Monad m => m a -> Stream (Of a) m r repeatM ma = loop where- loop = Delay $ do - a <- ma - return (Step (a :> loop))+ loop = do + a <- lift ma + yield a + loop {-# INLINEABLE repeatM #-} -- ---------------@@ -982,6 +1138,16 @@ [3,4] [3,4,5] + A simple way of including the scanned item with the accumulator is to use+ 'Control.Foldl.last'. See also 'Streaming.Prelude.scanned'++>>> let a >< b = (,) <$> a <*> b+>>> S.print $ L.purely S.scan (L.last >< L.sum) $ S.each [1..3]+(Nothing,0)+(Just 1,1)+(Just 2,3)+(Just 3,6)+ -} scan :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> Stream (Of b) m r scan step begin done = loop begin@@ -1025,6 +1191,43 @@ loop x' rest {-# INLINABLE scanM #-} +{- Label each element in a stream with a value accumulated according to a fold.+++>>> S.print $ S.scanned (*) 1 id $ S.each [100,200,300]+(100,100)+(200,20000)+(300,6000000)++>>> S.print $ L.purely S.scanned L.product $ S.each [100,200,300]+(100,100)+(200,20000)+(300,6000000)++-}++data Maybe' a = Just' a | Nothing'++scanned :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> Stream (Of (a,b)) m r+scanned step begin done = loop Nothing' begin+ where+ loop !m !x stream = do + case stream of + Return r -> return r+ Delay mn -> Delay $ liftM (loop m x) mn+ Step (a :> rest) -> do+ case m of + Nothing' -> do + let !acc = step x a+ yield (a, done acc)+ loop (Just' a) acc rest+ Just' _ -> do+ let !acc = done (step x a)+ yield (a, acc) + loop (Just' a) (step x a) rest+{-# INLINABLE scanned #-}++ -- --------------- -- sequence -- ---------------@@ -1086,7 +1289,32 @@ else Return (Step (a :> rest)) {-# INLINEABLE span #-} + +{-| 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 "+hello+world+>>> Prelude.mapM_ Prelude.putStrLn (Prelude.words "hello world ")+hello+world++-}++split :: (Eq a, Monad m) =>+ a -> Stream (Of a) m r -> Stream (Stream (Of a) m) m r+split t = loop where+ loop stream = do+ e <- lift $ next stream+ case e of+ Left r -> Return r+ Right (a, p') -> + if a /= t+ then Step $ fmap loop (yield a >> break (== t) p')+ else loop p'+{-#INLINABLE split #-}+ {-| Split a succession of layers after some number, returning a streaming or -- effectful pair. This function is the same as the 'splitsAt' exported by the -- @Streaming@ module, but since this module is imported qualified, it can @@ -1099,27 +1327,6 @@ splitAt = splitsAt {-# INLINE splitAt #-} --- {-| Split a stream of elements on each occurrence of a value, omitting the value;--- if it appears as the last item in the stream, an empty stream will follow.--- -}--- split :: (Monad m, Eq a) => a -> Stream (Of a) m r -> Stream (Stream (Of a) m) m r--- split a stream = -- loop where--- -- loop stream =--- case stream of--- Return r -> Return r--- Delay m -> Delay (liftM (split a) m)--- Step (a' :> rest) -> if a == a'--- then Step $ do--- e <- lift $ inspect $ split a rest--- case e of--- Left r -> Return (Return r)--- Right b -> b--- else Step $ do--- yield a'--- e <- lift $ inspect $ split a rest--- case e of--- Left r -> Return (Return r)--- Right b -> b -- --------------- -- take@@ -1164,8 +1371,39 @@ Return r -> Return () {-# INLINEABLE takeWhile #-} +{- Break a stream after the designated number of seconds. +>>> rest <- S.print $ S.timed 1 $ S.delay 0.3 $ S.each [1..]+1+2+3+>>> S.print $ S.take 3 rest+4+5+6+++++-}++timed :: MonadIO m => Double -> Stream (Of a) m r -> Stream (Of a) m (Stream (Of a) m r)+timed seconds str = do+ utc <- liftIO getCurrentTime+ loop utc str+ where+ cutoff = fromInteger $ truncate (1000000000 * seconds)+ loop utc str = do+ utc' <- liftIO getCurrentTime+ if diffUTCTime utc' utc > (cutoff / 1000000000)+ then return str+ else case str of+ Return r -> return (return r)+ Delay m -> Delay (liftM (loop utc) m)+ 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@@ -1378,6 +1616,8 @@ loop rest {-# INLINABLE toHandle #-} +{-| Print the elements of a stream as they arise.+-} print :: (MonadIO m, Show a) => Stream (Of a) m r -> m r print = loop where loop stream = case stream of @@ -1393,6 +1633,7 @@ -- {-# INLINABLE seq #-} {-| Write 'String's to 'IO.stdout' using 'putStrLn'; terminates on a broken output pipe+ (compare 'Pipes.Prelude.stdoutLn'). >>> S.stdoutLn $ S.show (S.each [1..3]) 1@@ -1422,21 +1663,11 @@ This does not handle a broken output pipe, but has a polymorphic return value, which makes this possible: ->>> rest <- stdoutLn' $ S.splitAt 3 $ S.show (each [1..5])-1-2-3->>> stdoutLn' rest-4-5-- Or indeed:- >>> rest <- stdoutLn' $ S.show $ S.splitAt 3 (each [1..5]) 1 2 3->>>S.sum rest+>>> S.sum rest 9 -}
streaming.cabal view
@@ -1,5 +1,5 @@ name: streaming-version: 0.1.1.0+version: 0.1.1.1 cabal-version: >=1.10 build-type: Simple synopsis: an elementary streaming prelude and a general monad transformer for streaming applications.@@ -10,6 +10,26 @@ for an explanation. Elementary usage can be divined from the ghci examples in @Streaming.Prelude@ .+ 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:+ .+ > 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.:+ .+ > Conduit.unfoldM Streaming.uncons :: Stream (Of a) m () -> Source m a+ .+ These conversions should never be more expensive than a single @>->@ or @=$=@. Further+ points of comparison are discussed in the + <https://hackage.haskell.org/package/streaming#readme readme>+ below.+ . Note also the <https://hackage.haskell.org/package/streaming-bytestring streaming bytestring> and @@ -49,6 +69,7 @@ , transformers >=0.4 && <0.5 , bytestring , random+ , time default-language: Haskell2010