git.iko.soy/streamly
on master help sign in
Add framing/buffering of streams using Maybe
ae1d8b1Harendra Kumar
src/Streamly/Internal/FileSystem/Handle.hs+20
compare with master
unifiedsplit
@@ -13,6 +13,11 @@
13-- fundamental stream IO APIs built on top of those are
13-- fundamental stream IO APIs built on top of those are
14-- 'readChunksWithBufferOf' and 'writeChunks'. Rest of this module is just
14-- 'readChunksWithBufferOf' and 'writeChunks'. Rest of this module is just
15-- combinatorial programming using these.
15-- 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.
16
21
17-- TODO: Need a separate module for pread/pwrite based reading writing for
22-- TODO: Need a separate module for pread/pwrite based reading writing for
18-- seekable devices. Stateless read/write can be helpful in multithreaded
23-- seekable devices. Stateless read/write can be helpful in multithreaded
@@ -51,6 +56,7 @@ module Streamly.Internal.FileSystem.Handle
51 -- , writeByFrames
56 -- , writeByFrames
52 -- , writeLines
57 -- , writeLines
53 , writeWithBufferOf
58 , writeWithBufferOf
59+ , writeMaybesWithBufferOf
54
60
55 , putBytes
61 , putBytes
56 , putBytesWithBufferOf
62 , putBytesWithBufferOf
@@ -101,6 +107,7 @@ where
101import Control.Exception (assert)
107import Control.Exception (assert)
102import Control.Monad.IO.Class (MonadIO(..))
108import Control.Monad.IO.Class (MonadIO(..))
103import Data.Function ((&))
109import Data.Function ((&))
110+import Data.Maybe (isNothing, fromJust)
104import Data.Word (Word8)
111import Data.Word (Word8)
105import Foreign.ForeignPtr (withForeignPtr)
112import Foreign.ForeignPtr (withForeignPtr)
106import Foreign.Ptr (minusPtr, plusPtr)
113import Foreign.Ptr (minusPtr, plusPtr)
@@ -509,6 +516,19 @@ writeChunksWithBufferOf n h = lpackArraysChunksOf n (writeChunks h)
509writeWithBufferOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()
516writeWithBufferOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()
510writeWithBufferOf n h = FL.chunksOf n (writeNUnsafe n) (writeChunks h)
517writeWithBufferOf n h = FL.chunksOf n (writeNUnsafe n) (writeChunks h)
511
518
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+
512-- | Like 'writeWithBufferOf' but uses the experimental 'Refold' API.
532-- | Like 'writeWithBufferOf' but uses the experimental 'Refold' API.
513--
533--
514-- /Internal/
534-- /Internal/
src/Streamly/Internal/Network/Socket.hs+15−1
@@ -47,6 +47,7 @@ module Streamly.Internal.Network.Socket
47 -- , writeUtf8ByLines
47 -- , writeUtf8ByLines
48 -- , writeByFrames
48 -- , writeByFrames
49 , writeWithBufferOf
49 , writeWithBufferOf
50+ , writeMaybesWithBufferOf
50
51
51 , putChunks
52 , putChunks
52 , putBytesWithBufferOf
53 , putBytesWithBufferOf
@@ -66,8 +67,8 @@ import Control.Exception (onException)
66import Control.Monad.Catch (MonadCatch, finally, MonadMask)
67import Control.Monad.Catch (MonadCatch, finally, MonadMask)
67import Control.Monad.IO.Class (MonadIO(..))
68import Control.Monad.IO.Class (MonadIO(..))
68import Control.Monad (forM_, when)
69import Control.Monad (forM_, when)
70+import Data.Maybe (isNothing, fromJust)
69import Data.Word (Word8)
71import Data.Word (Word8)
70−
71import Foreign.Ptr (minusPtr, plusPtr, Ptr, castPtr)
72import Foreign.Ptr (minusPtr, plusPtr, Ptr, castPtr)
72import Foreign.Storable (Storable(..))
73import Foreign.Storable (Storable(..))
73import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
74import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
@@ -497,6 +498,19 @@ putBytesWithBufferOf n h m = putChunks h $ AS.arraysOf n m
497writeWithBufferOf :: MonadIO m => Int -> Socket -> Fold m Word8 ()
498writeWithBufferOf :: MonadIO m => Int -> Socket -> Fold m Word8 ()
498writeWithBufferOf n h = FL.chunksOf n (A.writeNUnsafe n) (writeChunks h)
499writeWithBufferOf n h = FL.chunksOf n (A.writeNUnsafe n) (writeChunks h)
499
500
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+
500-- > write = 'writeWithBufferOf' defaultChunkSize
514-- > write = 'writeWithBufferOf' defaultChunkSize
501--
515--
502-- | Write a byte stream to a file handle. Combines the bytes in chunks of size
516-- | Write a byte stream to a file handle. Combines the bytes in chunks of size