5cc6933 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add (!!), insertBy, merge, mergeBy, splitAt, the to Prelude.

5cc6933Pranay Sashank8 years ago

Changelog.md+1

compare with master

unifiedsplit

@@ -2,6 +2,7 @@

3* Add benchmarks to measure composed and iterated operations

4* Performance improvements

5+* Add `(!!)`, `insertBy`, `mergeBy`, `merge`, `splitAt`, `the` to Prelude.

6## 0.5.2

7## 0.5.2

benchmark/Linear.hs+1

@@ -181,6 +181,7 @@ main =

181 , benchIOSink "zip" Ops.zip

182 , benchIOSink "zipM" Ops.zipM

183 ]

184+ , benchIOSink "merge" Ops.merge

184 , bgroup "mixedX4"

185 , bgroup "mixedX4"

185 [ benchIOSink "scan-map" (Ops.scanMap 4)

186 [ benchIOSink "scan-map" (Ops.scanMap 4)

186 , benchIOSink "drop-map" (Ops.dropMap 4)

187 , benchIOSink "drop-map" (Ops.dropMap 4)

benchmark/LinearOps.hs+7−1

@@ -352,8 +352,9 @@ iterateDropWhileTrue = iterateSource (S.dropWhile (<= maxValue)) maxIters

352

353{-# INLINE zip #-}

354{-# INLINE zipM #-}

355+{-# INLINE merge #-}

355{-# INLINE concat #-}

356{-# INLINE concat #-}

356−zip, zipM, concat :: Monad m => Stream m Int -> m ()

357+zip, zipM, concat, merge :: Monad m => Stream m Int -> m ()

357

358

358zip src = do

359zip src = do

359 r <- S.tail src

360 r <- S.tail src

@@ -364,6 +365,11 @@ zipM src = do

364 let src1 = fromJust r

365 let src1 = fromJust r

365 transform (S.zipWithM (curry return) src src1)

366 transform (S.zipWithM (curry return) src src1)

366

367

368+merge src = do

369+ r <- S.tail src

370+ let src1 = fromJust r

371+ transform (S.mergeBy P.compare src src1)

372+

367{-# INLINE zipAsync #-}

373{-# INLINE zipAsync #-}

368{-# INLINE zipAsyncM #-}

374{-# INLINE zipAsyncM #-}

369zipAsync, zipAsyncM :: S.MonadAsync m => Stream m Int -> m ()

375zipAsync, zipAsyncM :: S.MonadAsync m => Stream m Int -> m ()

src/Streamly/Prelude.hs+53−8

@@ -139,6 +139,7 @@ module Streamly.Prelude

139 , or

140 , eqBy

141 , cmpBy

142+ , (!!)

142

143

143 -- Full folds - need to go through all elements

144 -- Full folds - need to go through all elements

144 , length

145 , length

@@ -190,9 +191,11 @@ module Streamly.Prelude

190 , drop

191 , drop

191 , dropWhile

192 , dropWhile

192 , dropWhileM

193 , dropWhileM

194+ , splitAt

193

195

194 -- ** Inserting

196 -- ** Inserting

195 , intersperseM

197 , intersperseM

198+ , insertBy

196

199

197 -- * Reordering

200 -- * Reordering

198 , reverse

201 , reverse

@@ -215,6 +218,13 @@ module Streamly.Prelude

215 , Z.zipAsyncWith

218 , Z.zipAsyncWith

216 , Z.zipAsyncWithM

219 , Z.zipAsyncWithM

217

220

221+ -- * Merging

222+ , mergeBy

223+ , merge

224+

225+ -- * Transformation comprehensions

226+ , the

227+

218 -- * Deprecated

228 -- * Deprecated

219 , K.once

229 , K.once

220 , each

230 , each

@@ -230,7 +240,8 @@ import Prelude

230 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,

240 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,

231 foldl, mapM, mapM_, sequence, all, any, sum, product, elem,

241 foldl, mapM, mapM_, sequence, all, any, sum, product, elem,

232 notElem, maximum, minimum, head, last, tail, length, null,

242 notElem, maximum, minimum, head, last, tail, length, null,

233− reverse, iterate, init, and, or, lookup, foldr1)

243+ reverse, iterate, init, and, or, lookup, foldr1, (!!),

244+ splitAt)

234import qualified Prelude

245import qualified Prelude

235import qualified System.IO as IO

246import qualified System.IO as IO

236

247

@@ -528,7 +539,7 @@ foldr = P.foldr

528-- @since 0.5.0

539-- @since 0.5.0

529{-# INLINE foldr1 #-}

540{-# INLINE foldr1 #-}

530foldr1 :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)

541foldr1 :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)

531−foldr1 = K.foldr1

542+foldr1 f m = S.foldr1 f (toStreamS m)

532

543

533-- | Strict left fold with an extraction function. Like the standard strict

544-- | Strict left fold with an extraction function. Like the standard strict

534-- left fold, but applies a user supplied extraction function (the third

545-- left fold, but applies a user supplied extraction function (the third

@@ -702,12 +713,18 @@ minimum m = S.minimum (toStreamS m)

702maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)

713maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)

703maximum m = S.maximum (toStreamS m)

714maximum m = S.maximum (toStreamS m)

704

715

716+-- | Looks the element at the given index.

717+--

718+{-# INLINE (!!) #-}

719+(!!) :: Monad m => SerialT m a -> Int -> m (Maybe a)

720+m !! i = (toStreamS m) S.!! i

721+

705-- | Looks the given key up, treating the given stream as an association list.

722-- | Looks the given key up, treating the given stream as an association list.

706--

723--

707-- @since 0.5.0

724-- @since 0.5.0

708{-# INLINE lookup #-}

725{-# INLINE lookup #-}

709lookup :: (Monad m, Eq a) => a -> SerialT m (a, b) -> m (Maybe b)

726lookup :: (Monad m, Eq a) => a -> SerialT m (a, b) -> m (Maybe b)

710−lookup = K.lookup

727+lookup a m = S.lookup a (toStreamS m)

711

728

712-- | Returns the first element of the stream satisfying the given predicate,

729-- | Returns the first element of the stream satisfying the given predicate,

713-- if any.

730-- if any.

@@ -715,15 +732,15 @@ lookup = K.lookup

715-- @since 0.5.0

732-- @since 0.5.0

716{-# INLINE find #-}

733{-# INLINE find #-}

717find :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe a)

734find :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe a)

718−find = K.find

735+find p m = S.find p (toStreamS m)

719

736

720-- | Find all the indices where the element in the stream satisfies the given

737-- | Find all the indices where the element in the stream satisfies the given

721-- predicate.

738-- predicate.

722--

739--

723-- @since 0.5.0

740-- @since 0.5.0

724{-# INLINE findIndices #-}

741{-# INLINE findIndices #-}

725−findIndices :: IsStream t => (a -> Bool) -> t m a -> t m Int

742+findIndices :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m Int

726−findIndices = K.findIndices

743+findIndices p m = fromStreamS $ S.findIndices p (toStreamS m)

727

744

728-- | Gives the index of the first stream element satisfying the given

745-- | Gives the index of the first stream element satisfying the given

729-- preficate.

746-- preficate.

@@ -738,7 +755,7 @@ findIndex p = head . findIndices p

738--

755--

739-- @since 0.5.0

756-- @since 0.5.0

740{-# INLINE elemIndices #-}

757{-# INLINE elemIndices #-}

741−elemIndices :: (IsStream t, Eq a) => a -> t m a -> t m Int

758+elemIndices :: (IsStream t, Eq a, Monad m) => a -> t m a -> t m Int

742elemIndices a = findIndices (==a)

759elemIndices a = findIndices (==a)

743

760

744-- | Gives the first index of an element in the stream, which equals the given.

761-- | Gives the first index of an element in the stream, which equals the given.

@@ -889,6 +906,11 @@ dropWhile p m = fromStreamS $ S.dropWhile p $ toStreamS m

889dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a

906dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a

890dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m

907dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m

891

908

909+-- | Split the stream at position 'n'.

910+{-# INLINE splitAt #-}

911+splitAt :: (IsStream t, Monad m) => Int -> t m a -> (t m a, t m a)

912+splitAt n m = (take n m, drop n m)

913+

892------------------------------------------------------------------------------

914------------------------------------------------------------------------------

893-- Transformation by Mapping

915-- Transformation by Mapping

894------------------------------------------------------------------------------

916------------------------------------------------------------------------------

@@ -932,7 +954,7 @@ mapMSerial = Serial.mapM

932-- @since 0.1.0

954-- @since 0.1.0

933{-# INLINE sequence #-}

955{-# INLINE sequence #-}

934sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a

956sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a

935−sequence = K.sequence

957+sequence m = fromStreamS $ S.sequence (toStreamS m)

936

958

937------------------------------------------------------------------------------

959------------------------------------------------------------------------------

938-- Transformation by Map and Filter

960-- Transformation by Map and Filter

@@ -994,6 +1016,11 @@ reverse m = fromStream $ go K.nil (toStream m)

994intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a

1016intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a

995intersperseM = K.intersperseM

1017intersperseM = K.intersperseM

996

1018

1019+{-# INLINE insertBy #-}

1020+insertBy ::

1021+ (IsStream t, Monad m) => (a -> a -> Ordering) -> a -> t m a -> t m a

1022+insertBy cmp x m = fromStreamS $ S.insertBy cmp x (toStreamS m)

1023+

997------------------------------------------------------------------------------

1024------------------------------------------------------------------------------

998-- Zipping

1025-- Zipping

999------------------------------------------------------------------------------

1026------------------------------------------------------------------------------

@@ -1031,3 +1058,21 @@ cmpBy

1031 :: Monad m

1058 :: Monad m

1032 => (a -> b -> Ordering) -> SerialT m a -> SerialT m b -> m Ordering

1059 => (a -> b -> Ordering) -> SerialT m a -> SerialT m b -> m Ordering

1033cmpBy = P.cmpBy

1060cmpBy = P.cmpBy

1061+

1062+-- | Non-overloaded version of merge.

1063+{-# INLINABLE mergeBy #-}

1064+mergeBy ::

1065+ (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a

1066+mergeBy f m1 m2 = fromStreamS $ S.mergeBy f (toStreamS m1) (toStreamS m2)

1067+

1068+-- | Merge two sorted streams.

1069+{-# INLINABLE merge #-}

1070+merge ::

1071+ (IsStream t, Monad m, Ord a) => t m a -> t m a -> t m a

1072+merge = mergeBy compare

1073+

1074+-- | Ensures all the elements are identical and then returns

1075+-- that unique element.

1076+{-# INLINE the #-}

1077+the :: (Eq a, Monad m) => SerialT m a -> m (Maybe a)

1078+the m = S.the (toStreamS m)

src/Streamly/Streams/StreamD.hs+194−2

@@ -67,6 +67,7 @@ module Streamly.Streams.StreamD

67 -- ** General Folds

68 , foldr

69 , foldrM

70+ , foldr1

70 , foldl'

71 , foldl'

71 , foldlM'

72 , foldlM'

@@ -82,6 +83,10 @@ module Streamly.Streams.StreamD

82 , any

83 , any

83 , maximum

84 , maximum

84 , minimum

85 , minimum

86+ , findIndices

87+ , lookup

88+ , find

89+ , (!!)

86 -- ** Map and Fold

91 -- ** Map and Fold

87 , mapM_

92 , mapM_

@@ -110,6 +115,10 @@ module Streamly.Streams.StreamD

110 -- * Mapping

115 -- * Mapping

111 , map

116 , map

112 , mapM

117 , mapM

118+ , sequence

119+

120+ -- * Inserting

121+ , insertBy

113

122

114 -- ** Map and Filter

123 -- ** Map and Filter

115 , mapMaybe

124 , mapMaybe

@@ -122,6 +131,12 @@ module Streamly.Streams.StreamD

122 -- * Comparisions

131 -- * Comparisions

123 , eqBy

132 , eqBy

124 , cmpBy

133 , cmpBy

134+

135+ -- * Merging

136+ , mergeBy

137+

138+ -- * Transformation comprehensions

139+ , the

125 )

140 )

126where

141where

127

142

@@ -130,7 +145,8 @@ import GHC.Types ( SPEC(..) )

130import Prelude

145import Prelude

131 hiding (map, mapM, mapM_, repeat, foldr, last, take, filter,

146 hiding (map, mapM, mapM_, repeat, foldr, last, take, filter,

132 takeWhile, drop, dropWhile, all, any, maximum, minimum, elem,

147 takeWhile, drop, dropWhile, all, any, maximum, minimum, elem,

133− notElem, null, head, tail, zipWith)

148+ notElem, null, head, tail, zipWith, lookup, foldr1, sequence,

149+ (!!))

134

150

135import Streamly.SVar (MonadAsync, State(..), defState, rstState)

151import Streamly.SVar (MonadAsync, State(..), defState, rstState)

136

152

@@ -306,6 +322,14 @@ foldrM f z (Stream step state) = go SPEC state

306foldr :: Monad m => (a -> b -> b) -> b -> Stream m a -> m b

322foldr :: Monad m => (a -> b -> b) -> b -> Stream m a -> m b

307foldr f = foldrM (\a b -> return (f a b))

323foldr f = foldrM (\a b -> return (f a b))

308

324

325+{-# INLINE_NORMAL foldr1 #-}

326+foldr1 :: Monad m => (a -> a -> a) -> Stream m a -> m (Maybe a)

327+foldr1 f m = do

328+ r <- uncons m

329+ case r of

330+ Nothing -> return Nothing

331+ Just (h, t) -> fmap Just (foldr f h t)

332+

309{-# INLINE_NORMAL foldlM' #-}

333{-# INLINE_NORMAL foldlM' #-}

310foldlM' :: Monad m => (b -> a -> m b) -> b -> Stream m a -> m b

334foldlM' :: Monad m => (b -> a -> m b) -> b -> Stream m a -> m b

311foldlM' fstep begin (Stream step state) = go SPEC begin state

335foldlM' fstep begin (Stream step state) = go SPEC begin state

@@ -460,6 +484,53 @@ minimum (Stream step state) = go Nothing state

460 Skip s -> go (Just acc) s

484 Skip s -> go (Just acc) s

461 Stop -> return (Just acc)

485 Stop -> return (Just acc)

462

486

487+{-# INLINE_NORMAL (!!) #-}

488+(!!) :: (Monad m) => Stream m a -> Int -> m (Maybe a)

489+(Stream step state) !! i = go i state

490+ where

491+ go n st = do

492+ r <- step defState st

493+ case r of

494+ Yield x s | n < 0 -> return Nothing

495+ | n == 0 -> return $ Just x

496+ | otherwise -> go (n - 1) s

497+ Skip s -> go n s

498+ Stop -> return Nothing

499+

500+{-# INLINE_NORMAL lookup #-}

501+lookup :: (Monad m, Eq a) => a -> Stream m (a, b) -> m (Maybe b)

502+lookup e (Stream step state) = go state

503+ where

504+ go st = do

505+ r <- step defState st

506+ case r of

507+ Yield (a, b) s -> if e == a then return (Just b) else go s

508+ Skip s -> go s

509+ Stop -> return Nothing

510+

511+{-# INLINE_NORMAL find #-}

512+find :: Monad m => (a -> Bool) -> Stream m a -> m (Maybe a)

513+find p (Stream step state) = go state

514+ where

515+ go st = do

516+ r <- step defState st

517+ case r of

518+ Yield x s -> if p x then return (Just x) else go s

519+ Skip s -> go s

520+ Stop -> return Nothing

521+

522+{-# INLINE_NORMAL findIndices #-}

523+findIndices :: Monad m => (a -> Bool) -> Stream m a -> Stream m Int

524+findIndices p (Stream step state) = Stream step' (state, 0)

525+ where

526+ {-# INLINE_LATE step' #-}

527+ step' gst (st, i) = do

528+ r <- step (rstState gst) st

529+ return $ case r of

530+ Yield x s -> if p x then Yield i (s, i+1) else Skip (s, i+1)

531+ Skip s -> Skip (s, i+1)

532+ Stop -> Stop

533+

463------------------------------------------------------------------------------

534------------------------------------------------------------------------------

464-- Map and Fold

535-- Map and Fold

465------------------------------------------------------------------------------

536------------------------------------------------------------------------------

@@ -660,6 +731,41 @@ mapM f (Stream step state) = Stream step' state

660map :: Monad m => (a -> b) -> Stream m a -> Stream m b

731map :: Monad m => (a -> b) -> Stream m a -> Stream m b

661map f = mapM (return . f)

732map f = mapM (return . f)

662

733

734+{-# INLINE_NORMAL sequence #-}

735+sequence :: Monad m => Stream m (m a) -> Stream m a

736+sequence (Stream step state) = Stream step' state

737+ where

738+ {-# INLINE_LATE step' #-}

739+ step' gst st = do

740+ r <- step (rstState gst) st

741+ case r of

742+ Yield x s -> x >>= \a -> return (Yield a s)

743+ Skip s -> return $ Skip s

744+ Stop -> return Stop

745+

746+------------------------------------------------------------------------------

747+-- Inserting

748+------------------------------------------------------------------------------

749+

750+{-# INLINE_NORMAL insertBy #-}

751+insertBy :: Monad m => (a -> a -> Ordering) -> a -> Stream m a -> Stream m a

752+insertBy cmp a (Stream step state) = Stream step' (state, False)

753+ where

754+ step' gst (st, False) = do

755+ r <- step (rstState gst) st

756+ case r of

757+ Yield x s -> case cmp a x of

758+ GT -> return $ Yield x (s, False)

759+ _ -> return $ Yield a (st, True)

760+ Skip s -> return $ Skip (s, False)

761+ Stop -> return $ Yield a (st, True)

762+ step' gst (st, True) = do

763+ r <- step (rstState gst) st

764+ case r of

765+ Yield x s -> return $ Yield x (s, True)

766+ Skip s -> return $ Skip (s, True)

767+ Stop -> return Stop

768+

663------------------------------------------------------------------------------

769------------------------------------------------------------------------------

664-- Transformation by Map and Filter

770-- Transformation by Map and Filter

665------------------------------------------------------------------------------

771------------------------------------------------------------------------------

@@ -674,7 +780,7 @@ mapMaybeM :: Monad m => (a -> m (Maybe b)) -> Stream m a -> Stream m b

674mapMaybeM f = fmap fromJust . filter isJust . mapM f

780mapMaybeM f = fmap fromJust . filter isJust . mapM f

675

781

676------------------------------------------------------------------------------

782------------------------------------------------------------------------------

677−-- Instances

783+-- Zipping

678------------------------------------------------------------------------------

784------------------------------------------------------------------------------

679

785

680{-# INLINE_NORMAL zipWithM #-}

786{-# INLINE_NORMAL zipWithM #-}

@@ -768,6 +874,92 @@ cmpBy cmp (Stream step1 t1) (Stream step2 t2) = cmp_loop0 SPEC t1 t2

768 Skip s2' -> cmp_null s2'

874 Skip s2' -> cmp_null s2'

769 Stop -> return EQ

875 Stop -> return EQ

770

876

877+------------------------------------------------------------------------------

878+-- Merging

879+------------------------------------------------------------------------------

880+

881+data MergeWithState l r o

882+ = MergeBoth l r

883+ | MergeWithLeft l r o

884+ | MergeWithRight l r o

885+ | MergeNoMore (Either l r)

886+

887+{-# INLINE_NORMAL mergeBy #-}

888+mergeBy :: (Monad m) => (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a

889+mergeBy f (Stream stepa ta) (Stream stepb tb) =

890+ Stream step (MergeBoth ta tb)

891+ {-# INLINE_LATE step #-}

892+ step gst (MergeBoth sa sb) = do

893+ ra <- stepa (rstState gst) sa

894+ rb <- stepb (rstState gst) sb

895+ case (ra, rb) of

896+ (Yield xa sa', Yield xb sb') -> do

897+ case f xa xb of

898+ LT ->

899+ return $ Yield xa (MergeWithRight sa' sb' xb)

900+ _ ->

901+ return $ Yield xb (MergeWithLeft sa' sb' xa)

902+ (Yield xa sa', Stop) -> return $ Yield xa (MergeNoMore (Left sa'))

903+ (Stop, Yield xb sb') -> return $ Yield xb (MergeNoMore (Right sb'))

904+ (Yield xa sa', Skip sb') -> return $ Skip (MergeWithLeft sa' sb' xa)

905+ (Skip sa', Yield xb sb') -> return $ Skip (MergeWithRight sa' sb' xb)

906+ (Skip sa', Skip sb') -> return $ Skip (MergeBoth sa' sb')

907+ (Skip sa', Stop) -> return $ Skip (MergeNoMore (Left sa'))

908+ (Stop, Skip sb') -> return $ Skip (MergeNoMore (Right sb'))

909+ (Stop, Stop) -> return Stop

910+ step gst (MergeWithLeft sa sb x) = do

911+ rb <- stepb (rstState gst) sb

912+ case rb of

913+ Yield xb sb' -> do

914+ case f x xb of

915+ LT -> return $ Yield x (MergeWithRight sa sb' xb)

916+ _ -> return $ Yield xb (MergeWithLeft sa sb' x)

917+ Skip sb' -> return $ Skip (MergeWithLeft sa sb' x)

918+ Stop -> return $ Yield x (MergeNoMore (Left sa))

919+ step gst (MergeWithRight sa sb x) = do

920+ ra <- stepa (rstState gst) sa

921+ case ra of

922+ Yield xa sa' -> do

923+ case f xa x of

924+ LT -> return $ Yield xa (MergeWithRight sa' sb x)

925+ _ -> return $ Yield x (MergeWithLeft sa' sb xa)

926+ Skip sa' -> return $ Skip (MergeWithRight sa' sb x)

927+ Stop -> return $ Yield x (MergeNoMore (Right sb))

928+ step gst (MergeNoMore (Left sa)) = do

929+ ra <- stepa (rstState gst) sa

930+ case ra of

931+ Yield xa sa' -> return $ Yield xa (MergeNoMore (Left sa'))

932+ Skip sa' -> return $ Skip (MergeNoMore (Left sa'))

933+ Stop -> return Stop

934+ step gst (MergeNoMore (Right sb)) = do

935+ rb <- stepb (rstState gst) sb

936+ case rb of

937+ Yield xb sb' -> return $ Yield xb (MergeNoMore (Right sb'))

938+ Skip sb' -> return $ Skip (MergeNoMore (Right sb'))

939+ Stop -> return Stop

940+

941+------------------------------------------------------------------------------

942+-- Transformation comprehensions

943+------------------------------------------------------------------------------

944+

945+{-# INLINE_NORMAL the #-}

946+the :: (Eq a, Monad m) => Stream m a -> m (Maybe a)

947+the (Stream step state) = go state

948+ where

949+ go st = do

950+ r <- step defState st

951+ case r of

952+ Yield x s -> go' x s

953+ Skip s -> go s

954+ Stop -> return Nothing

955+ go' n st = do

956+ r <- step defState st

957+ case r of

958+ Yield x s | x == n -> go' n s

959+ | otherwise -> return Nothing

960+ Skip s -> go' n s

961+ Stop -> return (Just n)

962+

771------------------------------------------------------------------------------

963------------------------------------------------------------------------------

772-- Instances

964-- Instances

773------------------------------------------------------------------------------

965------------------------------------------------------------------------------

src/Streamly/Streams/StreamK.hs+107−1

@@ -94,6 +94,7 @@ module Streamly.Streams.StreamK

94 , findIndices

95 , lookup

96 , find

97+ , (!!)

98 -- ** Map and Fold

99 -- ** Map and Fold

99 , mapM_

100 , mapM_

@@ -121,6 +122,7 @@ module Streamly.Streams.StreamK

121

122

122 -- ** Inserting

123 -- ** Inserting

123 , intersperseM

124 , intersperseM

125+ , insertBy

124

126

125 -- ** Map and Filter

127 -- ** Map and Filter

126 , mapMaybe

128 , mapMaybe

@@ -129,6 +131,12 @@ module Streamly.Streams.StreamK

129 , zipWith

131 , zipWith

130 , zipWithM

132 , zipWithM

131

133

134+ -- ** Merging

135+ , mergeBy

136+

137+ -- ** Transformation comprehensions

138+ , the

139+

132 -- * Semigroup Style Composition

140 -- * Semigroup Style Composition

133 , serial

141 , serial

134

142

@@ -152,7 +160,7 @@ import Prelude

152 hiding (foldl, foldr, last, map, mapM, mapM_, repeat, sequence,

160 hiding (foldl, foldr, last, map, mapM, mapM_, repeat, sequence,

153 take, filter, all, any, takeWhile, drop, dropWhile, minimum,

161 take, filter, all, any, takeWhile, drop, dropWhile, minimum,

154 maximum, elem, notElem, null, head, tail, init, zipWith, lookup,

162 maximum, elem, notElem, null, head, tail, init, zipWith, lookup,

155− foldr1)

163+ foldr1, (!!))

156import qualified Prelude

164import qualified Prelude

157

165

158import Streamly.SVar

166import Streamly.SVar

@@ -726,6 +734,18 @@ maximum m = go Nothing (toStream m)

726 else go (Just res) r

734 else go (Just res) r

727 in unStream m1 defState stop single yieldk

735 in unStream m1 defState stop single yieldk

728

736

737+{-# INLINE (!!) #-}

738+(!!) :: (IsStream t, Monad m) => t m a -> Int -> m (Maybe a)

739+m !! i = go i (toStream m)

740+ where

741+ go n m1 =

742+ let single a | n == 0 = return $ Just a

743+ | otherwise = return Nothing

744+ yieldk a x | n < 0 = return Nothing

745+ | n == 0 = return $ Just a

746+ | otherwise = go (n - 1) x

747+ in unStream m1 defState (return Nothing) single yieldk

748+

729{-# INLINE lookup #-}

749{-# INLINE lookup #-}

730lookup :: (IsStream t, Monad m, Eq a) => a -> t m (a, b) -> m (Maybe b)

750lookup :: (IsStream t, Monad m, Eq a) => a -> t m (a, b) -> m (Maybe b)

731lookup e m = go (toStream m)

751lookup e m = go (toStream m)

@@ -914,6 +934,20 @@ intersperseM a m = fromStream $ prependingStart (toStream m)

914 yieldk i x = unStream (a |: return i |: go x) st stp sng yld

934 yieldk i x = unStream (a |: return i |: go x) st stp sng yld

915 in unStream m2 (rstState st) stp single yieldk

935 in unStream m2 (rstState st) stp single yieldk

916

936

937+{-# INLINE insertBy #-}

938+insertBy :: IsStream t => (a -> a -> Ordering) -> a -> t m a -> t m a

939+insertBy cmp x m = fromStream $ go (toStream m)

940+ where

941+ go m1 = Stream $ \st _ _ yld ->

942+ let single a = case cmp x a of

943+ GT -> yld a (yield x)

944+ _ -> yld x (yield a)

945+ stop = yld x nil

946+ yieldk a r = case cmp x a of

947+ GT -> yld a (go r)

948+ _ -> yld x (a `cons` r)

949+ in unStream m1 (rstState st) stop single yieldk

950+

917-------------------------------------------------------------------------------

951-------------------------------------------------------------------------------

918-- Map and Filter

952-- Map and Filter

919-------------------------------------------------------------------------------

953-------------------------------------------------------------------------------

@@ -971,6 +1005,78 @@ zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

971 yield1 = merge

1005 yield1 = merge

972 unStream mx (rstState st) stp single1 yield1

1006 unStream mx (rstState st) stp single1 yield1

973

1007

1008+------------------------------------------------------------------------------

1009+-- Merging

1010+------------------------------------------------------------------------------

1011+

1012+data MergeWithState l r o

1013+ = MergeBoth l r

1014+ | MergeWithLeft l r o

1015+ | MergeWithRight l r o

1016+

1017+{-# INLINE mergeByS #-}

1018+mergeByS :: (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a

1019+mergeByS cmp m1 m2 = go (MergeBoth m1 m2)

1020+ where

1021+ go (MergeBoth mx my) =

1022+ Stream $ \st stp sng yld -> do

1023+ let yield1 a ra =

1024+ let single2 b = sng (cond a b)

1025+ stop2 = yld a ra

1026+ yield2 b rb =

1027+ case cmp a b of

1028+ LT -> yld a (go (MergeWithRight ra rb b))

1029+ _ -> yld b (go (MergeWithLeft ra rb a))

1030+ in unStream my (rstState st) stop2 single2 yield2

1031+ single1 a = sng a

1032+ stop1 = unStream my (rstState st) stp sng yld

1033+ unStream mx (rstState st) stop1 single1 yield1

1034+ go (MergeWithLeft mx my x) =

1035+ Stream $ \st _ sng yld -> do

1036+ let single2 b = sng (cond x b)

1037+ stop2 = yld x mx

1038+ yield2 b rb =

1039+ case cmp x b of

1040+ LT -> yld x (go (MergeWithRight mx rb b))

1041+ _ -> yld b (go (MergeWithLeft mx rb x))

1042+ unStream my (rstState st) stop2 single2 yield2

1043+ go (MergeWithRight mx my x) =

1044+ Stream $ \st _ sng yld -> do

1045+ let single1 a = sng (cond a x)

1046+ stop1 = yld x my

1047+ yield1 a ra =

1048+ case cmp a x of

1049+ LT -> yld a (go (MergeWithRight ra my x))

1050+ _ -> yld x (go (MergeWithLeft ra my a))

1051+ unStream mx (rstState st) stop1 single1 yield1

1052+ cond x y =

1053+ if cmp x y == LT

1054+ then x

1055+ else y

1056+

1057+{-# INLINABLE mergeBy #-}

1058+mergeBy :: IsStream t => (a -> a -> Ordering) -> t m a -> t m a -> t m a

1059+mergeBy f m1 m2 = fromStream $ mergeByS f (toStream m1) (toStream m2)

1060+

1061+------------------------------------------------------------------------------

1062+-- Transformation comprehensions

1063+------------------------------------------------------------------------------

1064+

1065+{-# INLINE the #-}

1066+the :: (Eq a, IsStream t, Monad m) => t m a -> m (Maybe a)

1067+the m = do

1068+ r <- uncons m

1069+ case r of

1070+ Nothing -> return Nothing

1071+ Just (h, t) -> go h (toStream t)

1072+ where

1073+ go h m1 =

1074+ let single a | h == a = return $ Just h

1075+ | otherwise = return Nothing

1076+ yieldk a r | h == a = go h r

1077+ | otherwise = return Nothing

1078+ in unStream m1 defState (return $ Just h) single yieldk

1079+

974------------------------------------------------------------------------------

1080------------------------------------------------------------------------------

975-- Semigroup

1081-- Semigroup

976------------------------------------------------------------------------------

1082------------------------------------------------------------------------------

test/Prop.hs+29−3

@@ -13,7 +13,7 @@ import Data.Function ((&))

13import Data.IORef (readIORef, modifyIORef, newIORef)

14import Data.List

15 (sort, foldl', scanl', findIndices, findIndex, elemIndices,

16− elemIndex, find, intersperse, foldl1', (\\))

16+ elemIndex, find, insertBy, intersperse, foldl1', (\\))

17import Data.Maybe (mapMaybe)

18import GHC.Word (Word8)

@@ -320,7 +320,7 @@ functorOps constr desc eq t = do

320 prop (desc <> " fmap (+1)") $ transformFromList constr eq (fmap (+1)) $ t . fmap (+1)

321

322transformOps

323− :: IsStream t

323+ :: (IsStream t, Semigroup (t IO Int))

324 => ([Int] -> t IO Int)

325 -> String

326 -> ([Int] -> [Int] -> Bool)

@@ -358,6 +358,12 @@ transformOps constr desc eq t = do

358 prop (desc <> " drop 1") $ transform (drop 1) $ t . S.drop 1

359 prop (desc <> " drop 10") $ transform (drop 10) $ t . S.drop 10

360

361+ prop (desc <> " splitAt maxBound") $

362+ transform (\xs -> let (x, y) = splitAt maxBound xs in x++y) $

363+ t . (\m -> let (x,y) = S.splitAt maxBound m in x<>y)

364+ {-prop (desc <> " splitAt 4") $ transform (splitAt 4) $ t . S.splitAt 4

365+ prop (desc <> " splitAt 0") $ transform (splitAt 0) $ t . S.splitAt 0-}

366+

361 prop (desc <> " dropWhile True") $

367 prop (desc <> " dropWhile True") $

362 transform (dropWhile (const True)) $ t . S.dropWhile (const True)

368 transform (dropWhile (const True)) $ t . S.dropWhile (const True)

363 prop (desc <> " dropWhile False") $

369 prop (desc <> " dropWhile False") $

@@ -371,7 +377,12 @@ transformOps constr desc eq t = do

371 prop (desc <> " elemIndices") $ transform (elemIndices 3) $ t . S.elemIndices 3

377 prop (desc <> " elemIndices") $ transform (elemIndices 3) $ t . S.elemIndices 3

372

378

373 prop (desc <> " intersperseM") $ transform (intersperse 3) $ t . S.intersperseM (return 3)

379 prop (desc <> " intersperseM") $ transform (intersperse 3) $ t . S.intersperseM (return 3)

374−

380+ prop (desc <> " insertBy maxBound") $

381+ transform (insertBy compare maxBound) $ t . S.insertBy compare maxBound

382+ prop (desc <> " insertBy 0") $

383+ transform (insertBy compare 0) $ t . S.insertBy compare 0

384+ prop (desc <> " insertBy 4") $

385+ transform (insertBy compare 4) $ t . S.insertBy compare 4

375

386

376concurrentOps

387concurrentOps

377 :: IsStream t

388 :: IsStream t

@@ -502,6 +513,16 @@ transformCombineOpsOrdered constr desc eq t = do

502wrapMaybe :: ([a1] -> a2) -> [a1] -> Maybe a2

513wrapMaybe :: ([a1] -> a2) -> [a1] -> Maybe a2

503wrapMaybe f x = if null x then Nothing else Just (f x)

514wrapMaybe f x = if null x then Nothing else Just (f x)

504

515

516+wrapOutOfBounds :: ([a1] -> Int -> a2) -> Int -> [a1] -> Maybe a2

517+wrapOutOfBounds f i x | null x = Nothing

518+ | i >= length x = Nothing

519+ | otherwise = Just (f x i)

520+

521+wrapThe :: Eq a => [a] -> Maybe a

522+wrapThe (x:xs) | all (x ==) xs = Just x

523+ | otherwise = Nothing

524+wrapThe [] = Nothing

525+

505eliminationOps

526eliminationOps

506 :: ([Int] -> t IO Int)

527 :: ([Int] -> t IO Int)

507 -> String

528 -> String

@@ -532,11 +553,16 @@ eliminationOps constr desc t = do

532 prop (desc <> " findIndex") $ eliminateOp constr (findIndex odd) $ S.findIndex odd . t

553 prop (desc <> " findIndex") $ eliminateOp constr (findIndex odd) $ S.findIndex odd . t

533 prop (desc <> " elemIndex") $ eliminateOp constr (elemIndex 3) $ S.elemIndex 3 . t

554 prop (desc <> " elemIndex") $ eliminateOp constr (elemIndex 3) $ S.elemIndex 3 . t

534

555

556+ prop (desc <> " !!") $ eliminateOp constr (wrapOutOfBounds (!!) 5) $ (S.!! 5) . t

557+ prop (desc <> " !!") $ eliminateOp constr (wrapOutOfBounds (!!) 0) $ (S.!! 0) . t

558+

535 prop (desc <> " find") $ eliminateOp constr (find even) $ S.find even . t

559 prop (desc <> " find") $ eliminateOp constr (find even) $ S.find even . t

536 prop (desc <> " lookup") $

560 prop (desc <> " lookup") $

537 eliminateOp constr (lookup 3 . flip zip [1..]) $

561 eliminateOp constr (lookup 3 . flip zip [1..]) $

538 S.lookup 3 . S.zipWith (\a b -> (b, a)) (S.fromList [(1::Int)..]) . t

562 S.lookup 3 . S.zipWith (\a b -> (b, a)) (S.fromList [(1::Int)..]) . t

539

563

564+ prop (desc <> " the") $ eliminateOp constr wrapThe $ S.the . t

565+

540-- head/tail/last may depend on the order in case of parallel streams

566-- head/tail/last may depend on the order in case of parallel streams

541-- so we test these only for serial streams.

567-- so we test these only for serial streams.

542serialEliminationOps

568serialEliminationOps