50232af · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

rename/reorg/remove some fold APIs

50232afHarendra Kumar7 years ago

benchmark/FileIO.hs+8−8

compare with master

unifiedsplit

@@ -134,20 +134,20 @@ main = do

134 File.write outh (File.read inh)

135 ]

136 , bgroup "grouping"

137− [ mkBench "groupsOf 1 (toArray)" href $ do

137+ [ mkBench "chunksOf 1 (toArray)" href $ do

138 Handles inh _ <- readIORef href

139− S.length $ FL.groupsOf fileSize (A.toArrayN fileSize)

139+ S.length $ FL.chunksOf fileSize (A.toArrayN fileSize)

140 (File.read inh)

141

142− , mkBench "groupsOf 1" href $ do

142+ , mkBench "chunksOf 1" href $ do

143 Handles inh _ <- readIORef href

144− S.length $ FL.groupsOf 1 FL.drain (File.read inh)

144+ S.length $ FL.chunksOf 1 FL.drain (File.read inh)

145− , mkBench "groupsOf 10" href $ do

145+ , mkBench "chunksOf 10" href $ do

146 Handles inh _ <- readIORef href

147− S.length $ FL.groupsOf 10 FL.drain (File.read inh)

147+ S.length $ FL.chunksOf 10 FL.drain (File.read inh)

148− , mkBench "groupsOf 1000" href $ do

148+ , mkBench "chunksOf 1000" href $ do

149 Handles inh _ <- readIORef href

150− S.length $ FL.groupsOf 1000 FL.drain (File.read inh)

150+ S.length $ FL.chunksOf 1000 FL.drain (File.read inh)

151 ]

152

153 , let lf = fromIntegral (ord '\n')

benchmark/NanoBenchmarks.hs+1−1

@@ -9,7 +9,7 @@

9import Streamly (SerialT)

10import Streamly.SVar (MonadAsync)

12−import qualified Streamly.Array as A

12+import qualified Streamly.Mem.Array as A

13import qualified Streamly.FileSystem.File as File

14import qualified Streamly.Fold as FL

15import qualified Streamly.Prelude as S

src/Streamly/Fold.hs+87−88

@@ -146,16 +146,10 @@ module Streamly.Fold

146 -- ** Filtering

147 , lfilter

148 , lfilterM

149− , ltake

150− , ltakeWhile

151− {-

152− , ltakeWhileM

153− , ldrop

154− , ldropWhile

155− , ldropWhileM

156− , ldeleteBy

149+ -- , ldeleteBy

157− , luniq

150+ -- , luniq

158

151

152+ {-

159 -- ** Mapping Filters

153 -- ** Mapping Filters

160 , lmapMaybe

154 , lmapMaybe

161 , lmapMaybeM

155 , lmapMaybeM

@@ -174,23 +168,36 @@ module Streamly.Fold

174 , lreverse

168 , lreverse

175 -}

169 -}

176

170

177− -- * Stream Splitting Operations

171+ -- * Parsing

178− -- ** Spanning

172+ -- ** Trimming

179− -- | Spanning splits the input into two groups and applies two different

173+ , ltake

180− -- folds on each group.

174+ -- , lrunFor -- time

175+ , ltakeWhile

176+ {-

177+ , ltakeWhileM

178+ , ldrop

179+ , ldropWhile

180+ , ldropWhileM

181+ -}

181

182

182− -- Element unaware spanning

183+ -- ** Breaking

184+

185+ -- By chunks

183 , splitAt

186 , splitAt

187+ -- , splitIn -- time

184

188

185− -- Element aware spanning

189+ -- By elements

186 , span

190 , span

187− , break

191+ , break -- breakPre

192+ -- , breakPost

193+ -- , breakOn

188 -- , spanBy

194 -- , spanBy

189 -- , spanRollingBy

195 -- , spanRollingBy

190

196

191− -- ** Grouping

197+ -- By sequences

192− -- | Grouping splits the stream into groups and applies the supplied fold

198+ -- breakOnSeq

193− -- on each group.

199+

200+ -- ** Splitting

194

201

195 -- In imperative terms grouped folding can be considered as a nested loop

202 -- In imperative terms grouped folding can be considered as a nested loop

196 -- where we loop over the stream to group elements and then loop over

203 -- where we loop over the stream to group elements and then loop over

@@ -205,23 +212,17 @@ module Streamly.Fold

205 -- groups in memory and then map a fold on it to fold the groups. This kind

212 -- groups in memory and then map a fold on it to fold the groups. This kind

206 -- of grouping and folding would not work well when the group size is big.

213 -- of grouping and folding would not work well when the group size is big.

207

214

208− -- Element unaware grouping

215+ -- *** By Chunks

209− , groupsOf

216+ , chunksOf

210− , intervalsOf

217+ , spellsOf

211

218

212− -- Element aware grouping

219+ -- *** By Elements

213− , groups

214− , groupsBy

215− , groupsRollingBy

216− , grouped -- XXX experimental

217−

218− -- ** Splitting by an Element

219 , splitBy

220 , splitBy

220 , splitSuffixBy

221 , splitSuffixBy

221 -- , splitPrefixBy

222 -- , splitPrefixBy

222 , wordsBy

223 , wordsBy

223

224

224− -- ** Splitting on a Sequence

225+ -- *** By Sequences

225 , splitOn

226 , splitOn

226 , splitSuffixOn

227 , splitSuffixOn

227 -- , splitPrefixOn

228 -- , splitPrefixOn

@@ -237,7 +238,12 @@ module Streamly.Fold

237 -- , splitSuffixOnAny

238 -- , splitSuffixOnAny

238 -- , splitPrefixOnAny

239 -- , splitPrefixOnAny

239

240

241+ -- ** Grouping

242+ , groups

243+ , groupsBy

244+ , groupsRollingBy

245+

240− -- ** Distributing

246+ -- * Distributing

241 -- |

247 -- |

242 -- The 'Applicative' instance of a distributing 'Fold' distributes one copy

248 -- The 'Applicative' instance of a distributing 'Fold' distributes one copy

243 -- of the stream to each fold and combines the results using a function.

249 -- of the stream to each fold and combines the results using a function.

@@ -271,7 +277,7 @@ module Streamly.Fold

271 , tee

277 , tee

272 , distribute

278 , distribute

273

279

274− -- ** Partitioning

280+ -- * Partitioning

275 -- |

281 -- |

276 -- Direct items in the input stream to different folds using a function to

282 -- Direct items in the input stream to different folds using a function to

277 -- select the fold. This is useful to demultiplex the input stream.

283 -- select the fold. This is useful to demultiplex the input stream.

@@ -280,21 +286,15 @@ module Streamly.Fold

280 , demux_

286 , demux_

281 , classify

287 , classify

282

288

283− -- ** Unzipping

289+ -- * Unzipping

284 , unzip

290 , unzip

285 , unzipWith

291 , unzipWith

286 , unzipWithM

292 , unzipWithM

287

293

288− -- ** Nested Folds

294+ -- * Nested Folds

289 -- , concatMap

295 -- , concatMap

290− -- , groupsOf

296+ -- , chunksOf

291 , duplicate -- experimental

297 , duplicate -- experimental

292−

293− {-

294− -- * Splitter scans

295− -- | Scans that can be used to split and fold a stream using 'grouped'.

296− , _newline

297− -}

298 )

