4c00da9 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Merge pull request #76 from subbyte/master

4c00da9Harendra Kumar8 years ago

Changelog.md+1

compare with master

unifiedsplit

@@ -24,6 +24,7 @@

24 concurrently when used at appropriate stream types.

25* Added concurrent function application operators to run stages of a

26 stream processing function application pipeline concurrently.

27+* Added two functions `mapMaybe` and `mapMaybeM`.

28## 0.2.0

29## 0.2.0

src/Streamly/Prelude.hs+29

@@ -100,12 +100,14 @@ module Streamly.Prelude

100 , takeWhile

101 , drop

102 , dropWhile

103+ , mapMaybe

103

104

104 -- * Reordering

105 -- * Reordering

105 , reverse

106 , reverse

106

107

107 -- * Mapping

108 -- * Mapping

108 , mapM

109 , mapM

110+ , mapMaybeM

109 , sequence

111 , sequence

110

112

111 -- * Zipping

113 -- * Zipping

@@ -129,6 +131,7 @@ where

129import Control.Monad (void)

131import Control.Monad (void)

130import Control.Monad.IO.Class (MonadIO(..))

132import Control.Monad.IO.Class (MonadIO(..))

131import Data.Semigroup (Semigroup(..))

133import Data.Semigroup (Semigroup(..))

134+import Data.Maybe (isJust, fromJust)

132import Prelude hiding (filter, drop, dropWhile, take,

135import Prelude hiding (filter, drop, dropWhile, take,

133 takeWhile, zipWith, foldr, foldl,

136 takeWhile, zipWith, foldr, foldl,

134 mapM, mapM_, sequence, all, any,

137 mapM, mapM_, sequence, all, any,

@@ -577,6 +580,24 @@ dropWhile p m = fromStream $ go (toStream m)

577 | otherwise = yld a r

580 | otherwise = yld a r

578 in (S.runStream m1) ctx stp single yield

581 in (S.runStream m1) ctx stp single yield

579

582

583+-- | Map + filter: compute a stream of b from a stream of a and filter out

584+-- elements that are mapped to Nothing. It is similar to 'mapMaybe' in

585+-- 'Data.Maybe' but applies to streams instead of list.

586+--

587+-- @since 0.2.1

588+{-# INLINE mapMaybe #-}

589+mapMaybe :: (IsStream t) => (a -> Maybe b) -> t m a -> t m b

590+mapMaybe f m = go (toStream m)

591+ where

592+ go m1 = fromStream $ Stream $ \_ stp sng yld ->

593+ let single a = case f a of

594+ Just b -> sng b

595+ Nothing -> stp

596+ yield a r = case f a of

597+ Just b -> yld b (toStream $ go r)

598+ Nothing -> (S.runStream r) Nothing stp single yield

599+ in (S.runStream m1) Nothing stp single yield

600+

580-- | Determine whether all elements of a stream satisfy a predicate.

601-- | Determine whether all elements of a stream satisfy a predicate.

581--

602--

582-- @since 0.1.0

603-- @since 0.1.0

@@ -757,6 +778,14 @@ mapM f m = go (toStream m)

757 yield a r = S.runStream (toStream (f a |: (go r))) svr stp sng yld

778 yield a r = S.runStream (toStream (f a |: (go r))) svr stp sng yld

758 in (S.runStream m1) Nothing stp single yield

779 in (S.runStream m1) Nothing stp single yield

759

780

781+-- | Monadic version mapMaybe.

782+--

783+-- @since 0.2.1

784+{-# INLINE mapMaybeM #-}

785+mapMaybeM :: (IsStream t, MonadAsync m, Functor (t m))

786+ => (a -> m (Maybe b)) -> t m a -> t m b

787+mapMaybeM f = fmap fromJust . filter isJust . mapM f

788+

760-- XXX this can utilize parallel mapping if we implement it as runStream . mapM

789-- XXX this can utilize parallel mapping if we implement it as runStream . mapM

761-- | Apply a monadic action to each element of the stream and discard the

790-- | Apply a monadic action to each element of the stream and discard the

762-- output of the action.

791-- output of the action.

test/Prop.hs+4

@@ -9,6 +9,7 @@ import Control.Applicative (ZipList(..))

9import Control.Concurrent (MVar, takeMVar, putMVar, newEmptyMVar)

10import Control.Monad (replicateM, replicateM_)

11import Data.List (sort, foldl', scanl')

12+import Data.Maybe (mapMaybe)

12import GHC.Word (Word8)

13import GHC.Word (Word8)

14import Test.Hspec.QuickCheck (prop)

15import Test.Hspec.QuickCheck (prop)

@@ -295,6 +296,9 @@ transformOps constr desc t eq = do

295 prop (desc ++ " takeWhile > 0") $

296 prop (desc ++ " takeWhile > 0") $

296 transform (takeWhile (> 0)) $ t . (A.takeWhile (> 0))

297 transform (takeWhile (> 0)) $ t . (A.takeWhile (> 0))

297

298

299+ let f x = if odd x then Just (x + 100) else Nothing

300+ prop (desc ++ " mapMaybe") $ transform (mapMaybe f) $ t . (A.mapMaybe f)

301+

298 prop (desc ++ " drop maxBound") $

302 prop (desc ++ " drop maxBound") $

299 transform (drop maxBound) $ t . (A.drop maxBound)

303 transform (drop maxBound) $ t . (A.drop maxBound)

300 prop (desc ++ " drop 0") $ transform (drop 0) $ t . (A.drop 0)

304 prop (desc ++ " drop 0") $ transform (drop 0) $ t . (A.drop 0)