diff --git a/Streaming/Prelude.hs b/Streaming/Prelude.hs --- a/Streaming/Prelude.hs +++ b/Streaming/Prelude.hs @@ -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 -} diff --git a/streaming.cabal b/streaming.cabal --- a/streaming.cabal +++ b/streaming.cabal @@ -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 + 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 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 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 + + below. + . Note also the and @@ -49,6 +69,7 @@ , transformers >=0.4 && <0.5 , bytestring , random + , time default-language: Haskell2010