299where

300

@@ -303,7 +303,7 @@ import Control.Monad (void)

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

304import Data.Functor.Identity (Identity)

305import Data.Map.Strict (Map)

306−import Data.Maybe (fromJust, isJust)

306+import Data.Maybe (fromJust, isJust, isNothing)

307

308import Foreign.Storable (Storable(..))

309import Prelude

@@ -970,6 +970,9 @@ ltakeWhile predicate (Fold step initial done) = Fold step' initial' done'

970-- >>> splitAt_ 4 [1,2,3]

971-- > ([1,2,3],[])

972--

973+-- This can be considered as a two-fold version of 'ltake' where we take both

974+-- the segments instead of discarding the leftover.

975+--

973-- @since 0.7.0

976-- @since 0.7.0

974{-# INLINE splitAt #-}

977{-# INLINE splitAt #-}

975splitAt

978splitAt

@@ -997,39 +1000,43 @@ splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =

997-- | Group the input stream into groups of @n@ elements each and then fold each

1000-- | Group the input stream into groups of @n@ elements each and then fold each

998-- group using the provided fold function.

1001-- group using the provided fold function.

999--

1002--

1000−-- >> S.toList $ FL.groupsOf 2 FL.sum (S.enumerateFromTo 1 10)

1003+-- >> S.toList $ FL.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)

1001-- > [3,7,11,15,19]

1004-- > [3,7,11,15,19]

1002--

1005--

1006+-- This can be considered as an n-fold version of 'ltake' where we apply

1007+-- 'ltake' repeatedly on the leftover stream until the stream exhausts.

1008+--

1003-- @since 0.7.0

1009-- @since 0.7.0

1004−{-# INLINE groupsOf #-}

1010+{-# INLINE chunksOf #-}

1005−groupsOf

1011+chunksOf

1006 :: (IsStream t, Monad m)

1012 :: (IsStream t, Monad m)

1007 => Int -> Fold m a b -> t m a -> t m b

1013 => Int -> Fold m a b -> t m a -> t m b

1008−groupsOf n f m = D.fromStreamD $ D.groupsOf n f (D.toStreamD m)

1014+chunksOf n f m = D.fromStreamD $ D.groupsOf n f (D.toStreamD m)

1009

1015

1010-- | Transform a fold from a pure input to a 'Maybe' input, consuming only

1016-- | Transform a fold from a pure input to a 'Maybe' input, consuming only

1011-- 'Just' values.

1017-- 'Just' values.

1012lcatMaybes :: Monad m => Fold m a b -> Fold m (Maybe a) b

1018lcatMaybes :: Monad m => Fold m a b -> Fold m (Maybe a) b

1013lcatMaybes = lfilter isJust . lmap fromJust

1019lcatMaybes = lfilter isJust . lmap fromJust

1014

1020

1021+-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.

1015-- XXX add this example after fixing the serial stream rate control

1022-- XXX add this example after fixing the serial stream rate control

1016−-- >>> S.toList $ S.take 5 $ intervalsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1

1023+-- >>> S.toList $ S.take 5 $ spellsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1

1017-- > [3,7,11,15,19]

1024-- > [3,7,11,15,19]

1018--

1025--

1019-- | Group the input stream into windows of @n@ second each and then fold each

1026-- | Group the input stream into windows of @n@ second each and then fold each

1020-- group using the provided fold function.

1027-- group using the provided fold function.

1021--

1028--

1022-- @since 0.7.0

1029-- @since 0.7.0

1023−{-# INLINE intervalsOf #-}

1030+{-# INLINE spellsOf #-}

1024−intervalsOf

1031+spellsOf

1025 :: (IsStream t, MonadAsync m)

1032 :: (IsStream t, MonadAsync m)

1026 => Double -> Fold m a b -> t m a -> t m b

1033 => Double -> Fold m a b -> t m a -> t m b

1027−intervalsOf n f m = grouped (lcatMaybes f) s

1034+spellsOf n f m = splitSuffixBy' isNothing (lcatMaybes f) s

1028 where

1035 where

1029− s = S.map (\x -> (Just x, False)) m `parallel` S.repeatM timeout

1036+ s = S.map Just m `parallel` S.repeatM timeout

1030 timeout = do

1037 timeout = do

1031 liftIO $ threadDelay (round $ n * 1000000)

1038 liftIO $ threadDelay (round $ n * 1000000)

1032− return (Nothing, True)

1039+ return Nothing

1033

1040

1034------------------------------------------------------------------------------

1041------------------------------------------------------------------------------

1035-- Element Aware APIs

1042-- Element Aware APIs

@@ -1039,18 +1046,17 @@ intervalsOf n f m = grouped (lcatMaybes f) s

1039-- Binary APIs

1046-- Binary APIs

1040------------------------------------------------------------------------------

1047------------------------------------------------------------------------------

1041

1048

1042−{-

1043-- | Break the input stream into two groups, the first group takes the input as

1049-- | Break the input stream into two groups, the first group takes the input as

1044-- long as the predicate applied to the first element of the stream and next

1050-- long as the predicate applied to the first element of the stream and next

1045-- input element holds 'True', the second group takes the rest of the input.

1051-- input element holds 'True', the second group takes the rest of the input.

1046--

1052--

1047−spanBy

1053+_spanBy

1048 :: Monad m

1054 :: Monad m

1049 => (a -> a -> Bool)

1055 => (a -> a -> Bool)

1050 -> Fold m a b

1056 -> Fold m a b

1051 -> Fold m a c

1057 -> Fold m a c

1052 -> Fold m a (b, c)

1058 -> Fold m a (b, c)

1053−spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =

1059+_spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =

1054 Fold step init extract

1060 Fold step init extract

1055

1061

1056 where

1062 where

@@ -1071,7 +1077,6 @@ spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =

1071 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))

1077 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))

1072

1078

1073 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b

1079 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b

1074−-}

1075

1080

1076-- | Span as long as the predicate is 'True'. @span p f1 f2@ composes folds

1081-- | Span as long as the predicate is 'True'. @span p f1 f2@ composes folds

1077-- @f1@ and @f2@ such that the composed fold continues sending the input to

1082-- @f1@ and @f2@ such that the composed fold continues sending the input to

@@ -1089,6 +1094,9 @@ spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =

1089-- >>> span_ (< 4) [1,2,3]

1094-- >>> span_ (< 4) [1,2,3]

1090-- > ([1,2,3],[])

1095-- > ([1,2,3],[])

1091--

1096--

1097+-- This can be considered as a two-fold version of 'ltakeWhile' where we take

1098+-- both the segments instead of discarding the leftover.

1099+--

1092-- @since 0.7.0

1100-- @since 0.7.0

1093{-# INLINE span #-}

1101{-# INLINE span #-}

1094span

1102span

@@ -1117,6 +1125,8 @@ span p (Fold stepL initialL extractL) (Fold stepR initialR extractR) =

1117-- folds @f1@ and @f2@ such that @f1@ stops receiving input as soon as the

1125-- folds @f1@ and @f2@ such that @f1@ stops receiving input as soon as the

1118-- predicate @p@ becomes 'True'. The rest of the input is sent to @f2@.

1126-- predicate @p@ becomes 'True'. The rest of the input is sent to @f2@.

1119--

1127--

1128+-- This is the binary version of 'splitBy'.

1129+--

1120-- > let break_ p xs = FL.foldl' (FL.break p FL.toList FL.toList) $ S.fromList xs

1130-- > let break_ p xs = FL.foldl' (FL.break p FL.toList FL.toList) $ S.fromList xs

1121--

1131--

1122-- >>> break_ (< 1) [3,2,1]

1132-- >>> break_ (< 1) [3,2,1]

@@ -1138,17 +1148,16 @@ break

1138 -> Fold m a (b, c)

1148 -> Fold m a (b, c)

1139break p = span (not . p)

1149break p = span (not . p)

1140

1150

1141−{-

1142-- | Like 'spanBy' but applies the predicate in a rolling fashion i.e.

1151-- | Like 'spanBy' but applies the predicate in a rolling fashion i.e.

1143-- predicate is applied to the previous and the next input elements.

1152-- predicate is applied to the previous and the next input elements.

1144−{-# INLINE spanRollingBy #-}

1153+{-# INLINE _spanRollingBy #-}

1145−spanRollingBy

1154+_spanRollingBy

1146 :: Monad m

1155 :: Monad m

1147 => (a -> a -> Bool)

1156 => (a -> a -> Bool)

1148 -> Fold m a b

1157 -> Fold m a b

1149 -> Fold m a c

1158 -> Fold m a c

1150 -> Fold m a (b, c)

1159 -> Fold m a (b, c)

1151−spanRollingBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =

1160+_spanRollingBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =

1152 Fold step init extract

1161 Fold step init extract

1153

1162

1154 where

1163 where

@@ -1163,33 +1172,11 @@ spanRollingBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR)

1163 stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))

1172 stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))

1164

1173

1165 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b

1174 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b

1166−-}

