82c090d · streamly · git.iko.soy

Changelog.md+1−8

compare with master

unifiedsplit

@@ -35,16 +35,9 @@ multiple folds without breaking the stream. Combinators are provided for

35temporal and spatial window based fold operations, for example, to support

36folding and aggregating data for timeout or inactivity based sessions.

37

38−#### Composable Pipes

39−

40−`Streamly.Pipes` module provides composable pipes (stream consumers and

41−producers). Pipes can partition, split or distribute a stream into multiple

42−streams, apply transformations on each stream and merge back the results into a

43−single stream.

44−

45#### Streaming File IO

38#### Streaming File IO

46

39

47−`Streamly.FileSystem.File` provides handle based streaming file IO

40+`Streamly.FileSystem.Handle` provides handle based streaming file IO

48operations.

41operations.

49

42

50#### Streaming Network IO

43#### Streaming Network IO

benchmark/Linear.hs+5−5

@@ -20,7 +20,7 @@ import qualified LinearOps as Ops

20

21import Streamly

22import qualified Streamly.Fold as FL

23−import qualified Streamly.Pipe as Pipe

23+--import qualified Streamly.Pipe as Pipe

24import qualified Streamly.Mem.Array as A

25import qualified Streamly.Prelude as S

26-- import qualified Streamly.Sink as Sink

@@ -242,15 +242,15 @@ main =

242 , bgroup "folds-transforms"

