git.iko.soy/streamly
on master help sign in
Add write folds, fromBytes/toBytes for File module
5c4bbc5Harendra Kumar
examples/CamelCase.hs+2−2
compare with master
unifiedsplit
@@ -10,12 +10,12 @@ import qualified Streamly.Internal.FileSystem.Handle as FH
10
10
11camelCase :: Handle -> Handle -> IO ()
11camelCase :: Handle -> Handle -> IO ()
12camelCase src dst =
12camelCase src dst =
13− FH.fromStream dst
13+ FH.fromBytes dst
14 $ S.map fromJust
14 $ S.map fromJust
15 $ S.filter isJust
15 $ S.filter isJust
16 $ S.map snd
16 $ S.map snd
17 $ S.scanl' step (True, Nothing)
17 $ S.scanl' step (True, Nothing)
18− $ FH.toStream src
18+ $ FH.toBytes src
19
19
20 where
20 where
21
21
examples/FileIOExamples.hs+6−6
@@ -9,12 +9,12 @@ import System.Environment (getArgs)
9
9
10cat :: FilePath -> IO ()
10cat :: FilePath -> IO ()
11cat src =
11cat src =
12− File.writeChunks "/dev/stdout"
12+ File.fromChunks "/dev/stdout"
13 $ File.toChunksWithBufferOf (256*1024) src
13 $ File.toChunksWithBufferOf (256*1024) src
14
14
15cp :: FilePath -> FilePath -> IO ()
15cp :: FilePath -> FilePath -> IO ()
16cp src dst =
16cp src dst =
17− File.writeChunks dst
17+ File.fromChunks dst
18 $ File.toChunksWithBufferOf (256*1024) src
18 $ File.toChunksWithBufferOf (256*1024) src
19
19
20append :: FilePath -> FilePath -> IO ()
20append :: FilePath -> FilePath -> IO ()
@@ -28,17 +28,17 @@ ord' = (fromIntegral . ord)
28wcl :: FilePath -> IO ()
28wcl :: FilePath -> IO ()
29wcl src = print =<< (S.length
29wcl src = print =<< (S.length
30 $ S.splitOnSuffix (== ord' '\n') FL.drain
30 $ S.splitOnSuffix (== ord' '\n') FL.drain
31− $ File.read src)
31+ $ File.toBytes src)
32
32
33grepc :: String -> FilePath -> IO ()
33grepc :: String -> FilePath -> IO ()
34grepc pat src = print . (subtract 1) =<< (S.length
34grepc pat src = print . (subtract 1) =<< (S.length
35 $ IP.splitOnSeq (A.fromList (map ord' pat)) FL.drain
35 $ IP.splitOnSeq (A.fromList (map ord' pat)) FL.drain
36− $ File.read src)
36+ $ File.toBytes src)
37
37
38avgll :: FilePath -> IO ()
38avgll :: FilePath -> IO ()
39avgll src = print =<< (S.fold avg
39avgll src = print =<< (S.fold avg
40 $ S.splitOnSuffix (== ord' '\n') FL.length
40 $ S.splitOnSuffix (== ord' '\n') FL.length
41− $ File.read src)
41+ $ File.toBytes src)
42 where avg = (/) <$> toDouble FL.sum <*> toDouble FL.length
42 where avg = (/) <$> toDouble FL.sum <*> toDouble FL.length
43 toDouble = fmap (fromIntegral :: Int -> Double)
43 toDouble = fmap (fromIntegral :: Int -> Double)
44
44
@@ -46,7 +46,7 @@ llhisto :: FilePath -> IO ()
46llhisto src = print =<< (S.fold (FL.classify FL.length)
46llhisto src = print =<< (S.fold (FL.classify FL.length)
47 $ S.map bucket
47 $ S.map bucket
48 $ S.splitOnSuffix (== ord' '\n') FL.length
48 $ S.splitOnSuffix (== ord' '\n') FL.length
49− $ File.read src)
49+ $ File.toBytes src)
50 where
50 where
51 bucket n = let i = n `div` 10 in if i > 9 then (9,n) else (i,n)
51 bucket n = let i = n `div` 10 in if i > 9 then (9,n) else (i,n)
52
52
src/Streamly/Benchmark/FileIO/Stream.hs+3−3
@@ -188,7 +188,7 @@ inspect $ 'cat `hasNoType` ''D.ConcatMapUState
188-- | Send the file contents to /dev/null
188-- | Send the file contents to /dev/null
189{-# INLINE catStreamWrite #-}
189{-# INLINE catStreamWrite #-}
190catStreamWrite :: Handle -> Handle -> IO ()
190catStreamWrite :: Handle -> Handle -> IO ()
191−catStreamWrite devNull inh = IFH.fromStream devNull $ S.unfold FH.read inh
191+catStreamWrite devNull inh = IFH.fromBytes devNull $ S.unfold FH.read inh
192
192
193#ifdef INSPECTION
193#ifdef INSPECTION
194inspect $ hasNoTypeClasses 'catStreamWrite
194inspect $ hasNoTypeClasses 'catStreamWrite
@@ -216,8 +216,8 @@ inspect $ hasNoTypeClasses 'catBracket
216catBracketStream :: Handle -> Handle -> IO ()
216catBracketStream :: Handle -> Handle -> IO ()
217catBracketStream devNull inh =
217catBracketStream devNull inh =
218 let readEx = S.bracket (return ()) (\_ -> hClose inh)
218 let readEx = S.bracket (return ()) (\_ -> hClose inh)
219− (\_ -> IFH.toStream inh)
219+ (\_ -> IFH.toBytes inh)
220− in IFH.fromStream devNull $ readEx
220+ in IFH.fromBytes devNull $ readEx
221
221
222#ifdef INSPECTION
222#ifdef INSPECTION
223inspect $ hasNoTypeClasses 'catBracketStream
223inspect $ hasNoTypeClasses 'catBracketStream
src/Streamly/Internal/FileSystem/File.hs+110−16
@@ -61,11 +61,12 @@ module Streamly.Internal.FileSystem.File
61 -- , readFrames
61 -- , readFrames
62 -- , readChunks
62 -- , readChunks
63
63
64+ , toBytes
65+
64 -- -- * Array Read
66 -- -- * Array Read
65 -- , readArrayOf
67 -- , readArrayOf
66
68
67 , toChunksWithBufferOf
69 , toChunksWithBufferOf
68− -- , toChunksWithBufferOf
69 , toChunks
70 , toChunks
70
71
71 -- ** Write To File
72 -- ** Write To File
@@ -75,8 +76,12 @@ module Streamly.Internal.FileSystem.File
75 -- , writeByFrames
76 -- , writeByFrames
76 , writeWithBufferOf
77 , writeWithBufferOf
77
78
79+ , fromBytes
80+ , fromBytesWithBufferOf
81+
78 -- -- * Array Write
82 -- -- * Array Write
79 , writeArray
83 , writeArray
84+ , writeChunks
80 , fromChunks
85 , fromChunks
81
86
82 -- ** Append To File
87 -- ** Append To File
@@ -95,17 +100,24 @@ import Foreign.Storable (Storable(..))
95import System.IO (Handle, openFile, IOMode(..), hClose)
100import System.IO (Handle, openFile, IOMode(..), hClose)
96import Prelude hiding (read)
101import Prelude hiding (read)
97
102
103+import qualified Control.Monad.Catch as MC
98import qualified System.IO as SIO
104import qualified System.IO as SIO
99
105
106+import Streamly.Internal.Data.Fold.Types (Fold(..))
107+import Streamly.Internal.Data.Unfold.Types (Unfold(..))
108+import Streamly.Internal.Memory.Array.Types
100−import Streamly.Internal.Memory.Array.Types (Array(..), defaultChunkSize)
109+ (Array(..), defaultChunkSize, writeNUnsafe)
101import Streamly.Streams.Serial (SerialT)
110import Streamly.Streams.Serial (SerialT)
102import Streamly.Streams.StreamK.Type (IsStream)
111import Streamly.Streams.StreamK.Type (IsStream)
103import Streamly.Internal.Data.SVar (MonadAsync)
112import Streamly.Internal.Data.SVar (MonadAsync)
104-- import Streamly.Data.Fold (Fold)
113-- import Streamly.Data.Fold (Fold)
105-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
114-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
106
115
116+import qualified Streamly.Internal.Data.Fold.Types as FL
117+import qualified Streamly.Internal.Data.Unfold as UF
107import qualified Streamly.Internal.FileSystem.Handle as FH
118import qualified Streamly.Internal.FileSystem.Handle as FH
108import qualified Streamly.Internal.Memory.ArrayStream as AS
119import qualified Streamly.Internal.Memory.ArrayStream as AS
120+import qualified Streamly.Memory.Array as A
109import qualified Streamly.Prelude as S
121import qualified Streamly.Prelude as S
110
122
111-------------------------------------------------------------------------------
123-------------------------------------------------------------------------------
@@ -131,11 +143,29 @@ import qualified Streamly.Prelude as S
131-- raising an exception. If closing the handle raises an exception, then
143-- raising an exception. If closing the handle raises an exception, then
132-- this exception will be raised by 'withFile' rather than any exception
144-- this exception will be raised by 'withFile' rather than any exception
133-- raised by 'act'.
145-- raised by 'act'.
146+--
147+-- /Internal/
148+--
134{-# INLINE withFile #-}
149{-# INLINE withFile #-}
135withFile :: (IsStream t, MonadCatch m, MonadIO m)
150withFile :: (IsStream t, MonadCatch m, MonadIO m)
136 => FilePath -> IOMode -> (Handle -> t m a) -> t m a
151 => FilePath -> IOMode -> (Handle -> t m a) -> t m a
137withFile file mode = S.bracket (liftIO $ openFile file mode) (liftIO . hClose)
152withFile file mode = S.bracket (liftIO $ openFile file mode) (liftIO . hClose)
138
153
154+-- | Transform an 'Unfold' from a 'Handle' to an unfold from a 'FilePath'. The
155+-- resulting unfold opens a handle in 'ReadMode', uses it using the supplied
156+-- unfold and then makes sure that the handle is closed on normal termination
157+-- or in case of an exception. If closing the handle raises an exception, then
158+-- this exception will be raised by 'usingFile'.
159+--
160+-- /Internal/
161+--
162+{-# INLINABLE usingFile #-}
163+usingFile :: (MonadCatch m, MonadIO m)
164+ => Unfold m Handle a -> Unfold m FilePath a
165+usingFile =
166+ UF.bracket (\file -> liftIO $ openFile file ReadMode)
167+ (liftIO . hClose)
168+
139-------------------------------------------------------------------------------
169-------------------------------------------------------------------------------
140-- Array IO (Input)
170-- Array IO (Input)
141-------------------------------------------------------------------------------
171-------------------------------------------------------------------------------
@@ -195,6 +225,25 @@ toChunks = toChunksWithBufferOf defaultChunkSize
195-- read requests at the same time. For serial case we can use async IO. We can
225-- read requests at the same time. For serial case we can use async IO. We can
196-- also control the read throughput in mbps or IOPS.
226-- also control the read throughput in mbps or IOPS.
197
227
228+{-
229+-- | Unfolds the tuple @(bufsize, filepath)@ into a byte stream, read requests
230+-- to the IO device are performed using buffers of @bufsize@.
231+--
232+-- @since 0.7.0
233+{-# INLINE readWithBufferOf #-}
234+readWithBufferOf :: MonadIO m => Unfold m (Int, FilePath) Word8
235+readWithBufferOf = UF.concat (usingFilexxx FH.readChunksWithBufferOf) A.read
236+-}
237+
238+-- | Unfolds a file path into a byte stream. IO requests to the device are
239+-- performed in sizes of
240+-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
241+--
242+-- @since 0.7.0
243+{-# INLINE read #-}
244+read :: (MonadCatch m, MonadIO m) => Unfold m FilePath Word8
245+read = UF.concat (usingFile FH.readChunks) A.read
246+
198{-
247{-
199-- | @readInChunksOf chunkSize handle@ reads a byte stream from a file
248-- | @readInChunksOf chunkSize handle@ reads a byte stream from a file
200-- handle, reads are performed in chunks of up to @chunkSize@. The stream ends
249-- handle, reads are performed in chunks of up to @chunkSize@. The stream ends
@@ -209,14 +258,15 @@ readInChunksOf chunkSize h = A.flattenArrays $ toChunksWithBufferOf chunkSize h
209-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
258-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
210--
259--
211-- > read = 'readByChunks' defaultChunkSize
260-- > read = 'readByChunks' defaultChunkSize
212−-- | Generate a stream of elements of the given type from a file specified by
261+-- | Generate a stream of bytes from a file specified by path. The stream ends
213−-- path. The stream ends when EOF is encountered. File is locked using multiple
262+-- when EOF is encountered. File is locked using multiple reader and single
214−-- reader and single writer locking mode.
263+-- writer locking mode.
215--
264--
216−-- @since 0.7.0
265+-- /Internal/
266+--
217−{-# INLINE read #-}
267+{-# INLINE toBytes #-}
218−read :: (IsStream t, MonadCatch m, MonadIO m) => FilePath -> t m Word8
268+toBytes :: (IsStream t, MonadCatch m, MonadIO m) => FilePath -> t m Word8
219−read file = AS.concat $ withFile file ReadMode FH.toChunks
269+toBytes file = AS.concat $ withFile file ReadMode FH.toChunks
220
270
221{-
271{-
222-- | Generate a stream of elements of the given type from a file 'Handle'. The
272-- | Generate a stream of elements of the given type from a file 'Handle'. The
@@ -269,10 +319,10 @@ fromChunks = fromChunksMode WriteMode
269-- input elements.
319-- input elements.
270--
320--
271-- @since 0.7.0
321-- @since 0.7.0
272−{-# INLINE writeWithBufferOf #-}
322+{-# INLINE fromBytesWithBufferOf #-}
273−writeWithBufferOf :: (MonadAsync m, MonadCatch m)
323+fromBytesWithBufferOf :: (MonadAsync m, MonadCatch m)
274 => Int -> FilePath -> SerialT m Word8 -> m ()
324 => Int -> FilePath -> SerialT m Word8 -> m ()
275−writeWithBufferOf n file xs = fromChunks file $ AS.arraysOf n xs
325+fromBytesWithBufferOf n file xs = fromChunks file $ AS.arraysOf n xs
276
326
277-- > write = 'writeWithBufferOf' defaultChunkSize
327-- > write = 'writeWithBufferOf' defaultChunkSize
278--
328--
@@ -281,10 +331,10 @@ writeWithBufferOf n file xs = fromChunks file $ AS.arraysOf n xs
281-- truncated to zero size before writing. If the file does not exist it is
331-- truncated to zero size before writing. If the file does not exist it is
282-- created. File is locked using single writer locking mode.
332-- created. File is locked using single writer locking mode.
283--
333--
284−-- @since 0.7.0
334+-- /Internal/
285−{-# INLINE write #-}
335+{-# INLINE fromBytes #-}
286−write :: (MonadAsync m, MonadCatch m) => FilePath -> SerialT m Word8 -> m ()
336+fromBytes :: (MonadAsync m, MonadCatch m) => FilePath -> SerialT m Word8 -> m ()
287−write = writeWithBufferOf defaultChunkSize
337+fromBytes = fromBytesWithBufferOf defaultChunkSize
288
338
289{-
339{-
290{-# INLINE write #-}
340{-# INLINE write #-}
@@ -292,6 +342,50 @@ write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
292write = toHandleWith A.defaultChunkSize
342write = toHandleWith A.defaultChunkSize
293-}
343-}
294
344
345+-- | Write a stream of chunks to a handle. Each chunk in the stream is written
346+-- to the device as a separate IO request.
347+--
348+-- /Internal/
349+{-# INLINE writeChunks #-}
350+writeChunks :: (MonadIO m, MonadCatch m, Storable a)
351+ => FilePath -> Fold m (Array a) ()
352+writeChunks path = Fold step initial extract
353+ where
354+ initial = do
355+ h <- liftIO (openFile path WriteMode)
356+ fld <- FL.initialize (FH.writeChunks h)
357+ `MC.onException` (liftIO $ hClose h)
358+ return (fld, h)
359+ step (fld, h) x = do
360+ r <- FL.runStep fld x `MC.onException` (liftIO $ hClose h)
361+ return (r, h)
362+ extract ((Fold _ initial1 extract1), h) = do
363+ liftIO $ hClose h
364+ initial1 >>= extract1
365+
366+-- | @writeWithBufferOf chunkSize handle@ writes the input stream to @handle@.
367+-- Bytes in the input stream are collected into a buffer until we have a chunk
368+-- of size @chunkSize@ and then written to the IO device.
369+--
370+-- /Internal/
371+{-# INLINE writeWithBufferOf #-}
372+writeWithBufferOf :: (MonadIO m, MonadCatch m)
373+ => Int -> FilePath -> Fold m Word8 ()
374+writeWithBufferOf n path =
375+ FL.lchunksOf n (writeNUnsafe n) (writeChunks path)
376+
377+-- > write = 'writeWithBufferOf' A.defaultChunkSize
378+--
379+-- | Write a byte stream to a file. Accumulates the input in chunks of up to
380+-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize' before writing to
381+-- the IO device.
382+--
383+-- /Internal/
384+--
385+{-# INLINE write #-}
386+write :: (MonadIO m, MonadCatch m) => FilePath -> Fold m Word8 ()
387+write = writeWithBufferOf defaultChunkSize
388+
295-- | Append a stream of arrays to a file.
389-- | Append a stream of arrays to a file.
296--
390--
297-- @since 0.7.0
391-- @since 0.7.0
src/Streamly/Internal/FileSystem/Handle.hs+19−19
@@ -25,8 +25,8 @@ module Streamly.Internal.FileSystem.Handle
25 -- , readFrames
25 -- , readFrames
26 , readWithBufferOf
26 , readWithBufferOf
27
27
28− , toStream
28+ , toBytes
29− , toStreamWithBufferOf
29+ , toBytesWithBufferOf
30
30
31 -- -- * Array Read
31 -- -- * Array Read
32 -- , readArrayUpto
32 -- , readArrayUpto
@@ -47,8 +47,8 @@ module Streamly.Internal.FileSystem.Handle
47 , writeWithBufferOf
47 , writeWithBufferOf
48
48
49 -- Byte stream write (Streams)
49 -- Byte stream write (Streams)
50− , fromStream
50+ , fromBytes
51− , fromStreamWithBufferOf
51+ , fromBytesWithBufferOf
52
52
53 -- -- * Array Write
53 -- -- * Array Write
54 , writeArray
54 , writeArray
@@ -270,13 +270,13 @@ readChunks = UF.supplyFirst readChunksWithBufferOf defaultChunkSize
270readWithBufferOf :: MonadIO m => Unfold m (Int, Handle) Word8
270readWithBufferOf :: MonadIO m => Unfold m (Int, Handle) Word8
271readWithBufferOf = UF.concat readChunksWithBufferOf A.read
271readWithBufferOf = UF.concat readChunksWithBufferOf A.read
272
272
273−-- | @toStreamWithBufferOf bufsize handle@ reads a byte stream from a file
273+-- | @toBytesWithBufferOf bufsize handle@ reads a byte stream from a file
274-- handle, reads are performed in chunks of up to @bufsize@.
274-- handle, reads are performed in chunks of up to @bufsize@.
275--
275--
276-- /Internal/
276-- /Internal/
277−{-# INLINE toStreamWithBufferOf #-}
277+{-# INLINE toBytesWithBufferOf #-}
278−toStreamWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
278+toBytesWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
279−toStreamWithBufferOf chunkSize h = AS.concat $ toChunksWithBufferOf chunkSize h
279+toBytesWithBufferOf chunkSize h = AS.concat $ toChunksWithBufferOf chunkSize h
280
280
281-- TODO
281-- TODO
282-- Generate a stream of elements of the given type from a file 'Handle'.
282-- Generate a stream of elements of the given type from a file 'Handle'.
@@ -294,9 +294,9 @@ read = UF.supplyFirst readWithBufferOf defaultChunkSize
294-- | Generate a byte stream from a file 'Handle'.
294-- | Generate a byte stream from a file 'Handle'.
295--
295--
296-- /Internal/
296-- /Internal/
297−{-# INLINE toStream #-}
297+{-# INLINE toBytes #-}
298−toStream :: (IsStream t, MonadIO m) => Handle -> t m Word8
298+toBytes :: (IsStream t, MonadIO m) => Handle -> t m Word8
299−toStream = AS.concat . toChunks
299+toBytes = AS.concat . toChunks
300
300
301-------------------------------------------------------------------------------
301-------------------------------------------------------------------------------
302-- Writing
302-- Writing
@@ -353,15 +353,15 @@ fromChunksWithBufferOf :: (MonadIO m, Storable a)
353 => Int -> Handle -> SerialT m (Array a) -> m ()
353 => Int -> Handle -> SerialT m (Array a) -> m ()
354fromChunksWithBufferOf n h xs = fromChunks h $ AS.compact n xs
354fromChunksWithBufferOf n h xs = fromChunks h $ AS.compact n xs
355
355
356−-- | @fromStreamWithBufferOf bufsize handle stream@ writes @stream@ to @handle@
356+-- | @fromBytesWithBufferOf bufsize handle stream@ writes @stream@ to @handle@
357-- in chunks of @bufsize@. A write is performed to the IO device as soon as we
357-- in chunks of @bufsize@. A write is performed to the IO device as soon as we
358-- collect the required input size.
358-- collect the required input size.
359--
359--
360-- @since 0.7.0
360-- @since 0.7.0
361−{-# INLINE fromStreamWithBufferOf #-}
361+{-# INLINE fromBytesWithBufferOf #-}
362−fromStreamWithBufferOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
362+fromBytesWithBufferOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
363−fromStreamWithBufferOf n h m = fromChunks h $ S.arraysOf n m
363+fromBytesWithBufferOf n h m = fromChunks h $ S.arraysOf n m
364−-- fromStreamWithBufferOf n h m = fromChunks h $ AS.arraysOf n m
364+-- fromBytesWithBufferOf n h m = fromChunks h $ AS.arraysOf n m
365
365
366-- > write = 'writeWithBufferOf' A.defaultChunkSize
366-- > write = 'writeWithBufferOf' A.defaultChunkSize
367--
367--
@@ -372,9 +372,9 @@ fromStreamWithBufferOf n h m = fromChunks h $ S.arraysOf n m
372-- need some extra perf boost.
372-- need some extra perf boost.
373--
373--
374-- @since 0.7.0
374-- @since 0.7.0
375−{-# INLINE fromStream #-}
375+{-# INLINE fromBytes #-}
376−fromStream :: MonadIO m => Handle -> SerialT m Word8 -> m ()
376+fromBytes :: MonadIO m => Handle -> SerialT m Word8 -> m ()
377−fromStream = fromStreamWithBufferOf defaultChunkSize
377+fromBytes = fromBytesWithBufferOf defaultChunkSize
378
378
379-- | Write a stream of arrays to a handle. Each array in the stream is written
379-- | Write a stream of arrays to a handle. Each array in the stream is written
380-- to the device as a separate IO request.
380-- to the device as a separate IO request.
src/Streamly/Internal/Network/Inet/TCP.hs+48−43
@@ -35,16 +35,16 @@ module Streamly.Internal.Network.Inet.TCP
35 -- * TCP clients
35 -- * TCP clients
36 -- | IP Address based operations.
36 -- | IP Address based operations.
37 , connect
37 , connect
38− , withServerAddrM
38+ , withConnectionM
39
39
40 -- ** Unfolds
40 -- ** Unfolds
41− , withServerAddr
41+ , usingConnection
42 , read
42 , read
43
43
44 -- ** Streams
44 -- ** Streams
45 , withConnection
45 , withConnection
46 -- *** Source
46 -- *** Source
47− , toStream
47+ , toBytes
48 -- , readUtf8
48 -- , readUtf8
49 -- , readLines
49 -- , readLines
50 -- , readFrames
50 -- , readFrames
@@ -63,11 +63,14 @@ module Streamly.Internal.Network.Inet.TCP
63 -- , writeUtf8
63 -- , writeUtf8
64 -- , writeUtf8ByLines
64 -- , writeUtf8ByLines
65 -- , writeByFrames
65 -- , writeByFrames
66− -- , writeRequestsOf
66+ , writeWithBufferOf
67+ , fromBytes
68+ , fromBytesWithBufferOf
67
69
68 -- -- * Array Write
70 -- -- * Array Write
69 -- , writeArray
71 -- , writeArray
70 , writeChunks
72 , writeChunks
73+ , fromChunks
71 {-
74 {-
72 -- ** Sink Servers
75 -- ** Sink Servers
73
76
@@ -109,7 +112,9 @@ import Streamly.Streams.Serial (SerialT)
109import Streamly.Internal.Memory.Array.Types (Array(..), defaultChunkSize, writeNUnsafe)
112import Streamly.Internal.Memory.Array.Types (Array(..), defaultChunkSize, writeNUnsafe)
110import Streamly.Streams.StreamK.Type (IsStream)
113import Streamly.Streams.StreamK.Type (IsStream)
111
114
115+import qualified Control.Monad.Catch as MC
112import qualified Network.Socket as Net
116import qualified Network.Socket as Net
117+
113import qualified Streamly.Internal.Data.Unfold as UF
118import qualified Streamly.Internal.Data.Unfold as UF
114import qualified Streamly.Internal.Memory.Array as A
119import qualified Streamly.Internal.Memory.Array as A
115import qualified Streamly.Internal.Memory.ArrayStream as AS
120import qualified Streamly.Internal.Memory.ArrayStream as AS
@@ -225,16 +230,16 @@ connect addr port = do
225 return sock
230 return sock
226
231
227-- | Connect to a remote host using IP address and port and run the supplied
232-- | Connect to a remote host using IP address and port and run the supplied
228−-- action on the resulting socket. 'withServerAddrM' makes sure that the
233+-- action on the resulting socket. 'withConnectionM' makes sure that the
229-- socket is closed on normal termination or in case of an exception. If
234-- socket is closed on normal termination or in case of an exception. If
230-- closing the socket raises an exception, then this exception will be raised
235-- closing the socket raises an exception, then this exception will be raised
231−-- by 'withServerAddrM'.
236+-- by 'withConnectionM'.
232--
237--
233-- /Internal/
238-- /Internal/
234−{-# INLINABLE withServerAddrM #-}
239+{-# INLINABLE withConnectionM #-}
235−withServerAddrM :: (MonadMask m, MonadIO m)
240+withConnectionM :: (MonadMask m, MonadIO m)
236 => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> m ()) -> m ()
241 => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> m ()) -> m ()
237−withServerAddrM addr port =
242+withConnectionM addr port =
238 bracket (liftIO $ connect addr port) (liftIO . Net.close)
243 bracket (liftIO $ connect addr port) (liftIO . Net.close)
239
244
240-------------------------------------------------------------------------------
245-------------------------------------------------------------------------------
@@ -245,14 +250,14 @@ withServerAddrM addr port =
245-- address and port. The resulting unfold opens a socket, uses it using the
250-- address and port. The resulting unfold opens a socket, uses it using the
246-- supplied unfold and then makes sure that the socket is closed on normal
251-- supplied unfold and then makes sure that the socket is closed on normal
247-- termination or in case of an exception. If closing the socket raises an
252-- termination or in case of an exception. If closing the socket raises an
248−-- exception, then this exception will be raised by 'withServerAddr'.
253+-- exception, then this exception will be raised by 'usingConnection'.
249--
254--
250-- /Internal/
255-- /Internal/
251−{-# INLINABLE withServerAddr #-}
256+{-# INLINABLE usingConnection #-}
252−withServerAddr :: (MonadCatch m, MonadIO m)
257+usingConnection :: (MonadCatch m, MonadIO m)
253 => Unfold m Socket a
258 => Unfold m Socket a
254 -> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) a
259 -> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) a
255−withServerAddr =
260+usingConnection =
256 UF.bracket (\(addr, port) -> liftIO $ connect addr port)
261 UF.bracket (\(addr, port) -> liftIO $ connect addr port)
257 (liftIO . Net.close)
262 (liftIO . Net.close)
258
263
@@ -284,35 +289,33 @@ withConnection addr port =
284{-# INLINE read #-}
289{-# INLINE read #-}
285read :: (MonadCatch m, MonadIO m)
290read :: (MonadCatch m, MonadIO m)
286 => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Word8
291 => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Word8
287−read = UF.concat (withServerAddr ISK.readChunks) A.read
292+read = UF.concat (usingConnection ISK.readChunks) A.read
288
293
289-- | Read a stream from the supplied IPv4 host address and port number.
294-- | Read a stream from the supplied IPv4 host address and port number.
290--
295--
291-- @since 0.7.0
296-- @since 0.7.0
292−{-# INLINE toStream #-}
297+{-# INLINE toBytes #-}
293−toStream :: (IsStream t, MonadCatch m, MonadIO m)
298+toBytes :: (IsStream t, MonadCatch m, MonadIO m)
294 => (Word8, Word8, Word8, Word8) -> PortNumber -> t m Word8
299 => (Word8, Word8, Word8, Word8) -> PortNumber -> t m Word8
295−toStream addr port = AS.concat $ withConnection addr port ISK.toChunks
300+toBytes addr port = AS.concat $ withConnection addr port ISK.toChunks
296
301
297-------------------------------------------------------------------------------
302-------------------------------------------------------------------------------
298-- Writing
303-- Writing
299-------------------------------------------------------------------------------
304-------------------------------------------------------------------------------
300
305
301−{-
302-- | Write a stream of arrays to the supplied IPv4 host address and port
306-- | Write a stream of arrays to the supplied IPv4 host address and port
303-- number.
307-- number.
304--
308--
305-- @since 0.7.0
309-- @since 0.7.0
306−{-# INLINE writeChunks #-}
310+{-# INLINE fromChunks #-}
307−writeChunks
311+fromChunks
308 :: (MonadCatch m, MonadAsync m)
312 :: (MonadCatch m, MonadAsync m)
309 => (Word8, Word8, Word8, Word8)
313 => (Word8, Word8, Word8, Word8)
310 -> PortNumber
314 -> PortNumber
311 -> SerialT m (Array Word8)
315 -> SerialT m (Array Word8)
312 -> m ()
316 -> m ()
313−writeChunks addr port xs =
317+fromChunks addr port xs =
314− S.drain $ withConnection addr port (\sk -> S.yieldM $ SK.writeChunks sk xs)
318+ S.drain $ withConnection addr port (\sk -> S.yieldM $ ISK.fromChunks sk xs)
315−-}
316
319
317-- | Write a stream of arrays to the supplied IPv4 host address and port
320-- | Write a stream of arrays to the supplied IPv4 host address and port
318-- number.
321-- number.
@@ -320,7 +323,7 @@ writeChunks addr port xs =
320-- @since 0.7.0
323-- @since 0.7.0
321{-# INLINE writeChunks #-}
324{-# INLINE writeChunks #-}
322writeChunks
325writeChunks
323− :: (MonadAsync m)
326+ :: (MonadAsync m, MonadCatch m)
324 => (Word8, Word8, Word8, Word8)
327 => (Word8, Word8, Word8, Word8)
325 -> PortNumber
328 -> PortNumber
326 -> Fold m (Array Word8) ()
329 -> Fold m (Array Word8) ()
@@ -328,56 +331,58 @@ writeChunks addr port = Fold step initial extract
328 where
331 where
329 initial = do
332 initial = do
330 skt <- liftIO (connect addr port)
333 skt <- liftIO (connect addr port)
331− FL.initialize (SK.writeChunks skt)
334+ fld <- FL.initialize (SK.writeChunks skt)
332− step = FL.runStep
335+ `MC.onException` (liftIO $ Net.close skt)
336+ return (fld, skt)
337+ step (fld, skt) x = do
338+ r <- FL.runStep fld x `MC.onException` (liftIO $ Net.close skt)
339+ return (r, skt)
333− extract (Fold _ initial1 extract1) = initial1 >>= extract1
340+ extract ((Fold _ initial1 extract1), skt) = do
341+ liftIO $ Net.close skt
342+ initial1 >>= extract1
334
343
335−{-
336-- | Like 'write' but provides control over the write buffer. Output will
344-- | Like 'write' but provides control over the write buffer. Output will
337-- be written to the IO device as soon as we collect the specified number of
345-- be written to the IO device as soon as we collect the specified number of
338-- input elements.
346-- input elements.
339--
347--
340-- @since 0.7.0
348-- @since 0.7.0
341−{-# INLINE writeRequestsOf #-}
349+{-# INLINE fromBytesWithBufferOf #-}
342−writeRequestsOf
350+fromBytesWithBufferOf
343 :: (MonadCatch m, MonadAsync m)
351 :: (MonadCatch m, MonadAsync m)
344 => Int
352 => Int
345 -> (Word8, Word8, Word8, Word8)
353 -> (Word8, Word8, Word8, Word8)
346 -> PortNumber
354 -> PortNumber
347 -> SerialT m Word8
355 -> SerialT m Word8
348 -> m ()
356 -> m ()
349−writeRequestsOf n addr port m = writeChunks addr port $ AS.arraysOf n m
357+fromBytesWithBufferOf n addr port m = fromChunks addr port $ AS.arraysOf n m
350−-}
351
358
352-- | Like 'write' but provides control over the write buffer. Output will
359-- | Like 'write' but provides control over the write buffer. Output will
353-- be written to the IO device as soon as we collect the specified number of
360-- be written to the IO device as soon as we collect the specified number of
354-- input elements.
361-- input elements.
355--
362--
356-- @since 0.7.0
363-- @since 0.7.0
357−{-# INLINE writeRequestsOf #-}
364+{-# INLINE writeWithBufferOf #-}
358−writeRequestsOf
365+writeWithBufferOf
359− :: (MonadAsync m)
366+ :: (MonadAsync m, MonadCatch m)
360 => Int
367 => Int
361 -> (Word8, Word8, Word8, Word8)
368 -> (Word8, Word8, Word8, Word8)
362 -> PortNumber
369 -> PortNumber
363 -> Fold m Word8 ()
370 -> Fold m Word8 ()
364−writeRequestsOf n addr port =
371+writeWithBufferOf n addr port =
365 FL.lchunksOf n (writeNUnsafe n) (writeChunks addr port)
372 FL.lchunksOf n (writeNUnsafe n) (writeChunks addr port)
366
373
367−{-
368-- | Write a stream to the supplied IPv4 host address and port number.
374-- | Write a stream to the supplied IPv4 host address and port number.
369--
375--
370-- @since 0.7.0
376-- @since 0.7.0
371−{-# INLINE write #-}
377+{-# INLINE fromBytes #-}
372−write :: (MonadCatch m, MonadAsync m)
378+fromBytes :: (MonadCatch m, MonadAsync m)
373 => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Word8 -> m ()
379 => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Word8 -> m ()
374−write = writeRequestsOf defaultChunkSize
380+fromBytes = fromBytesWithBufferOf defaultChunkSize
375−-}
376
381
377-- | Write a stream to the supplied IPv4 host address and port number.
382-- | Write a stream to the supplied IPv4 host address and port number.
378--
383--
379-- @since 0.7.0
384-- @since 0.7.0
380{-# INLINE write #-}
385{-# INLINE write #-}
381−write :: MonadAsync m
386+write :: (MonadAsync m, MonadCatch m)
382 => (Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()
387 => (Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()
383−write = writeRequestsOf defaultChunkSize
388+write = writeWithBufferOf defaultChunkSize
src/Streamly/Internal/Network/Socket.hs+17−13
@@ -35,7 +35,6 @@ module Streamly.Internal.Network.Socket
35 -- , readLines
35 -- , readLines
36 -- , readFrames
36 -- , readFrames
37 -- , readByChunks
37 -- , readByChunks
38− , toStream
39
38
40 -- -- * Array Read
39 -- -- * Array Read
41 -- , readArrayUpto
40 -- , readArrayUpto
@@ -47,6 +46,7 @@ module Streamly.Internal.Network.Socket
47
46
48 , toChunksWithBufferOf
47 , toChunksWithBufferOf
49 , toChunks
48 , toChunks
49+ , toBytes
50
50
51 -- * Write to connection
51 -- * Write to connection
52 , write
52 , write
@@ -55,6 +55,10 @@ module Streamly.Internal.Network.Socket
55 -- , writeByFrames
55 -- , writeByFrames
56 , writeWithBufferOf
56 , writeWithBufferOf
57
57
58+ , fromChunks
59+ , fromBytesWithBufferOf
60+ , fromBytes
61+
58 -- -- * Array Write
62 -- -- * Array Write
59 , writeArray
63 , writeArray
60 , writeChunks
64 , writeChunks
@@ -370,9 +374,9 @@ readWithBufferOf chunkSize h = A.flattenArrays $ readChunksUpto chunkSize h
370-- stream ends when EOF is encountered.
374-- stream ends when EOF is encountered.
371--
375--
372-- @since 0.7.0
376-- @since 0.7.0
373−{-# INLINE toStream #-}
377+{-# INLINE toBytes #-}
374−toStream :: (IsStream t, MonadIO m) => Socket -> t m Word8
378+toBytes :: (IsStream t, MonadIO m) => Socket -> t m Word8
375−toStream = AS.concat . toChunks
379+toBytes = AS.concat . toChunks
376
380
377-- | Unfolds the tuple @(bufsize, socket)@ into a byte stream, read requests
381-- | Unfolds the tuple @(bufsize, socket)@ into a byte stream, read requests
378-- to the socket are performed using buffers of @bufsize@.
382-- to the socket are performed using buffers of @bufsize@.
@@ -398,10 +402,10 @@ read = UF.supplyFirst readWithBufferOf A.defaultChunkSize
398-- | Write a stream of arrays to a handle.
402-- | Write a stream of arrays to a handle.
399--
403--
400-- @since 0.7.0
404-- @since 0.7.0
401−{-# INLINE writeChunksS #-}
405+{-# INLINE fromChunks #-}
402−writeChunksS :: (MonadIO m, Storable a)
406+fromChunks :: (MonadIO m, Storable a)
403 => Socket -> SerialT m (Array a) -> m ()
407 => Socket -> SerialT m (Array a) -> m ()
404−writeChunksS h m = S.mapM_ (liftIO . writeArray h) m
408+fromChunks h m = S.mapM_ (liftIO . writeArray h) m
405
409
406-- | Write a stream of arrays to a socket. Each array in the stream is written
410-- | Write a stream of arrays to a socket. Each array in the stream is written
407-- to the socket as a separate IO request.
411-- to the socket as a separate IO request.
@@ -424,9 +428,9 @@ writeChunks h = FL.drainBy (liftIO . writeArray h)
424-- input elements.
428-- input elements.
425--
429--
426-- @since 0.7.0
430-- @since 0.7.0
427−{-# INLINE writeWithBufferOfS #-}
431+{-# INLINE fromBytesWithBufferOf #-}
428−writeWithBufferOfS :: MonadIO m => Int -> Socket -> SerialT m Word8 -> m ()
432+fromBytesWithBufferOf :: MonadIO m => Int -> Socket -> SerialT m Word8 -> m ()
429−writeWithBufferOfS n h m = writeChunksS h $ AS.arraysOf n m
433+fromBytesWithBufferOf n h m = fromChunks h $ AS.arraysOf n m
430
434
431-- | Write a byte stream to a socket. Accumulates the input in chunks of
435-- | Write a byte stream to a socket. Accumulates the input in chunks of
432-- specified number of bytes before writing.
436-- specified number of bytes before writing.
@@ -443,9 +447,9 @@ writeWithBufferOf n h = FL.lchunksOf n (A.writeNUnsafe n) (writeChunks h)
443-- depends on the 'IOMode' and the current seek position of the handle.
447-- depends on the 'IOMode' and the current seek position of the handle.
444--
448--
445-- @since 0.7.0
449-- @since 0.7.0
446−{-# INLINE _writeS #-}
450+{-# INLINE fromBytes #-}
447−_writeS :: MonadIO m => Socket -> SerialT m Word8 -> m ()
451+fromBytes :: MonadIO m => Socket -> SerialT m Word8 -> m ()
448−_writeS = writeWithBufferOfS A.defaultChunkSize
452+fromBytes = fromBytesWithBufferOf A.defaultChunkSize
449
453
450-- | Write a byte stream to a socket. Accumulates the input in chunks of
454-- | Write a byte stream to a socket. Accumulates the input in chunks of
451-- up to 'A.defaultChunkSize' bytes before writing.
455-- up to 'A.defaultChunkSize' bytes before writing.