1167

1175

1168------------------------------------------------------------------------------

1176------------------------------------------------------------------------------

1169-- N-ary APIs

1177-- N-ary APIs

1170------------------------------------------------------------------------------

1178------------------------------------------------------------------------------

1171--

1179--

1172−-- XXX should we use a strict pair?

1173−--

1174−-- | The splitter returns True if the current element is the last element of

1175−-- the group, otherwise returns false.

1176−--

1177−-- @since 0.7.0

1178−{-# INLINE grouped #-}

1179−grouped

1180− :: (IsStream t, Monad m)

1181− => Fold m a b

1182− -> t m (a, Bool)

1183− -> t m b

1184−grouped f m = D.fromStreamD $ D.grouped f (D.toStreamD m)

1185−

1186−-- An example of a splitter to be used with grouped.

1187−_newline :: IsStream t => t m Char -> t m (Char,Bool)

1188−_newline m = S.foldrS (\x xs ->

1189− if x == '\n'

1190− then (x,True) `K.cons` xs

1191− else (x,False) `K.cons` xs) K.nil m

1192−

1193-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the

1180-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the

1194-- first group, if @a \`cmp` b@ is 'True' then @b@ is also assigned to the same

1181-- first group, if @a \`cmp` b@ is 'True' then @b@ is also assigned to the same

1195-- group. If @a \`cmp` c@ is 'True' then @c@ is also assigned to the same

1182-- group. If @a \`cmp` c@ is 'True' then @c@ is also assigned to the same

@@ -1292,6 +1279,10 @@ breakOn pat f m = undefined

1292-- >>> splitBy_ (== '.') "a..b"

1279-- >>> splitBy_ (== '.') "a..b"

1293-- > ["a","","b"]

1280-- > ["a","","b"]

1294--

1281--

1282+-- This can be considered as an n-fold version of 'break' where we apply

1283+-- 'break' successively on the input stream, dropping the first element

1284+-- of the second segment after each break.

1285+--

1295-- @since 0.7.0

1286-- @since 0.7.0

1296{-# INLINE splitBy #-}

1287{-# INLINE splitBy #-}

1297splitBy

1288splitBy

@@ -1334,6 +1325,10 @@ splitBy predicate f m =

1334--

1325--

1335-- > lines = splitSuffixBy (== '\n')

1326-- > lines = splitSuffixBy (== '\n')

1336--

1327--

1328+-- This can be considered as an n-fold version of 'breakPost' where we apply

1329+-- 'breakPost' successively on the input stream, dropping the first element

1330+-- of the second segment after each break.

1331+--

1337-- @since 0.7.0

1332-- @since 0.7.0

1338{-# INLINE splitSuffixBy #-}

1333{-# INLINE splitSuffixBy #-}

1339splitSuffixBy

1334splitSuffixBy

@@ -1402,15 +1397,19 @@ wordsBy predicate f m =

1402-- >>> splitSuffixBy'_ (== '.') "a..b.."

1397-- >>> splitSuffixBy'_ (== '.') "a..b.."

1403-- > ["a.",".","b.","."]

1398-- > ["a.",".","b.","."]

1404--

1399--

1400+-- This can be considered as an n-fold version of 'breakPost' where we apply

1401+-- 'breakPost' successively on the input stream.

1402+--

1405-- @since 0.7.0

1403-- @since 0.7.0

1406−{-# INLINE _splitSuffixBy' #-}

1404+{-# INLINE splitSuffixBy' #-}

1407−_splitSuffixBy'

1405+splitSuffixBy'

1408 :: (IsStream t, Monad m)

1406 :: (IsStream t, Monad m)

1409 => (a -> Bool) -> Fold m a b -> t m a -> t m b

1407 => (a -> Bool) -> Fold m a b -> t m a -> t m b

1410−_splitSuffixBy' predicate f m = grouped f (S.map (\a -> (a, predicate a)) m)

1408+splitSuffixBy' predicate f m =

1409+ D.fromStreamD $ D.splitSuffixBy' predicate f (D.toStreamD m)

1411

1410

1412------------------------------------------------------------------------------

1411------------------------------------------------------------------------------

1413−-- Split on a delimiter

1412+-- Split on a delimiter sequence

1414------------------------------------------------------------------------------

1413------------------------------------------------------------------------------

1415

1414

1416-- Int list examples for splitOn:

1415-- Int list examples for splitOn:

@@ -1986,6 +1985,6 @@ lconcatMap s f1 f2 = undefined

1986-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c

1985-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c

1987--

1986--

1988-- @

1987-- @

1989−lgroupsOf :: Int -> Fold m a b -> Fold m b c -> Fold m a c

1988+lchunksOf :: Int -> Fold m a b -> Fold m b c -> Fold m a c

1990−lgroupsOf n f1 f2 = undefined

1989+lchunksOf n f1 f2 = undefined

1991-}

1990-}

src/Streamly/Prelude.hs+1

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

358

359 , findIndices

360 , elemIndices

361+ -- , seqIndices -- search a sequence in the stream

361

362

362 -- ** Insertion

363 -- ** Insertion

363 -- | Insertion adds more elements to the stream.

364 -- | Insertion adds more elements to the stream.

src/Streamly/Streams/StreamD.hs+9−13

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

131

132 -- ** Grouping

133 , groupsOf

134− , grouped

135 , groupsBy

134 , groupsBy

136 , groupsRollingBy

135 , groupsRollingBy

137

136

@@ -922,9 +921,11 @@ concatMap f = concatMapM (return . f)

922-- Grouping/Splitting

921-- Grouping/Splitting

923------------------------------------------------------------------------------

922------------------------------------------------------------------------------

924

923

925−{-# INLINE_NORMAL grouped #-}

924+{-# INLINE_NORMAL splitSuffixBy' #-}

925+splitSuffixBy' :: Monad m

926−grouped :: Monad m => Fold m a b -> Stream m (a, Bool) -> Stream m b

926+ => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b

927+splitSuffixBy' predicate f (Stream step state) =

927−grouped f (Stream step state) = Stream (stepOuter f) (Just state)

928+ Stream (stepOuter f) (Just state)

928

929

929 where

930 where

930

931

@@ -932,10 +933,10 @@ grouped f (Stream step state) = Stream (stepOuter f) (Just state)

932 stepOuter (Fold fstep initial done) gst (Just st) = do

933 stepOuter (Fold fstep initial done) gst (Just st) = do

933 res <- step (adaptState gst) st

934 res <- step (adaptState gst) st

934 case res of

935 case res of

935− Yield (x,r) s -> do

936+ Yield x s -> do

936 acc <- initial

937 acc <- initial

937 acc' <- fstep acc x

938 acc' <- fstep acc x

938− if r

939+ if (predicate x)

939 then done acc' >>= \val -> return $ Yield val (Just s)

940 then done acc' >>= \val -> return $ Yield val (Just s)

940 else go SPEC s acc'

941 else go SPEC s acc'

941

942

@@ -947,9 +948,9 @@ grouped f (Stream step state) = Stream (stepOuter f) (Just state)

947 go !_ stt !acc = do

948 go !_ stt !acc = do

948 res <- step (adaptState gst) stt

949 res <- step (adaptState gst) stt

949 case res of

950 case res of

950− Yield (x,r) s -> do

951+ Yield x s -> do

951 acc' <- fstep acc x

952 acc' <- fstep acc x

952− if r

953+ if (predicate x)

953 then done acc' >>= \val -> return $ Yield val (Just s)

954 then done acc' >>= \val -> return $ Yield val (Just s)

954 else go SPEC s acc'

955 else go SPEC s acc'

955 Skip s -> go SPEC s acc

956 Skip s -> go SPEC s acc

@@ -1071,11 +1072,6 @@ groupsRollingBy cmp f (Stream step state) =

1071

1072

1072 stepOuter _ _ (Nothing, _) = return Stop

1073 stepOuter _ _ (Nothing, _) = return Stop

1073

1074

1074−{-# INLINE_EARLY splitSuffixBy' #-}

1075−splitSuffixBy' :: Monad m

1076− => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b

1077−splitSuffixBy' predicate f m = grouped f (map (\a -> (a, predicate a)) m)

1078−

1079{-# INLINE_NORMAL splitBy #-}

1075{-# INLINE_NORMAL splitBy #-}

1080splitBy :: Monad m => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b

1076splitBy :: Monad m => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b

1081splitBy predicate f (Stream step state) = Stream (step' f) (Just state)

1077splitBy predicate f (Stream step state) = Stream (step' f) (Just state)

src/Streamly/Time.hs+5−56

@@ -1,5 +1,3 @@

1−{-# LANGUAGE FlexibleContexts #-}

2−

3-- |

1-- |

4-- Module : Streamly.Time

2-- Module : Streamly.Time

@@ -12,26 +10,16 @@

12-- Time utilities for reactive programming.

10-- Time utilities for reactive programming.

14module Streamly.Time

12module Streamly.Time

15− ( intervalsOf

13+{-# DEPRECATED

16−

14+ "Please use the \"rate\" combinator instead of the functions in this module"

17− -- * Deprecated

15+ #-}

18− , periodic

16+ ( periodic

19 , withClock

17 , withClock

20 )

18 )

21where

19where

23−import Control.Concurrent (threadDelay)

24import Control.Monad (when)

21import Control.Monad (when)

25−import Control.Monad.IO.Class (MonadIO(liftIO))

22+import Control.Concurrent (threadDelay)

26−import Data.Maybe (fromJust, isJust)

27−

28−import Streamly (parallel)

29−import Streamly.SVar (MonadAsync)

30−import Streamly.Streams.StreamK (IsStream(..))

31−import Streamly.Fold (Fold)

32−

33−import qualified Streamly.Fold as FL

34−import qualified Streamly.Prelude as S

36-- | Run an action forever periodically at the given frequency specified in per

24-- | Run an action forever periodically at the given frequency specified in per

37-- second (Hz).

25-- second (Hz).

@@ -84,42 +72,3 @@ withClock clock freq action = do

84 lastPeriod = (baseTime - lastAdj) `div` freq

72 lastPeriod = (baseTime - lastAdj) `div` freq

85 newDelay = max 0 (delay + period - lastPeriod)

73 newDelay = max 0 (delay + period - lastPeriod)

86 return (baseTime, newTick, newDelay)

74 return (baseTime, newTick, newDelay)

87−

88−-------------------------------------------------------------------------------

89−-- Stream element unaware: Time based buffering

90−-------------------------------------------------------------------------------

91−

92−-- There are two ways in which time based buffering can be implemented.

93−--

94−-- 1) One is to use an API like groupsOf to group elements by time and then

95−-- flush when the group is complete. This will require checking time on each

96−-- element yield.

97−--

98−-- 2) Use an async thread to write. The producer thread would just queue the

99−-- elements to be written on a queue associated with the writer thread. When

100−-- the write thread is scheduled elements that have been buffered will be

101−-- written based on a buffering policy. This will require the producer to

102−-- synchronize on the queue, send a doorbell when the queue has items in it.

103−-- The queue can be an array which can be written directly to the IO device.

104−--

105−-- We can try both and see which one performs better.

106−

107−-- XXX add this example after fixing the serial stream rate control

108−-- >>> S.toList $ S.take 5 $ intervalsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1

109−-- > [3,7,11,15,19]

110−--

111−-- | Group the input stream into windows of @n@ second each and then fold each

112−-- group using the provided fold function.

113−--

114−-- @since 0.7.0

115−{-# INLINE intervalsOf #-}

116−intervalsOf

117− :: (IsStream t, MonadAsync m)

118− => Double -> Fold m a b -> t m a -> t m b

119−intervalsOf n f m = FL.grouped (catMaybes f) s

120− where

121− s = S.map (\x -> (Just x, False)) m `parallel` S.repeatM timeout

122− timeout = do

123− liftIO $ threadDelay (round $ n * 1000000)

124− return (Nothing, True)

125− catMaybes = FL.lfilter isJust . FL.lmap fromJust