4b49390 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Fix uniqBy and rollingMap

4b49390Ranjeet Kumar Ranjan5 years ago

benchmark/Streamly/Benchmark/Prelude/Serial/Transformation.hs+1−1

compare with master

unifiedsplit

@@ -151,7 +151,7 @@ pollCounts n =

151

152 where

153

154− f = Internal.rollingMap (-) . Internal.delayPost 1

154+ f = Internal.rollingMap2 (-) . Internal.delayPost 1

155

156{-# INLINE timestamped #-}

157timestamped :: (S.MonadAsync m) => SerialT m Int -> m ()

core/src/Streamly/Internal/Data/Stream/StreamD/Transform.hs+23−14

@@ -113,6 +113,7 @@ module Streamly.Internal.Data.Stream.StreamD.Transform

113 -- | Map using the previous element.

114 , rollingMap

115 , rollingMapM

116+ , rollingMap2

116

117

117 -- * Maybe Streams

118 -- * Maybe Streams

118 , mapMaybe

119 , mapMaybe

@@ -1200,32 +1201,36 @@ slicesBy p (Stream step1 state1) = Stream step (Just (state1, 0, 0))

1200-- Rolling map

1201-- Rolling map

1201------------------------------------------------------------------------------

1202------------------------------------------------------------------------------

1202

1203

1203−data RollingMapState s a = RollingMapInit s | RollingMapGo s a

1204+data RollingMapState s a = RollingMapGo s a

1204

1205

1205{-# INLINE rollingMapM #-}

1206{-# INLINE rollingMapM #-}

1206−rollingMapM :: Monad m => (a -> a -> m b) -> Stream m a -> Stream m b

1207+rollingMapM :: Monad m => (Maybe a -> a -> m b) -> Stream m a -> Stream m b

1207−rollingMapM f (Stream step1 state1) = Stream step (RollingMapInit state1)

1208+rollingMapM f (Stream step1 state1) = Stream step (RollingMapGo state1 Nothing)

1209+

1208 where

1210 where

1209− step gst (RollingMapInit st) = do

1210− r <- step1 (adaptState gst) st

1211− return $ case r of

1212− Yield x s -> Skip $ RollingMapGo s x

1213− Skip s -> Skip $ RollingMapInit s

1214− Stop -> Stop

1215

1211

1216− step gst (RollingMapGo s1 x1) = do

1212+ step gst (RollingMapGo s1 curr) = do

1217 r <- step1 (adaptState gst) s1

1213 r <- step1 (adaptState gst) s1

1218 case r of

1214 case r of

1219 Yield x s -> do

1215 Yield x s -> do

1220− !res <- f x x1

1216+ !res <- f curr x

1221− return $ Yield res $ RollingMapGo s x

1217+ return $ Yield res $ RollingMapGo s (Just x)

1222− Skip s -> return $ Skip $ RollingMapGo s x1

1218+ Skip s -> return $ Skip $ RollingMapGo s curr

1223 Stop -> return Stop

1219 Stop -> return Stop

1224

1220

1225{-# INLINE rollingMap #-}

1221{-# INLINE rollingMap #-}

1226−rollingMap :: Monad m => (a -> a -> b) -> Stream m a -> Stream m b

1222+rollingMap :: Monad m => (Maybe a -> a -> b) -> Stream m a -> Stream m b

1227rollingMap f = rollingMapM (\x y -> return $ f x y)

1223rollingMap f = rollingMapM (\x y -> return $ f x y)

1228

1224

1225+{-# INLINE rollingMap2 #-}

1226+rollingMap2 :: Monad m => (a -> a -> b) -> Stream m a -> Stream m b

1227+rollingMap2 f = catMaybes . rollingMap g

1228+

1229+ where

1230+

1231+ g Nothing _ = Nothing

1232+ g (Just x) y = Just (f x y)

1233+

1229------------------------------------------------------------------------------

1234------------------------------------------------------------------------------

1230-- Maybe Streams

1235-- Maybe Streams

1231------------------------------------------------------------------------------

1236------------------------------------------------------------------------------

@@ -1238,3 +1243,7 @@ mapMaybe f = fmap fromJust . filter isJust . map f

1238{-# INLINE_NORMAL mapMaybeM #-}

1243{-# INLINE_NORMAL mapMaybeM #-}

1239mapMaybeM :: Monad m => (a -> m (Maybe b)) -> Stream m a -> Stream m b

1244mapMaybeM :: Monad m => (a -> m (Maybe b)) -> Stream m a -> Stream m b

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

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

1246+

1247+{-# INLINE catMaybes #-}

1248+catMaybes :: (Monad m) => Stream m (Maybe a) -> Stream m a

1249+catMaybes = fmap fromJust . filter isJust

src/Streamly/Internal/Data/Stream/IsStream/Transform.hs+15−4

@@ -156,6 +156,7 @@ module Streamly.Internal.Data.Stream.IsStream.Transform

156 -- | Map using the previous element.

157 , rollingMapM

158 , rollingMap

159+ , rollingMap2

159

160

160 -- * Maybe Streams

161 -- * Maybe Streams

161 -- Move these to Streamly.Data.Maybe.Stream?

162 -- Move these to Streamly.Data.Maybe.Stream?

@@ -921,8 +922,14 @@ filterM p m = fromStreamD $ D.filterM p $ toStreamD m

921{-# INLINE uniqBy #-}

922{-# INLINE uniqBy #-}

922uniqBy :: (IsStream t, Monad m, Functor (t m)) =>

923uniqBy :: (IsStream t, Monad m, Functor (t m)) =>

923 (a -> a -> Bool) -> t m a -> t m a

924 (a -> a -> Bool) -> t m a -> t m a

924−uniqBy eq =

925+uniqBy eq = catMaybes . rollingMap f

926+

927+ where

928+

929+ f pre curr =

930+ case pre of

931+ Nothing -> Just curr

925− catMaybes . rollingMap (\x y -> if x `eq` y then Nothing else Just y)

932+ Just x -> if x `eq` curr then Nothing else Just curr

926

933

927-- | Drop repeated elements that are adjacent to each other.

934-- | Drop repeated elements that are adjacent to each other.

928--

935--

@@ -1519,7 +1526,7 @@ elemIndices a = findIndices (== a)

1519-- /Pre-release/

1526-- /Pre-release/

1520--

1527--

1521{-# INLINE rollingMap #-}

1528{-# INLINE rollingMap #-}

1522−rollingMap :: (IsStream t, Monad m) => (a -> a -> b) -> t m a -> t m b

1529+rollingMap :: (IsStream t, Monad m) => (Maybe a -> a -> b) -> t m a -> t m b

1523rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m

1530rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m

1524

1531

1525-- | Like 'rollingMap' but with an effectful map function.

1532-- | Like 'rollingMap' but with an effectful map function.

@@ -1527,9 +1534,13 @@ rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m

1527-- /Pre-release/

1534-- /Pre-release/

1528--

1535--

1529{-# INLINE rollingMapM #-}

1536{-# INLINE rollingMapM #-}

1530−rollingMapM :: (IsStream t, Monad m) => (a -> a -> m b) -> t m a -> t m b

1537+rollingMapM :: (IsStream t, Monad m) => (Maybe a -> a -> m b) -> t m a -> t m b

1531rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m

1538rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m

1532

1539

1540+{-# INLINE rollingMap2 #-}

1541+rollingMap2 :: (IsStream t, Monad m) => (a -> a -> b) -> t m a -> t m b

1542+rollingMap2 f m = fromStreamD $ D.rollingMap2 f $ toStreamD m

1543+

1533------------------------------------------------------------------------------

1544------------------------------------------------------------------------------

1534-- Maybe Streams

1545-- Maybe Streams

1535------------------------------------------------------------------------------

1546------------------------------------------------------------------------------

test/Streamly/Test/Prelude/Top.hs+5−5

@@ -178,10 +178,10 @@ intersectBy ::

178 )

179 -> (Int -> Int -> a)

180 -> Property

181−intersectBy _srt intersectFunc cmp =

181+intersectBy srt intersectFunc cmp =

182 forAll (listOf (chooseInt (min_value, max_value))) $ \ls0 ->

183 forAll (listOf (chooseInt (min_value, max_value))) $ \ls1 ->

184− monadicIO $ action (sort ls0) (sort ls1)

184+ monadicIO $ action (srt ls0) (srt ls1)

185

186 where

187

@@ -194,7 +194,7 @@ intersectBy _srt intersectFunc cmp =

194 (S.fromList ls0)

195 (S.fromList ls1)

196 let v2 = ls0 `intersect` ls1

197− assert (v1 == sort v2)

197+ assert (sort v1 == sort v2)

198

199-------------------------------------------------------------------------------

200-- Main

@@ -215,7 +215,7 @@ main = hspec $ do

215 prop "joinLeft" Main.joinLeft

216 prop "joinLeftMap" Main.joinLeftMap

217 -- intersect

218− -- XXX currently API is broken https://github.com/composewell/streamly/issues/1471

218+ prop "intersectBy"

219− --prop "intersectBy" (intersectBy id Top.intersectBy (==))

219+ (intersectBy id Top.intersectBy (==))

220 prop "intersectBySorted"

221 (intersectBy sort Top.intersectBySorted compare)