243 [ benchIOSink "drain" (FL.foldl' FL.drain)

244 , benchIOSink "lmap" (FL.foldl' (FL.lmap (+1) FL.drain))

245− , benchIOSink "pipe-mapM"

245+ {-, benchIOSink "pipe-mapM"

246− (FL.foldl' (FL.transform (Pipe.mapM (\x -> return $ x + 1)) FL.drain))

246+ (FL.foldl' (FL.transform (Pipe.mapM (\x -> return $ x + 1)) FL.drain))-}

247 ]

248 , bgroup "folds-compositions" -- Applicative

249 [

250 benchIOSink "all,any" (FL.foldl' ((,) <$> FL.all (<= Ops.maxValue)

251 <*> FL.any (> Ops.maxValue)))

252 , benchIOSink "sum,length" (FL.foldl' ((,) <$> FL.sum <*> FL.length))

253− ]

253+ ] {-

254 , bgroup "pipes"

255 [ benchIOSink "mapM" (Ops.transformMapM serially 1)

256 , benchIOSink "compose" (Ops.transformComposeMapM serially 1)

@@ -262,7 +262,7 @@ main =

262 , benchIOSink "compose" (Ops.transformComposeMapM serially 4)

263 , benchIOSink "tee" (Ops.transformTeeMapM serially 4)

264 , benchIOSink "zip" (Ops.transformZipMapM serially 4)

265− ]

265+ ] -}

266 , bgroup "transformation"

267 [ benchIOSink "scanl" (Ops.scan 1)

268 , benchIOSink "scanl1'" (Ops.scanl1' 1)

benchmark/LinearOps.hs+15−7

@@ -30,7 +30,7 @@ import GHC.Generics (Generic)

30

31import qualified Streamly as S hiding (foldMapWith, runStream)

32import qualified Streamly.Prelude as S

33−import qualified Streamly.Pipe as Pipe

33+-- import qualified Streamly.Pipe as Pipe

34

35value, maxValue, value2 :: Int

36#ifdef LINEAR_ASYNC

@@ -370,14 +370,9 @@ scan, scanl1', map, fmap, mapMaybe, filterEven, filterAllOut,

370mapMaybeM, intersperse :: S.MonadAsync m => Int -> Stream m Int -> m ()

371

372{-# INLINE mapM #-}

373−{-# INLINE transformMapM #-}

374−{-# INLINE transformComposeMapM #-}

375−{-# INLINE transformTeeMapM #-}

376−{-# INLINE transformZipMapM #-}

377{-# INLINE map' #-}

373{-# INLINE map' #-}

378{-# INLINE fmap' #-}

374{-# INLINE fmap' #-}

379−mapM, map', transformMapM, transformComposeMapM, transformTeeMapM,

380− transformZipMapM :: (S.IsStream t, S.MonadAsync m)

375+mapM, map' :: (S.IsStream t, S.MonadAsync m)

381 => (t m Int -> S.SerialT m Int) -> Int -> t m Int -> m ()

376 => (t m Int -> S.SerialT m Int) -> Int -> t m Int -> m ()

382

377

383fmap' :: (S.IsStream t, S.MonadAsync m, P.Functor (t m))

378fmap' :: (S.IsStream t, S.MonadAsync m, P.Functor (t m))

@@ -395,6 +390,18 @@ map n = composeN n $ S.map (+1)

395map' t n = composeN' n $ t . S.map (+1)

390map' t n = composeN' n $ t . S.map (+1)

396mapM t n = composeN' n $ t . S.mapM return

391mapM t n = composeN' n $ t . S.mapM return

397

392

393+-- Temporarily commented these ops as they depend on the hidden

394+-- Pipe module.

395+{-

396+{-# INLINE transformMapM #-}

397+{-# INLINE transformComposeMapM #-}

398+{-# INLINE transformTeeMapM #-}

399+{-# INLINE transformZipMapM #-}

400+

401+transformMapM, transformComposeMapM, transformTeeMapM,

402+ transformZipMapM :: (S.IsStream t, S.MonadAsync m)

403+ => (t m Int -> S.SerialT m Int) -> Int -> t m Int -> m ()

404+

398transformMapM t n = composeN' n $ t . S.transform (Pipe.mapM return)

405transformMapM t n = composeN' n $ t . S.transform (Pipe.mapM return)

399transformComposeMapM t n = composeN' n $ t . S.transform

406transformComposeMapM t n = composeN' n $ t . S.transform

400 (Pipe.mapM (\x -> return (x + 1))

407 (Pipe.mapM (\x -> return (x + 1))

@@ -405,6 +412,7 @@ transformTeeMapM t n = composeN' n $ t . S.transform

405transformZipMapM t n = composeN' n $ t . S.transform

412transformZipMapM t n = composeN' n $ t . S.transform

406 (Pipe.zipWith (+) (Pipe.mapM (\x -> return (x + 1)))

413 (Pipe.zipWith (+) (Pipe.mapM (\x -> return (x + 1)))

407 (Pipe.mapM (\x -> return (x + 2))))

414 (Pipe.mapM (\x -> return (x + 2))))

415+-}

408

416

409mapMaybe n = composeN n $ S.mapMaybe

417mapMaybe n = composeN n $ S.mapMaybe

410 (\x -> if Prelude.odd x then Nothing else Just x)

418 (\x -> if Prelude.odd x then Nothing else Just x)

examples/FileSinkServer.hs+6−3

@@ -9,21 +9,24 @@ import System.Environment (getArgs)

9

10import Streamly

11import Streamly.String

12−import qualified Streamly.FileSystem.File as File

12+import qualified Streamly.FileSystem.Handle as FH

13import qualified Streamly.Fold as FL

14import qualified Streamly.Mem.Array as A

15import qualified Streamly.Network.Socket as NS

16import qualified Streamly.Network.Server as NS

17import qualified Streamly.Prelude as S

18

19+import System.IO (withFile, IOMode(..))

20+

19main :: IO ()

21main :: IO ()

20main = do

22main = do

21 file <- fmap head getArgs

23 file <- fmap head getArgs

22− File.append file

24+ withFile file AppendMode

25+ (\src -> FH.write src

23 $ encodeChar8Unchecked

26 $ encodeChar8Unchecked

24 $ S.concatMap A.read

27 $ S.concatMap A.read

25 $ S.concatMapBy parallel (flip NS.withSocketS recv)

28 $ S.concatMapBy parallel (flip NS.withSocketS recv)

26− $ NS.connectionsOnAllAddrs 8090

29+ $ NS.connectionsOnAllAddrs 8090)

27

30

28 where

31 where

29

32

examples/FromFileClient.hs+7−3

@@ -7,10 +7,14 @@ import System.Environment (getArgs)

7

8import Streamly

9import qualified Streamly.Prelude as S

10−import qualified Streamly.FileSystem.File as File

10+import qualified Streamly.FileSystem.Handle as FH

11import qualified Streamly.Network.Client as Client

12

13+import System.IO (withFile, IOMode(..))

14+

13main :: IO ()

15main :: IO ()

14main =

16main =

17+ let sendFile file =

18+ withFile file ReadMode $ \src ->

15− let sendFile = Client.writeArrays (127,0,0,1) 8090 . File.readArrays

19+ Client.writeArrays (127, 0, 0, 1) 8090 $ FH.readArrays src

16− in getArgs >>= S.drain . parallely . S.mapM sendFile . S.fromList

20+ in getArgs >>= S.drain . parallely . S.mapM sendFile . S.fromList

src/Streamly/Fold.hs+14−12

@@ -136,7 +136,7 @@ module Streamly.Fold

136 , mapM

137

138 -- ** Mapping

139− , transform

139+ --, transform

140 , lmap

141 --, lsequence

142 , lmapM

@@ -320,6 +320,7 @@ module Streamly.Fold

320 -- , chunksOf

321 , duplicate -- experimental

322

323+ {-

323 -- * Windowed Classification

324 -- * Windowed Classification

324 -- | Split the stream into windows or chunks in space or time. Each window

325 -- | Split the stream into windows or chunks in space or time. Each window

325 -- can be associated with a key, all events associated with a particular

326 -- can be associated with a key, all events associated with a particular

@@ -333,13 +334,13 @@ module Streamly.Fold

333 -- ** Tumbling Windows

334 -- ** Tumbling Windows

334 -- | A new window starts after the previous window is finished.

335 -- | A new window starts after the previous window is finished.

335 -- , classifyChunksOf

336 -- , classifyChunksOf

336− , classifySessionsOf

337+ -- , classifySessionsOf

337

338

338 -- ** Keep Alive Windows

339 -- ** Keep Alive Windows

339 -- | The window size is extended if an event arrives within the specified

340 -- | The window size is extended if an event arrives within the specified

340 -- window size. This can represent sessions with idle or inactive timeout.

341 -- window size. This can represent sessions with idle or inactive timeout.

341 -- , classifyKeepAliveChunks

342 -- , classifyKeepAliveChunks

342− , classifyKeepAliveSessions

343+ -- , classifyKeepAliveSessions

343

344

344 {-

345 {-

345 -- ** Sliding Windows

346 -- ** Sliding Windows

@@ -351,6 +352,7 @@ module Streamly.Fold

351 -- ** Sliding Window Buffers

352 -- ** Sliding Window Buffers

352 -- , slidingChunkBuffer

353 -- , slidingChunkBuffer

353 -- , slidingSessionBuffer

354 -- , slidingSessionBuffer

355+ -}

354 )

356 )

355where

357where

356

358

@@ -492,9 +494,9 @@ mapM f = sequence . fmap f

492-- | Apply a transformation on a 'Fold' using a 'Pipe'.

494-- | Apply a transformation on a 'Fold' using a 'Pipe'.

493--

495--

494-- @since 0.7.0

496-- @since 0.7.0

495−{-# INLINE transform #-}

497+{-# INLINE _transform #-}

496−transform :: Monad m => Pipe m a b -> Fold m b c -> Fold m a c

498+_transform :: Monad m => Pipe m a b -> Fold m b c -> Fold m a c

497−transform (Pipe pstep1 pstep2 pinitial) (Fold fstep finitial fextract) =

499+_transform (Pipe pstep1 pstep2 pinitial) (Fold fstep finitial fextract) =

498 Fold step initial extract

500 Fold step initial extract

499

501

500 where

502 where

@@ -2636,14 +2638,14 @@ classifySessionsBy tick timeout reset (Fold step initial extract) str =

2636-- only if no event is received within the specified session window size.

2638-- only if no event is received within the specified session window size.

2637--

2639--

2638-- @since 0.7.0

2640-- @since 0.7.0

2639−{-# INLINABLE classifyKeepAliveSessions #-}

2641+{-# INLINABLE _classifyKeepAliveSessions #-}

2640−classifyKeepAliveSessions

2642+_classifyKeepAliveSessions

2641 :: (IsStream t, MonadAsync m, Ord k)

2643 :: (IsStream t, MonadAsync m, Ord k)

2642 => Double -- ^ session inactive timeout

2644 => Double -- ^ session inactive timeout

2643 -> Fold m a b -- ^ Fold to be applied to session payload data

2645 -> Fold m a b -- ^ Fold to be applied to session payload data

2644 -> t m (k, a, Bool, AbsTime) -- ^ session key, data, close flag, timestamp

2646 -> t m (k, a, Bool, AbsTime) -- ^ session key, data, close flag, timestamp

2645 -> t m (k, b)

2647 -> t m (k, b)

2646−classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True

2648+_classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True

2647

2649

2648------------------------------------------------------------------------------

2650------------------------------------------------------------------------------

2649-- Keyed tumbling windows

2651-- Keyed tumbling windows

@@ -2685,11 +2687,11 @@ classifyChunksOf wsize = classifyChunksBy wsize False

2685-- the timestamps with a clock resolution of 1 second.

2687-- the timestamps with a clock resolution of 1 second.

2686--

2688--

2687-- @since 0.7.0

2689-- @since 0.7.0

2688−{-# INLINABLE classifySessionsOf #-}

2690+{-# INLINABLE _classifySessionsOf #-}

2689−classifySessionsOf

2691+_classifySessionsOf

2690 :: (IsStream t, MonadAsync m, Ord k)

2692 :: (IsStream t, MonadAsync m, Ord k)

2691 => Double -- ^ time window size

2693 => Double -- ^ time window size

2692 -> Fold m a b -- ^ Fold to be applied to window events

2694 -> Fold m a b -- ^ Fold to be applied to window events

2693 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp

2695 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp

2694 -> t m (k, b)

2696 -> t m (k, b)

2695−classifySessionsOf interval = classifySessionsBy 1 interval False

2697+_classifySessionsOf interval = classifySessionsBy 1 interval False

src/Streamly/Prelude.hs+4−4

@@ -240,7 +240,7 @@ module Streamly.Prelude

240

241 -- * Transformation

242

243− , transform

243+ --, transform

244

245 -- ** Mapping

246 -- | In imperative terms a map operation can be considered as a loop over

@@ -1581,9 +1581,9 @@ toHandle h m = go m

1581------------------------------------------------------------------------------

1582

1583-- | Use a 'Pipe' to transform a stream.

1584−{-# INLINE transform #-}

1584+{-# INLINE _transform #-}

1585−transform :: (IsStream t, Monad m) => Pipe m a b -> t m a -> t m b

1585+_transform :: (IsStream t, Monad m) => Pipe m a b -> t m a -> t m b

1586−transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)

1586+_transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)

1587

1588------------------------------------------------------------------------------

1589-- Transformation by Folding (Scans)

streamly.cabal+8−8

@@ -204,27 +204,27 @@ library

204 , Streamly.Sink.Types

205 , Streamly.Sink

206 , Streamly.Pipe.Types

207+ , Streamly.Pipe

207

208

208 , Streamly.FileSystem.IOVec

209 , Streamly.FileSystem.IOVec

209 , Streamly.FileSystem.FDIO

210 , Streamly.FileSystem.FDIO

211+ , Streamly.FileSystem.FD

212+ , Streamly.FileSystem.File

213+

214+ -- Time

215+ , Streamly.Time.Units

216+ , Streamly.Time.Clock

210

217

211 exposed-modules: Streamly.Prelude

218 exposed-modules: Streamly.Prelude

212 , Streamly.Time

219 , Streamly.Time

213 , Streamly

220 , Streamly

214 , Streamly.Fold

221 , Streamly.Fold

215− , Streamly.Pipe

216

222

217 , Streamly.String

223 , Streamly.String

218

224

219 -- IO devices

225 -- IO devices

220 , Streamly.Mem.Array

226 , Streamly.Mem.Array

221− , Streamly.FileSystem.FD

222 , Streamly.FileSystem.Handle

227 , Streamly.FileSystem.Handle

223− , Streamly.FileSystem.File

224−

225− -- Time

226− , Streamly.Time.Units

227− , Streamly.Time.Clock

228

229 , Streamly.Tutorial

230 , Streamly.Internal

@@ -1041,7 +1041,7 @@ executable FileIOExamples

1041 hs-source-dirs: examples

1042 ghc-options: -Wall

1043 if flag(examples) || flag(examples-sdl)

1044− buildable: True

1044+ buildable: False

1045 build-Depends:

1046 streamly

1047 , base >= 4.8 && < 5