2723c15 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Implement stream parsing

2723c15Harendra Kumar7 years ago

benchmark/FileIO.hs+9

compare with master

unifiedsplit

@@ -212,6 +212,12 @@ main = do

212 , mkBench "sumChunksOf 1" href $ do

213 Handles inh _ <- readIORef href

214 BFS.chunksOfSum 1 inh

215+ , mkBench "sumChunksOf (single chunk) (parseChunks)" href $ do

216+ Handles inh _ <- readIORef href

217+ BFS.parseChunksOfSum fileSize inh

218+ , mkBench "sumChunksOf 1 (parseChunks)" href $ do

219+ Handles inh _ <- readIORef href

220+ BFS.parseChunksOfSum 1 inh

215

221

216 , mkBench "arraysOf 1" href $ do

222 , mkBench "arraysOf 1" href $ do

217 Handles inh _ <- readIORef href

223 Handles inh _ <- readIORef href

@@ -262,6 +268,9 @@ main = do

262 , mkBench "splitOnSuffix \\n (line count)" href $ do

268 , mkBench "splitOnSuffix \\n (line count)" href $ do

263 Handles inh _ <- readIORef href

269 Handles inh _ <- readIORef href

264 BFS.splitOnSuffix inh

270 BFS.splitOnSuffix inh

271+ , mkBench "splitOn \\n (line count) (parseChunks)" href $ do

272+ Handles inh _ <- readIORef href

273+ BFS.parseChunksSplitOn inh

265 , mkBench "wordsBy isSpace (word count)" href $ do

274 , mkBench "wordsBy isSpace (word count)" href $ do

266 Handles inh _ <- readIORef href

275 Handles inh _ <- readIORef href

267 BFS.wordsBy inh

276 BFS.wordsBy inh

benchmark/Streamly/Benchmark/Data/Fold.hs+17

@@ -20,6 +20,7 @@ import Prelude (IO, Int, Double, String, (>), (<*>), (<$>), (+), ($),

20import qualified Streamly as S hiding (runStream)

21import qualified Streamly.Prelude as S

22import qualified Streamly.Internal.Data.Fold as FL

23+import qualified Streamly.Internal.Data.Parse as PR

23import qualified Streamly.Internal.Data.Pipe as Pipe

24import qualified Streamly.Internal.Data.Pipe as Pipe

25import qualified Streamly.Internal.Data.Sink as Sink

26import qualified Streamly.Internal.Data.Sink as Sink

@@ -205,6 +206,21 @@ o_n_heap_serial_folds value =

205 ]

206 ]

207 ]

207

208

209+o_1_space_serial_parse :: Int -> [Benchmark]

210+o_1_space_serial_parse value =

211+ [ bgroup

212+ "serially"

213+ [ bgroup "parser"

214+ [

215+ benchIOSink value "any" $ IP.parse (PR.any (> value))

216+ , benchIOSink value "all" $ IP.parse (PR.all (<= value))

217+ , benchIOSink value "(all,any)" $

218+ IP.parse ((,) <$> PR.all (<= value) <*> PR.any (> value))

219+ , benchIOSink value "take" (IP.parse (FL.take value FL.drain))

220+ ]

221+ ]

222+ ]

223+

208-------------------------------------------------------------------------------

224-------------------------------------------------------------------------------

209-- Driver

225-- Driver

210-------------------------------------------------------------------------------

226-------------------------------------------------------------------------------

@@ -223,6 +239,7 @@ main = do

223 , o_1_space_serial_foldsTransforms value

239 , o_1_space_serial_foldsTransforms value

224 , o_1_space_serial_foldsCompositions value

240 , o_1_space_serial_foldsCompositions value

225 ]

241 ]

242+ , bgroup "parser" $ o_1_space_serial_parse value

226 ]

243 ]

227 , bgroup

244 , bgroup

228 "o-n-heap"

245 "o-n-heap"

benchmark/Streamly/Benchmark/FileIO/Stream.hs+16−1

@@ -57,10 +57,12 @@ module Streamly.Benchmark.FileIO.Stream

57 , decodeUtf8Lax

58 , copyCodecUtf8Lenient

59 , chunksOfSum

60+ , parseChunksOfSum

60 , chunksOf

61 , chunksOf

61 , chunksOfD

62 , chunksOfD

62 , splitOn

63 , splitOn

63 , splitOnSuffix

64 , splitOnSuffix

65+ , parseChunksSplitOn

64 , wordsBy

66 , wordsBy

65 , splitOnSeq

67 , splitOnSeq

66 , splitOnSeqUtf8

68 , splitOnSeqUtf8

@@ -81,11 +83,12 @@ import qualified Streamly.Memory.Array as A

81import qualified Streamly.Internal.Memory.Array.Types as AT

83import qualified Streamly.Internal.Memory.Array.Types as AT

82import qualified Streamly.Prelude as S

84import qualified Streamly.Prelude as S

83import qualified Streamly.Data.Fold as FL

85import qualified Streamly.Data.Fold as FL

84−-- import qualified Streamly.Internal.Data.Fold as IFL

86+import qualified Streamly.Internal.Data.Fold as IFL

85import qualified Streamly.Data.Unicode.Stream as SS

87import qualified Streamly.Data.Unicode.Stream as SS

86import qualified Streamly.Internal.Data.Unicode.Stream as IUS

88import qualified Streamly.Internal.Data.Unicode.Stream as IUS

87import qualified Streamly.Internal.Memory.Unicode.Array as IUA

89import qualified Streamly.Internal.Memory.Unicode.Array as IUA

88import qualified Streamly.Internal.Data.Unfold as IUF

90import qualified Streamly.Internal.Data.Unfold as IUF

91+import qualified Streamly.Internal.Data.Parse as PR

89import qualified Streamly.Internal.Prelude as IP

92import qualified Streamly.Internal.Prelude as IP

90import qualified Streamly.Internal.Data.Stream.StreamD as D

93import qualified Streamly.Internal.Data.Stream.StreamD as D

@@ -442,6 +445,11 @@ inspect $ 'chunksOfD `hasNoType` ''AT.FlattenState

442inspect $ 'chunksOfD `hasNoType` ''D.ConcatMapUState

445inspect $ 'chunksOfD `hasNoType` ''D.ConcatMapUState

