ae1d8b1 · streamly · git.iko.soy

git.iko.soy/streamly

Add framing/buffering of streams using Maybe

commitae1d8b1404b4e44b7b6021a506f723a9d47cbaa6 authorHarendra Kumar <harendra@composewell.com> dateSun 7 Nov 2021, 19:55 (5 years ago) parent09d4257

2 files changed+35−1

compare with master

unifiedsplit

src/Streamly/Internal/FileSystem/Handle.hs+20 src/Streamly/Internal/Network/Socket.hs+15−1

src/Streamly/Internal/FileSystem/Handle.hs+20

@@ -13,6 +13,11 @@

1313-- fundamental stream IO APIs built on top of those are

1414-- 'readChunksWithBufferOf' and 'writeChunks'. Rest of this module is just

1515-- combinatorial programming using these.

16+--

17+-- We can achieve line buffering by folding lines in the input stream into a

18+-- stream of arrays using Stream.splitOn or Fold.takeEndBy_ and similar

19+-- operations. One can wrap the input stream in 'Maybe' type and then use

20+-- 'writeMaybesWithBufferOf' to achieve user controlled buffering.

1621

1722-- TODO: Need a separate module for pread/pwrite based reading writing for

1823-- seekable devices. Stateless read/write can be helpful in multithreaded

@@ -51,6 +56,7 @@ module Streamly.Internal.FileSystem.Handle

5156 -- , writeByFrames

5257 -- , writeLines

5358 , writeWithBufferOf

59+ , writeMaybesWithBufferOf

5460

5561 , putBytes

5662 , putBytesWithBufferOf

@@ -101,6 +107,7 @@ where

101107import Control.Exception (assert)

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

103109import Data.Function ((&))

110+import Data.Maybe (isNothing, fromJust)

104111import Data.Word (Word8)

105112import Foreign.ForeignPtr (withForeignPtr)

106113import Foreign.Ptr (minusPtr, plusPtr)

@@ -509,6 +516,19 @@ writeChunksWithBufferOf n h = lpackArraysChunksOf n (writeChunks h)

509516writeWithBufferOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()

510517writeWithBufferOf n h = FL.chunksOf n (writeNUnsafe n) (writeChunks h)

511518

519+-- | Write a stream of 'Maybe' values. Keep buffering the just values in an

520+-- array until a 'Nothing' is encountered or the buffer size exceeds the

521+-- specified limit, at that point flush the buffer to the handle.

522+--

523+-- /Pre-release/

524+{-# INLINE writeMaybesWithBufferOf #-}

525+writeMaybesWithBufferOf :: (MonadIO m )

526+ => Int -> Handle -> Fold m (Maybe Word8) ()

527+writeMaybesWithBufferOf n h =

528+ let writeNJusts = FL.lmap fromJust $ A.writeN n

529+ writeOnNothing = FL.takeEndBy_ isNothing writeNJusts

530+ in FL.many writeOnNothing (writeChunks h)

531+

512532-- | Like 'writeWithBufferOf' but uses the experimental 'Refold' API.

513533--

514534-- /Internal/

src/Streamly/Internal/Network/Socket.hs+15−1

@@ -47,6 +47,7 @@ module Streamly.Internal.Network.Socket

4747 -- , writeUtf8ByLines

4848 -- , writeByFrames

4949 , writeWithBufferOf

50+ , writeMaybesWithBufferOf

5051

5152 , putChunks

5253 , putBytesWithBufferOf

@@ -66,8 +67,8 @@ import Control.Exception (onException)

6667import Control.Monad.Catch (MonadCatch, finally, MonadMask)

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

6869import Control.Monad (forM_, when)

70+import Data.Maybe (isNothing, fromJust)

6971import Data.Word (Word8)

70−

7172import Foreign.Ptr (minusPtr, plusPtr, Ptr, castPtr)

7273import Foreign.Storable (Storable(..))

7374import GHC.ForeignPtr (mallocPlainForeignPtrBytes)

@@ -497,6 +498,19 @@ putBytesWithBufferOf n h m = putChunks h $ AS.arraysOf n m

497498writeWithBufferOf :: MonadIO m => Int -> Socket -> Fold m Word8 ()

498499writeWithBufferOf n h = FL.chunksOf n (A.writeNUnsafe n) (writeChunks h)

499500

501+-- | Write a stream of 'Maybe' values. Keep buffering the 'Just' values in an

502+-- array. Write the array to the 'Handle' as soon as a 'Nothing' is encountered

503+-- or the buffer size exceeds the specified limit.

504+--

505+-- /Pre-release/

506+{-# INLINE writeMaybesWithBufferOf #-}

507+writeMaybesWithBufferOf :: (MonadIO m )

508+ => Int -> Socket -> Fold m (Maybe Word8) ()

509+writeMaybesWithBufferOf n h =

510+ let writeNJusts = FL.lmap fromJust $ A.writeN n

511+ writeOnNothing = FL.takeEndBy_ isNothing writeNJusts

512+ in FL.many writeOnNothing (writeChunks h)

513+

500514-- > write = 'writeWithBufferOf' defaultChunkSize

501515--

502516-- | Write a byte stream to a file handle. Combines the bytes in chunks of size