443#endif

446#endif

444

447

448+{-# INLINE parseChunksOfSum #-}

449+parseChunksOfSum :: Int -> Handle -> IO Int

450+parseChunksOfSum n inh =

451+ S.length $ IP.parseChunks (IFL.take n FL.sum) (S.unfold FH.read inh)

452+

445{-# INLINE linesUnlinesCopy #-}

453{-# INLINE linesUnlinesCopy #-}

446linesUnlinesCopy :: Handle -> Handle -> IO ()

454linesUnlinesCopy :: Handle -> Handle -> IO ()

447linesUnlinesCopy inh outh =

455linesUnlinesCopy inh outh =

@@ -600,6 +608,13 @@ inspect $ 'splitOnSuffix `hasNoType` ''AT.FlattenState

600inspect $ 'splitOnSuffix `hasNoType` ''D.ConcatMapUState

608inspect $ 'splitOnSuffix `hasNoType` ''D.ConcatMapUState

601#endif

609#endif

602

610

611+-- | Split on line feed.

612+{-# INLINE parseChunksSplitOn #-}

613+parseChunksSplitOn :: Handle -> IO Int

614+parseChunksSplitOn inh =

615+ (S.length $ IP.parseChunks (PR.endOn (== lf) FL.drain)

616+ (S.unfold FH.read inh)) -- >>= print

617+

603-- | Words by space

618-- | Words by space

604{-# INLINE wordsBy #-}

619{-# INLINE wordsBy #-}

605wordsBy :: Handle -> IO Int

620wordsBy :: Handle -> IO Int

src/Streamly/Internal/Data/Fold.hs+55−7

@@ -129,6 +129,7 @@ module Streamly.Internal.Data.Fold

129

130 -- * Parsing

131 -- ** Trimming

132+ , take

132 , ltake

133 , ltake

133 -- , lrunFor -- time

134 -- , lrunFor -- time

134 , ltakeWhile

135 , ltakeWhile

@@ -193,15 +194,15 @@ module Streamly.Internal.Data.Fold

193 -- , unzipWith

194 -- , unzipWith

194 -- , unzipWithM

195 -- , unzipWithM

195

196

197+ -- * Nested Folds

198+ -- , concatMap

199+ , foldChunks

200+ , duplicate

201+

196 -- * Running Folds

202 -- * Running Folds

197 , initialize

203 , initialize

198 , runStep

204 , runStep

199

205

200− -- * Nested Folds

201− -- , concatMap

202− -- , chunksOf

203− , duplicate -- experimental

204−

205 -- * Folding to SVar

206 -- * Folding to SVar

206 , toParallelSVar

207 , toParallelSVar

207 , toParallelSVarLimited

208 , toParallelSVarLimited

@@ -227,6 +228,7 @@ import qualified Prelude

227

228

228import Streamly.Internal.Data.Pipe.Types (Pipe (..), PipeState(..))

229import Streamly.Internal.Data.Pipe.Types (Pipe (..), PipeState(..))

229import Streamly.Internal.Data.Fold.Types

230import Streamly.Internal.Data.Fold.Types

231+import Streamly.Internal.Data.Parse.Types

230import Streamly.Internal.Data.Strict

232import Streamly.Internal.Data.Strict

231import Streamly.Internal.Data.SVar

233import Streamly.Internal.Data.SVar

232

234

@@ -827,10 +829,41 @@ or = Fold (\x a -> return $ x || a) (return False) return

827-- Grouping without looking at elements

829-- Grouping without looking at elements

828------------------------------------------------------------------------------

830------------------------------------------------------------------------------

829--

831--

832+-- XXX Once we have terminating folds, this Parse should get replaced by Fold.

833+-- Alternatively, we can name it "chunkOf" and the corresponding time domain

834+-- combinator as "intervalOf" or even "chunk" and "interval".

835+--

836+-- | Take at most @n@ input elements and fold them using the supplied fold.

837+--

838+-- Stops after @n@ elements.

839+-- Never fails.

840+--

841+-- /Internal/

842+--

843+{-# INLINE take #-}

844+take :: Monad m => Int -> Fold m a b -> Parse m a b

845+take n (Fold fstep finitial fextract) = Parse step initial extract

846+

847+ where

848+

849+ initial = (Tuple' 0) <$> finitial

850+

851+ step (Tuple' i r) a = do

852+ res <- fstep r a

853+ let i1 = i + 1

854+ s1 = Tuple' i1 res

855+ if i1 < n

856+ then return $ Keep 0 s1

857+ else return $ Halt s1

858+

859+ extract (Tuple' _ r) = fmap f (fextract r)

860+ where f x = Right (0, x)

861+

830------------------------------------------------------------------------------

862------------------------------------------------------------------------------

831-- Binary APIs

863-- Binary APIs

832------------------------------------------------------------------------------

864------------------------------------------------------------------------------

833--

865--

866+-- XXX These would just be applicative compositions of terminating folds.

834

867

835-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@

868-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@

836-- elements of its input are consumed by fold @f1@ and the rest of the stream

869-- elements of its input are consumed by fold @f1@ and the rest of the stream

@@ -1494,7 +1527,7 @@ unzip = unzipWith id

1494------------------------------------------------------------------------------

1527------------------------------------------------------------------------------

1495-- Nesting

1528-- Nesting

1496------------------------------------------------------------------------------

1529------------------------------------------------------------------------------

1497−--

1530+

1498{-

1531{-

1499-- All the stream flattening transformations can also be applied to a fold

1532-- All the stream flattening transformations can also be applied to a fold

1500-- input stream.

1533-- input stream.

@@ -1507,9 +1540,24 @@ lconcatMap s f1 f2 = undefined

1507-}

1540-}

1508

1541

1509-- All the grouping transformation that we apply to a stream can also be

1542-- All the grouping transformation that we apply to a stream can also be

1510−-- applied to a fold input stream.

1543+-- applied to a fold input stream. groupBy et al can be written as terminating

1544+-- folds and then we can apply foldChunks to use those repeatedly on a stream.

1545+

1546+-- | Apply a terminating fold repeatedly to the input of another fold.

1547+--

1548+-- Compare with: Streamly.Prelude.concatMap, Streamly.Prelude.foldChunks

1549+--

1550+-- /Unimplemented/

1551+--

1552+{-# INLINABLE foldChunks #-}

1553+foldChunks ::

1554+ -- Monad m =>

1555+ Fold m a b -> Fold m b c -> Fold m a c

1556+foldChunks = undefined

1511

1557

1512{-

1558{-

1559+-- XXX this would be an application of foldChunks using a terminating fold.

1560+--

1513-- | Group the input stream into groups of elements between @low@ and @high@.

1561-- | Group the input stream into groups of elements between @low@ and @high@.

1514-- Collection starts in chunks of @low@ and then keeps doubling until we reach

1562-- Collection starts in chunks of @low@ and then keeps doubling until we reach

1515-- @high@. Each chunk is folded using the provided fold function.

1563-- @high@. Each chunk is folded using the provided fold function.

src/Streamly/Internal/Data/Fold/Types.hs+13

@@ -277,6 +277,12 @@ lfilterM f (Fold step begin done) = Fold step' begin done

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

278lcatMaybes = lfilter isJust . lmap fromJust

279

280+------------------------------------------------------------------------------

281+-- Parsing

282+------------------------------------------------------------------------------

283+

284+-- XXX These should become terminating folds.

285+--

280-- | Take first 'n' elements from the stream and discard the rest.

286-- | Take first 'n' elements from the stream and discard the rest.

281--

287--

282-- @since 0.7.0

288-- @since 0.7.0

@@ -348,6 +354,13 @@ runStep (Fold step initial extract) a = do

348 r <- step i a

354 r <- step i a

349 return $ (Fold step (return r) extract)

355 return $ (Fold step (return r) extract)

350

356

357+------------------------------------------------------------------------------

358+-- Parsing

359+------------------------------------------------------------------------------

360+

361+-- XXX These can be expressed using foldChunks repeatedly on the input of a

362+-- fold.

363+

351-- | For every n input items, apply the first fold and supply the result to the

364-- | For every n input items, apply the first fold and supply the result to the

352-- next fold.

365-- next fold.

353--

366--

src/Streamly/Internal/Data/Parse.hsnew file+254

@@ -0,0 +1,254 @@

1+{-# LANGUAGE BangPatterns #-}

2+{-# LANGUAGE CPP #-}

3+{-# LANGUAGE RankNTypes #-}

4+{-# LANGUAGE RecordWildCards #-}

5+{-# LANGUAGE ScopedTypeVariables #-}

7+-- |

8+-- Module : Streamly.Internal.Data.Parse

10+-- License : BSD3

11+-- Maintainer : streamly@composewell.com

12+-- Stability : experimental

13+-- Portability : GHC

14+--

15+-- Parsers.

16+

17+module Streamly.Internal.Data.Parse

18+ (

19+ Parse (..)

20+

21+ -- * Parsers

22+ , any

23+ , all

24+

25+ , takeWhile

26+ , takeExact

27+ , takeAtLeast

28+

29+ , endOn

30+ , endAfter

31+ , endBefore

32+

33+ , finishBy

34+ )

35+where

36+

37+import Prelude

38+ hiding (any, all, takeWhile)

39+import Streamly.Internal.Data.Parse.Types (Parse(..), Step(..))

40+import Streamly.Internal.Data.Fold.Types (Fold(..))

41+

42+import Streamly.Internal.Data.Strict

43+

44+-------------------------------------------------------------------------------

45+-- Terminating folds

46+-------------------------------------------------------------------------------

47+--

48+-- XXX any/all should be terminating folds when we have a terminating fold

49+-- type.

50+--

51+-- >>> S.parse (PR.any (== 0)) $ S.fromList [1,0,1]

52+-- > Right True

53+--

54+{-# INLINABLE any #-}

55+any :: Monad m => (a -> Bool) -> Parse m a Bool

56+any predicate = Parse step initial (\x -> return $ Right (0,x))

57+ where

58+ initial = return False

59+ step s a = return $

60+ if s

61+ then Halt True

62+ else

63+ if predicate a

64+ then Halt True

65+ else Keep 0 False

66+

67+-- >>> S.parse (PR.any (== 0)) $ S.fromList [1,0,1]

68+-- > Right False

69+--

70+{-# INLINABLE all #-}

71+all :: Monad m => (a -> Bool) -> Parse m a Bool

72+all predicate = Parse step initial (\x -> return $ Right (0,x))

73+ where

74+ initial = return True

75+ step s a = return $

76+ if s

77+ then

78+ if predicate a

79+ then Keep 0 True

80+ else Halt False

81+ else Halt False

82+

83+-------------------------------------------------------------------------------

84+-- Taking elements

85+-------------------------------------------------------------------------------

86+--

87+-- | Stops after taking exactly @n@ input elements.

88+--

89+-- * Stops - after @n@ elements.

90+-- * Fails - if the stream ends before it can collect @n@ elements.

91+--

92+-- >>> S.parse (PR.takeExact 4 FL.toList) $ S.fromList [1,0,1]

93+-- > Left "takeExact: Expecting exactly 4 elements, got 3"

94+--

95+-- /Internal/

96+--

97+{-# INLINABLE takeExact #-}

98+takeExact :: Monad m => Int -> Fold m a b -> Parse m a b

99+takeExact n (Fold fstep finitial fextract) = Parse step initial extract

100+

101+ where

102+

103+ initial = (Tuple' 0) <$> finitial

104+

105+ step (Tuple' i r) a = do

106+ res <- fstep r a

107+ let i1 = i + 1

108+ s1 = Tuple' i1 res

109+ return $ if i1 < n then Hold s1 else Halt s1

110+

111+ extract (Tuple' i r) = fmap f (fextract r)

112+

113+ where

114+

115+ err =

116+ "takeExact: Expecting exactly " ++ show n

117+ ++ " elements, got " ++ show i

118+

119+ f x =

120+ if n == i

121+ then Right (0, x)

122+ else Left err

123+

124+-- | Take at least @n@ input elements, but can collect more.

125+--

126+-- * Stops - never.

127+-- * Fails - if the stream end before producing @n@ elements.

128+--

129+-- >>> S.parse (PR.takeAtLeast 4 FL.toList) $ S.fromList [1,0,1]

130+-- > Left "takeAtLeast: Expecting at least 4 elements, got only 3"

131+--

132+-- >>> S.parse (PR.takeAtLeast 4 FL.toList) $ S.fromList [1,0,1,0,1]

133+-- > Right [1,0,1,0,1]

134+--

135+-- /Internal/

136+--

137+{-# INLINABLE takeAtLeast #-}

138+takeAtLeast :: Monad m => Int -> Fold m a b -> Parse m a b

139+takeAtLeast n (Fold fstep finitial fextract) = Parse step initial extract

140+

141+ where

142+

143+ initial = (Tuple' 0) <$> finitial

144+

145+ step (Tuple' i r) a = do

146+ res <- fstep r a

147+ let i1 = i + 1

148+ s1 = Tuple' i1 res

149+ return $ Hold s1

150+

151+ extract (Tuple' i r) = fmap f (fextract r)

152+

153+ where

154+

155+ err =

156+ "takeAtLeast: Expecting at least " ++ show n

157+ ++ " elements, got only " ++ show i

158+

159+ f x =

160+ if i >= n

161+ then Right (0, x)

162+ else Left err

163+

164+-- | Take until the predicate fails. Does not take the failing element.

165+--

166+-- * Stops - when the predicate fails.

167+-- * Fails - never.

168+--

169+-- >>> S.parse (PR.takeWhile (== 0) FL.toList) $ S.fromList [0,0,1,0,1]

170+-- > Right [0,0]

171+--

172+-- /Internal/

173+--

174+{-# INLINABLE takeWhile #-}

175+takeWhile :: Monad m => (a -> Bool) -> Fold m a b -> Parse m a b

176+takeWhile predicate (Fold fstep finitial fextract) =

177+ Parse step initial extract

178+

179+ where

180+

181+ initial = finitial

182+ step s a = do

183+ if predicate a

184+ then Keep 0 <$> fstep s a

185+ else return $ Halt s

186+ extract s = do

187+ b <- fextract s

188+ return $ Right (1, b)

189+

190+-- | Keep taking elements until the predicate succeeds. Drop the succeeding

191+-- element.

192+--

193+-- * Stops - when the predicate succeeds.

194+-- * Fails - never.

195+--

196+-- >>> S.parse (PR.endOn (== 1) FL.toList) $ S.fromList [0,0,1,0,1]

197+-- > Right [0,0]

198+--

199+-- >>> S.toList $ S.parseChunks (PR.endOn (== 1) FL.toList) $ S.fromList [0,0,1,0,1]

200+-- > [[0,0],[0],[]]

201+--

202+-- /Internal/

203+--

204+{-# INLINABLE endOn #-}

205+endOn :: Monad m => (a -> Bool) -> Fold m a b -> Parse m a b

206+endOn predicate (Fold fstep finitial fextract) =

207+ Parse step initial extract

208+

209+ where

210+

211+ initial = finitial

212+ step s a = do

213+ if not (predicate a)

214+ then Keep 0 <$> fstep s a

215+ else return $ Halt s

216+ extract s = do

217+ b <- fextract s

218+ return $ Right (0, b)

219+

220+-- | Keep taking elements until the predicate succeeds. Take the succeeding

221+-- element as well.

222+--

223+-- * Stops - when the predicate succeeds.

224+-- * Fails - never.

225+--

226+-- /Unimplemented/

227+--

228+{-# INLINABLE endAfter #-}

229+endAfter ::

230+ -- Monad m =>

231+ (a -> Bool) -> Fold m a b -> Parse m a b

232+endAfter = undefined

233+

234+-- | Keep taking elements until the predicate succeeds. Return the succeeding

235+-- element back to the input.

236+--

237+-- * Stops - when the predicate succeeds.

238+-- * Fails - never.

239+--

240+-- /Unimplemented/

241+--

242+{-# INLINABLE endBefore #-}

243+endBefore ::

244+ -- Monad m =>

245+ (a -> Bool) -> Fold m a b -> Parse m a b

246+endBefore = undefined

247+

248+-- | Distribute the input to a parse and a fold until the parse succeeds or

249+-- fails. The parse can be used to decide the termination of the fold.

250+--

251+-- /Unimplemented/

252+--

253+finishBy :: Parse m a x -> Fold m a y -> Parse m a (x, y)

254+finishBy = undefined

src/Streamly/Internal/Data/Parse/Types.hsnew file+169

@@ -0,0 +1,169 @@

1+{-# LANGUAGE ExistentialQuantification #-}

2+{-# LANGUAGE BangPatterns #-}

3+{-# LANGUAGE DeriveFunctor #-}

4+{-# LANGUAGE DeriveAnyClass #-}

5+{-# LANGUAGE DeriveGeneric #-}

7+-- |

8+-- Module : Streamly.Parse.Types

10+-- License : BSD3

11+-- Maintainer : streamly@composewell.com

12+-- Stability : experimental

13+-- Portability : GHC

14+--

15+-- Streaming and backtracking parsers.

16+--

17+-- A parser is a generalization of a classical left fold. A fold never fails,

18+-- it always produces a result, it has a default initial value. A parser may

19+-- terminate or fail even without producing any result.

20+

21+-- Differences between a terminating fold and a parser:

22+--

23+-- \* A Parser is a superset of terminating fold with failure and backtracking

24+-- added.

25+-- \* A fold may be upgraded to a parser but vice versa is not true.

26+-- \* A fold is not a partial function, as it may never fail.

27+-- \* A fold's Step result is simpler as it does not backtrack.

28+-- \* A fold's result type is simpler as it does not fail and it does not return

29+-- unused input count.

30+--

31+-- Parse and Zip (zipWith) is possible only if both the parses are unskipping

32+-- or are guaranteed to have outputs in sync. Or else we will have to keep the

33+-- input buffered and may have to end the zip when one of them finishes. This

34+-- is simple if the type guarantees no skip e.g. in case of classical left

35+-- folds.

36+--

37+-- Terminating folds can have Sequential and Distribute and zip style

38+-- applicatives but cannot have Alternative.

39+--

40+-- Parsers can have Sequential Applicative and Alternative but not Distribute

41+-- and zip style applicative.

42+--

43+-- We can distribute and scan/parse a stream using both folds and parsers and

44+-- merge the resulting streams using different merge strategies (e.g.

45+-- interleaving or serial).

46+

47+module Streamly.Internal.Data.Parse.Types

48+ (

49+ Step (..)

50+ , Parse (..)

51+ )

52+where

53+

54+-- | The return type of a fold step. A fold is driven by a fold driver one step

55+-- at a time, at any time the driver may @extract@ and examine the result of

56+-- the fold. The fold may backtrack at any point, therefore, the driver holds

57+-- the input in a backtracking buffer. The buffer grows or shrinks based on the

58+-- return values of the fold step execution.

59+--

60+-- When a fold step is executed it generates a new state of the fold along with

61+-- a command to the fold driver. The command tells the fold driver whether to

62+-- keep the input for future backtracking or drop it. The constructors of

63+-- 'Step' represent the commands to the driver.

64+--

65+-- /Internal/

66+--

67+data Step s =

68+ Hold s -- ^ Hold the current input in the buffer, the input is

69+ -- accepted tentatively, but the fold may later fail with the

70+ -- input remaining unused.

71+ | Keep Int s -- ^ Keep @n@ most recent inputs, including the input of this

72+ -- step invocation, and discard the rest. This indicates a

73+ -- commit of an intermediate result of the fold. A fold cannot

74+ -- fail beyond this commit point. If the fold terminates it

75+ -- will return at least the result up to this commit and the

76+ -- count of unused elements after that.

77+ | Back Int s -- ^ Go back in the buffer by @n@ items, the count @n@

78+ -- includes the input element of this step as well. The fold

79+ -- driver backtracks and starts driving the fold with past

80+ -- inputs from the requested point in the buffer. Note that

81+ -- the driver cannot go back beyond the last commit point

82+ -- created by 'Keep'.

83+ | Halt s -- ^ Halt because the result of the fold will not change after

84+ -- this point. Once a fold halts, driving it further may

85+ -- produce undefined behavior. An @extract@ after the 'Halt'

86+ -- would return a count of unused elements which includes the

87+ -- element on which 'Halt' occurred.

88+

89+-- Note: Instead of returning the unused count as part of extract, we can

90+-- return it via Halt command. However, some folds may not ever halt, but they

91+-- may still have unused input after the last commit point.

92+

93+-- | A parsing fold is represented as @Parse step initial extract@. Before we

94+-- drive a fold we call the @initial@ action to retrieve the initial state of

95+-- the fold. The driver invokes @step@ with the previous state and an input

96+-- element. It results into a new state and a command to the driver represented

97+-- by 'Step' type. At any point of time the driver can call @extract@ to

98+-- inspect the result of the fold. It may result in an error or an output value

99+-- along with the count (zero or more) of last unused elements.

100+--

101+data Parse m a b =

102+ forall s. Parse (s -> a -> m (Step s))

103+ (m s)

104+ (s -> m (Either String (Int, b)))

105+

106+-- | Maps a function over the output of the fold.

107+--

108+instance Monad m => Functor (Parse m a) where

109+ {-# INLINE fmap #-}

110+ fmap f (Parse step initial extract) = Parse step initial (fmap4 f extract)

111+

112+ where

113+

114+ fmap4 g = fmap (fmap (fmap (fmap g)))

115+

116+ {-# INLINE (<$) #-}

117+ (<$) b = \_ -> pure b

118+

119+data SeqParseState sl f sr =

120+ SeqParseL sl | SeqParseR f sr | SeqParseLErr String

121+

122+-- | Apply two parsing folds sequentially to an input. The input is provided to

123+-- the first parser, if the parser fails then whole composition fails. If the

124+-- parser succeeds, the remaining input is supplied to the second parser. If

125+-- the second parser succeeds the composed parser succeeds otherwise it fails.

126+--

127+instance Monad m => Applicative (Parse m a) where

128+ {-# INLINE pure #-}

129+ pure b = Parse (\_ _ -> return $ Halt ())

130+ (return ())

131+ (\_ -> return $ Right (0, b))

132+

133+ {-# INLINE (<*>) #-}

134+ (Parse stepL initialL extractL) <*> (Parse stepR initialR extractR) =

135+ Parse step initial extract

136+

137+ where

138+

139+ initial = SeqParseL <$> initialL

140+

141+ -- Note: For the composed parse to terminate, the left parser has to be

142+ -- a terminating parser returning a Halt at some point.

143+ step (SeqParseL st) a = do

144+ r <- stepL st a

145+ case r of

146+ Keep n s -> return $ Keep n (SeqParseL s)

147+ Hold s -> return $ Hold (SeqParseL s)

148+ Back n s -> return $ Back n (SeqParseL s)

149+ Halt s -> do

150+ res <- extractL s

151+ case res of

152+ Left err -> return $ Halt (SeqParseLErr err)

153+ Right (n, f) -> Back n <$> (SeqParseR f <$> initialR)

154+

155+ step (SeqParseR f st) a = do

156+ r <- stepR st a

157+ return $ case r of

158+ Keep n s -> Keep n (SeqParseR f s)

159+ Hold s -> Hold (SeqParseR f s)

160+ Back n s -> Back n (SeqParseR f s)

161+ Halt s -> Halt (SeqParseR f s)

162+

163+ step s@(SeqParseLErr _) _ = return $ Halt s

164+

165+ -- XXX bimap to add "<*>: Right parse failed" to the error message

166+ extract (SeqParseR f s) = fmap (fmap (fmap f)) (extractR s)

167+ extract (SeqParseL _) = return $ Left "<*>: Incomplete left parse"

168+ extract (SeqParseLErr err) =

169+ return $ Left $ "<*>: Left parse failed\n" ++ err

src/Streamly/Internal/Data/Stream/Prelude.hs+13

@@ -38,6 +38,7 @@ module Streamly.Internal.Data.Stream.Prelude

38 , foldlMx'

39 , foldl'

40 , runFold

41+ , parselMx'

42 -- Lazy left folds are useful only for reversing the stream

43 -- Lazy left folds are useful only for reversing the stream

43 , foldlS

44 , foldlS

@@ -72,6 +73,7 @@ import Prelude hiding (foldr, minimum, maximum)

72import qualified Prelude

73import qualified Prelude

74import Streamly.Internal.Data.Fold.Types (Fold (..))

75import Streamly.Internal.Data.Fold.Types (Fold (..))

76+import Streamly.Internal.Data.Parse.Types (Step)

76#ifdef USE_STREAMK_ONLY

78#ifdef USE_STREAMK_ONLY

77import qualified Streamly.Internal.Data.Stream.StreamK as S

79import qualified Streamly.Internal.Data.Stream.StreamK as S

@@ -157,6 +159,17 @@ foldlMx' :: (IsStream t, Monad m)

157 => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b

159 => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b

158foldlMx' step begin done m = S.foldlMx' step begin done $ toStreamS m

160foldlMx' step begin done m = S.foldlMx' step begin done $ toStreamS m

159

161

162+{-# INLINE parselMx' #-}

163+parselMx'

164+ :: (IsStream t, Monad m)

165+ => (s -> a -> m (Step s))

166+ -> m s

167+ -> (s -> m (Either String (Int, b)))

168+ -> t m a

169+ -> m (Either String b)

170+parselMx' step initial extract m =

171+ D.parselMx' step initial extract $ D.toStreamD m

172+

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

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

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

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

162-- argument) to the folded value at the end. This is designed to work with the

175-- argument) to the folded value at the end. This is designed to work with the

src/Streamly/Internal/Data/Stream/StreamD.hs+186−2

@@ -128,6 +128,9 @@ module Streamly.Internal.Data.Stream.StreamD

128 , foldlMx'

129 , runFold

130

131+ , parselMx'

132+ , parseChunks

133+

131 -- ** Specialized Folds

134 -- ** Specialized Folds

132 , tap

135 , tap

133 , tapOffsetEvery

136 , tapOffsetEvery

@@ -320,7 +323,7 @@ where

320

323

321import Control.Concurrent (killThread, myThreadId, takeMVar, threadDelay)

324import Control.Concurrent (killThread, myThreadId, takeMVar, threadDelay)

322import Control.Exception

325import Control.Exception

323− (Exception, SomeException, AsyncException, fromException, mask_)

326+ (assert, Exception, SomeException, AsyncException, fromException, mask_)

324import Control.Monad (void, when, forever)

327import Control.Monad (void, when, forever)

325import Control.Monad.Catch (MonadCatch, throwM)

328import Control.Monad.Catch (MonadCatch, throwM)

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

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

@@ -343,13 +346,14 @@ import Prelude

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

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

344 notElem, null, head, tail, zipWith, lookup, foldr1, sequence,

347 notElem, null, head, tail, zipWith, lookup, foldr1, sequence,

345 (!!), scanl, scanl1, concatMap, replicate, enumFromTo, concat,

348 (!!), scanl, scanl1, concatMap, replicate, enumFromTo, concat,

346− reverse, iterate)

349+ reverse, iterate, splitAt)

347

350

348import qualified Control.Monad.Catch as MC

351import qualified Control.Monad.Catch as MC

349import qualified Control.Monad.Reader as Reader

352import qualified Control.Monad.Reader as Reader

350import qualified Control.Monad.State.Strict as State

353import qualified Control.Monad.State.Strict as State

351import qualified Prelude

354import qualified Prelude

352

355

356+import Fusion.Plugin.Types (Fuse(..))

353import Streamly.Internal.Mutable.Prim.Var

357import Streamly.Internal.Mutable.Prim.Var

354 (Prim, Var, readVar, newVar, modifyVar')

358 (Prim, Var, readVar, newVar, modifyVar')

355import Streamly.Internal.Data.Time.Units

359import Streamly.Internal.Data.Time.Units

@@ -358,6 +362,7 @@ import Streamly.Internal.Data.Time.Units

358import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)

362import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)

359import Streamly.Internal.Memory.Array.Types (Array(..))

363import Streamly.Internal.Memory.Array.Types (Array(..))

360import Streamly.Internal.Data.Fold.Types (Fold(..))

364import Streamly.Internal.Data.Fold.Types (Fold(..))

365+import Streamly.Internal.Data.Parse.Types (Parse(..))

361import Streamly.Internal.Data.Pipe.Types (Pipe(..), PipeState(..))

366import Streamly.Internal.Data.Pipe.Types (Pipe(..), PipeState(..))

362import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

367import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

363import Streamly.Internal.Data.Time.Units

368import Streamly.Internal.Data.Time.Units

@@ -374,6 +379,7 @@ import qualified Streamly.Internal.Memory.Array.Types as A

374import qualified Streamly.Internal.Data.Fold as FL

379import qualified Streamly.Internal.Data.Fold as FL

375import qualified Streamly.Memory.Ring as RB

380import qualified Streamly.Memory.Ring as RB

376import qualified Streamly.Internal.Data.Stream.StreamK as K

381import qualified Streamly.Internal.Data.Stream.StreamK as K

382+import qualified Streamly.Internal.Data.Parse.Types as PR

377

383

378------------------------------------------------------------------------------

384------------------------------------------------------------------------------

379-- Construction

385-- Construction

@@ -942,6 +948,184 @@ foldlS fstep begin (Stream step state) = Stream step' (Left (state, begin))

942 Skip s -> Skip (Right (Stream stp s))

948 Skip s -> Skip (Right (Stream stp s))

943 Stop -> Stop

949 Stop -> Stop

944

950

951+------------------------------------------------------------------------------

952+-- Parses

953+------------------------------------------------------------------------------

954+

955+-- Inlined definition. Without the inline "serially/parser/take" benchmark

956+-- degrades and parseChunks does not fuse. Even using "inline" at the callsite

957+-- does not help.

958+{-# INLINE splitAt #-}

959+splitAt :: Int -> [a] -> ([a],[a])

960+splitAt n ls

961+ | n <= 0 = ([], ls)

962+ | otherwise = splitAt' n ls

963+ where

964+ splitAt' :: Int -> [a] -> ([a], [a])

965+ splitAt' _ [] = ([], [])

966+ splitAt' 1 (x:xs) = ([x], xs)

967+ splitAt' m (x:xs) = (x:xs', xs'')

968+ where

969+ (xs', xs'') = splitAt' (m - 1) xs

970+

971+-- | Run a 'Parse' over a stream.

972+{-# INLINE_NORMAL parselMx' #-}

973+parselMx'

974+ :: Monad m

975+ => (s -> a -> m (PR.Step s))

976+ -> m s

977+ -> (s -> m (Either String (Int, b)))

978+ -> Stream m a

979+ -> m (Either String b)

980+parselMx' pstep initial extract (Stream step state) = do

981+ initial >>= go SPEC state []

982+

983+ where

984+

985+ -- XXX currently we are using a dumb list based approach for backtracking

986+ -- buffer. This can be replaced by a sliding/ring buffer using Data.Array.

987+ -- That will allow us more efficient random back and forth movement.

988+ {-# INLINE go #-}

989+ go !_ st buf !pst = do

990+ r <- step defState st

991+ case r of

992+ Yield x s -> do

993+ pRes <- pstep pst x

994+ case pRes of

995+ -- PR.Keep 0 pst1 -> go SPEC s [] pst1

996+ PR.Keep n pst1 -> do

997+ assert (n <= length (x:buf)) (return ())

998+ go SPEC s (Prelude.take n (x:buf)) pst1

999+ PR.Back n pst1 -> do

1000+ assert (n <= length (x:buf)) (return ())

1001+ let (src0, buf1) = splitAt n (x:buf)

1002+ src = Prelude.reverse src0

1003+ gobuf SPEC s buf1 src pst1

1004+ PR.Hold pst1 -> go SPEC s (x:buf) pst1

1005+ PR.Halt pst1 -> fmap (fmap snd) (extract pst1)

1006+ Skip s -> go SPEC s buf pst

1007+ Stop -> fmap (fmap snd) (extract pst)

1008+

1009+ gobuf !_ s buf [] !pst = go SPEC s buf pst

1010+ gobuf !_ s buf (x:xs) !pst = do

1011+ pRes <- pstep pst x

1012+ case pRes of

1013+ -- PR.Keep 0 pst1 -> go SPEC s [] pst1

1014+ PR.Keep n pst1 -> do

1015+ assert (n <= length (x:buf)) (return ())

1016+ gobuf SPEC s (Prelude.take n (x:buf)) xs pst1

1017+ PR.Back n pst1 -> do

1018+ assert (n <= length (x:buf)) (return ())

1019+ let (src0, buf1) = splitAt n (x:buf)

1020+ src = Prelude.reverse src0 ++ xs

1021+ gobuf SPEC s buf1 src pst1

1022+ PR.Hold pst1 -> gobuf SPEC s (x:buf) xs pst1

1023+ PR.Halt pst1 -> fmap (fmap snd) (extract pst1)

1024+

1025+------------------------------------------------------------------------------

1026+-- Repeated parsing

1027+------------------------------------------------------------------------------

1028+

1029+{-# ANN type ParseChunksState Fuse #-}

1030+data ParseChunksState x inpBuf st pst =

1031+ ParseChunksInit inpBuf st

1032+ | ParseChunksInitLeftOver inpBuf

1033+ | ParseChunksStream st inpBuf pst

1034+ | ParseChunksBuf inpBuf st inpBuf pst

1035+ | ParseChunksYield x (ParseChunksState x inpBuf st pst)

1036+

1037+{-# INLINE_NORMAL parseChunks #-}

1038+parseChunks

1039+ :: Monad m

1040+ => Parse m a b

1041+ -> Stream m a

1042+ -> Stream m b

1043+parseChunks (Parse pstep initial extract) (Stream step state) =

1044+ Stream stepOuter (ParseChunksInit [] state)

1045+

1046+ where

1047+

1048+ {-# INLINE_LATE stepOuter #-}

1049+ -- Buffer is empty, go to stream processing loop

1050+ stepOuter _ (ParseChunksInit [] st) = do

1051+ initial >>= return . Skip . ParseChunksStream st []

1052+

1053+ -- Buffer is not empty, go to buffered processing loop

1054+ stepOuter _ (ParseChunksInit src st) = do

1055+ initial >>= return . Skip . ParseChunksBuf src st []

1056+

1057+ -- XXX we just discard any leftover input at the end

1058+ stepOuter _ (ParseChunksInitLeftOver _) = return Stop

1059+

1060+ -- Buffer is empty process elements from the stream

1061+ stepOuter gst (ParseChunksStream st buf pst) = do

1062+ r <- step (adaptState gst) st

1063+ case r of

1064+ Yield x s -> do

1065+ pRes <- pstep pst x

1066+ case pRes of

1067+ -- PR.Keep 0 pst1 -> go SPEC s [] pst1

1068+ PR.Keep n pst1 -> do

1069+ assert (n <= length (x:buf)) (return ())

1070+ let buf1 = Prelude.take n (x:buf)

1071+ return $ Skip $ ParseChunksStream s buf1 pst1

1072+ PR.Back n pst1 -> do

1073+ assert (n <= length (x:buf)) (return ())

1074+ let (src0, buf1) = splitAt n (x:buf)

1075+ src = Prelude.reverse src0

1076+ return $ Skip $ ParseChunksBuf src s buf1 pst1

1077+ PR.Hold pst1 ->

1078+ return $ Skip $ ParseChunksStream s (x:buf) pst1

1079+ PR.Halt pst1 -> do

1080+ res <- extract pst1

1081+ case res of

1082+ Left err -> error err

1083+ -- XXX (0,a) common case

1084+ Right (n, a) -> do

1085+ assert (n <= length (x:buf)) (return ())

1086+ let src = Prelude.reverse (Prelude.take n (x:buf))

1087+ return $ Skip $ ParseChunksYield a (ParseChunksInit src s)

1088+ Skip s -> return $ Skip $ ParseChunksStream s buf pst

1089+ Stop -> do

1090+ res <- extract pst

1091+ case res of

1092+ Left err -> error err

1093+ Right (n, a) -> do

1094+ let src = Prelude.reverse (Prelude.take n (buf))

1095+ return $ Skip $ ParseChunksYield a (ParseChunksInitLeftOver src)

1096+

1097+ -- go back to stream processing mode

1098+ stepOuter _ (ParseChunksBuf [] s buf pst) =

1099+ return $ Skip $ ParseChunksStream s buf pst

1100+

1101+ -- buffered processing loop

1102+ stepOuter _ (ParseChunksBuf (x:xs) s buf pst) = do

1103+ pRes <- pstep pst x

1104+ case pRes of

1105+ -- PR.Keep 0 pst1 ->

1106+ PR.Keep n pst1 -> do

1107+ assert (n <= length (x:buf)) (return ())

1108+ let buf1 = Prelude.take n (x:buf)

1109+ return $ Skip $ ParseChunksBuf xs s buf1 pst1

1110+ PR.Back n pst1 -> do

1111+ assert (n <= length (x:buf)) (return ())

1112+ let (src0, buf1) = splitAt n (x:buf)

1113+ src = Prelude.reverse src0 ++ xs

1114+ return $ Skip $ ParseChunksBuf src s buf1 pst1

1115+ PR.Hold pst1 ->

1116+ return $ Skip $ ParseChunksBuf xs s (x:buf) pst1

1117+ PR.Halt pst1 -> do

1118+ res <- extract pst1

1119+ case res of

1120+ Left err -> error err

1121+ -- XXX have (0,a) case

1122+ Right (n, a) -> do

1123+ assert (n <= length (x:buf)) (return ())

1124+ let src = Prelude.reverse (Prelude.take n (x:buf)) ++ xs

1125+ return $ Skip $ ParseChunksYield a (ParseChunksInit src s)

1126+

1127+ stepOuter _ (ParseChunksYield a next) = return $ Yield a next

1128+

945------------------------------------------------------------------------------

1129------------------------------------------------------------------------------

946-- Specialized Folds

1130-- Specialized Folds

947------------------------------------------------------------------------------

1131------------------------------------------------------------------------------

src/Streamly/Internal/Prelude.hs+42

@@ -92,6 +92,7 @@ module Streamly.Internal.Prelude

93 -- ** Composable Left Folds

94 , fold

95+ , parse

96 -- ** Concurrent Folds

97 -- ** Concurrent Folds

97 , foldAsync

98 , foldAsync

@@ -265,6 +266,9 @@ module Streamly.Internal.Prelude

265 , reverse

266 , reverse

266 , reverse'

267 , reverse'

267

268

269+ -- ** Parsing

270+ , parseChunks

271+

268 -- ** Trimming

272 -- ** Trimming

269 , take

273 , take

270 , takeByTime

274 , takeByTime

@@ -535,6 +539,7 @@ import qualified System.IO as IO

535

539

536import Streamly.Internal.Data.Stream.Enumeration (Enumerable(..), enumerate, enumerateTo)

540import Streamly.Internal.Data.Stream.Enumeration (Enumerable(..), enumerate, enumerateTo)

537import Streamly.Internal.Data.Fold.Types (Fold (..), Fold2 (..))

541import Streamly.Internal.Data.Fold.Types (Fold (..), Fold2 (..))

542+import Streamly.Internal.Data.Parse.Types (Parse (..))

538import Streamly.Internal.Data.Unfold.Types (Unfold)

543import Streamly.Internal.Data.Unfold.Types (Unfold)

539import Streamly.Internal.Memory.Array.Types (Array, writeNUnsafe)

544import Streamly.Internal.Memory.Array.Types (Array, writeNUnsafe)

540-- import Streamly.Memory.Ring (Ring)

545-- import Streamly.Memory.Ring (Ring)

@@ -1188,6 +1193,18 @@ runSink :: Monad m => Sink m a -> SerialT m a -> m ()

1188runSink = fold . toFold

1193runSink = fold . toFold

1189-}

1194-}

1190

1195

1196+------------------------------------------------------------------------------

1197+-- Running a Parse

1198+------------------------------------------------------------------------------

1199+

1200+-- | Parse a stream using the supplied 'Parse'.

1201+--

1202+-- /Internal/

1203+--

1204+{-# INLINE parse #-}

1205+parse :: Monad m => Parse m a b -> SerialT m a -> m (Either String b)

1206+parse (Parse step initial extract) = P.parselMx' step initial extract

1207+

1191------------------------------------------------------------------------------

1208------------------------------------------------------------------------------

1192-- Specialized folds

1209-- Specialized folds

1193------------------------------------------------------------------------------

1210------------------------------------------------------------------------------

@@ -3092,6 +3109,31 @@ concatMapTreeYieldLeavesWith

3092 -> t m b

3109 -> t m b

3093concatMapTreeYieldLeavesWith combine f = concatMapLoopWith combine f yield

3110concatMapTreeYieldLeavesWith combine f = concatMapLoopWith combine f yield

3094

3111

3112+------------------------------------------------------------------------------

3113+-- Parsing

3114+------------------------------------------------------------------------------

3115+

3116+-- Splitting operations that take a predicate and a Fold can be

3117+-- expressed using parseChunks. Operations like chunksOf, intervalsOf, split*,

3118+-- can be expressed using parseChunks when used with an appropriate Parse.

3119+--

3120+-- | Apply a 'Parse' repeatedly on a stream and emit the parsed values in the

3121+-- output stream.

3122+--

3123+-- >>> S.toList $ S.parseChunks (PR.take 2 $ PR.fromFold FL.sum) $ S.fromList [1..10]

3124+-- > [3,7,11,15,19]

3125+--

3126+-- >>> S.toList $ S.parseChunks (PR.line FL.toList) $ S.fromList "hello\nworld"

3127+-- > ["hello\n","world"]

3128+--

3129+{-# INLINE parseChunks #-}

3130+parseChunks

3131+ :: (IsStream t, Monad m)

3132+ => Parse m a b

3133+ -> t m a

3134+ -> t m b

3135+parseChunks f m = D.fromStreamD $ D.parseChunks f (D.toStreamD m)

3136+

3095------------------------------------------------------------------------------

3137------------------------------------------------------------------------------

3096-- Grouping/Splitting

3138-- Grouping/Splitting

3097------------------------------------------------------------------------------

3139------------------------------------------------------------------------------

streamly.cabal+2

@@ -324,6 +324,8 @@ library

324 , Streamly.Internal.Memory.ArrayStream

325 , Streamly.Internal.Data.Fold.Types

326 , Streamly.Internal.Data.Fold

327+ , Streamly.Internal.Data.Parse

328+ , Streamly.Internal.Data.Parse.Types

327 , Streamly.Internal.Data.Sink.Types

329 , Streamly.Internal.Data.Sink.Types

328 , Streamly.Internal.Data.Sink

330 , Streamly.Internal.Data.Sink

329

331