2{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE MagicHash #-}
5{-# LANGUAGE RecordWildCards #-}
6{-# LANGUAGE UnboxedTuples #-}
11-- Module : Streamly.Internal.Network.Socket
12-- Copyright : (c) 2018 Composewell Technologies
15-- Maintainer : streamly@composewell.com
16-- Stability : experimental
19module Streamly.Internal.Network.Socket
25 -- * Read from connection
44 -- * Write to connection
55 -- reading/writing datagrams
59import Control.Concurrent (threadWaitWrite, rtsSupportsBoundThreads)
60import Control.Monad.Catch (MonadCatch, finally, MonadMask)
61import Control.Monad.IO.Class (MonadIO(..))
62import Control.Monad (when)
63import Data.Word (Word8)
64import Foreign.ForeignPtr (withForeignPtr)
65import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
66import Foreign.Ptr (minusPtr, plusPtr, Ptr, castPtr)
67import Foreign.Storable (Storable(..))
68import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
69import Network.Socket (Socket, sendBuf, recvBuf)
70#if MIN_VERSION_network(3,1,0)
71import Network.Socket (withFdSocket)
73import Network.Socket (fdSocket)
75import Prelude hiding (read)
77import qualified Network.Socket as Net
79import Streamly.Internal.Data.Unfold.Types (Unfold(..))
80import Streamly.Internal.Memory.Array.Types (Array(..))
81import Streamly.Streams.Serial (SerialT)
82import Streamly.Streams.StreamK.Type (IsStream, mkStream)
83import Streamly.Data.Fold (Fold)
84-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
86import qualified Streamly.Data.Fold as FL
87import qualified Streamly.Internal.Data.Fold.Types as FL
88import qualified Streamly.Internal.Data.Unfold as UF
89import qualified Streamly.Memory.Array as A
90import qualified Streamly.Internal.Memory.ArrayStream as AS
91import qualified Streamly.Internal.Memory.Array.Types as A
92import qualified Streamly.Prelude as S
93import qualified Streamly.Streams.StreamD.Type as D
95-- | @'useSocketM' socket act@ runs the monadic computation @act@ passing the
96-- socket handle to it. The handle will be closed on exit from 'useSocketM',
97-- whether by normal termination or by raising an exception. If closing the
98-- handle raises an exception, then this exception will be raised by
99-- 'useSocketM' rather than any exception raised by 'act'.
102{-# INLINE useSocketM #-}
103useSocketM :: (MonadMask m, MonadIO m) => Socket -> (Socket -> m ()) -> m ()
104useSocketM sk f = finally (liftIO (Net.close sk)) (f sk)
106-- | Like 'useSocketM' but runs a streaming computation instead of a monadic
110{-# INLINE useSocket #-}
111useSocket :: (IsStream t, MonadCatch m, MonadIO m)
112 => Socket -> (Socket -> t m a) -> t m a
113useSocket sk = S.bracket (return sk) (liftIO . Net.close)
115-------------------------------------------------------------------------------
117-------------------------------------------------------------------------------
119{-# INLINABLE readArrayUptoWith #-}
121 :: (h -> Ptr Word8 -> Int -> IO Int)
125readArrayUptoWith f size h = do
126 ptr <- mallocPlainForeignPtrBytes size
127 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
128 withForeignPtr ptr $ \p -> do
132 , aEnd = p `plusPtr` n
133 , aBound = p `plusPtr` size
135 -- XXX shrink only if the diff is significant
139-- | Read a 'ByteArray' from a file handle. If no data is available on the
140-- handle it blocks until some data becomes available. If data is available
141-- then it immediately returns that data without blocking. It reads a maximum
142-- of up to the size requested.
143{-# INLINABLE readArrayOf #-}
144readArrayOf :: Int -> Socket -> IO (Array Word8)
145readArrayOf = readArrayUptoWith recvBuf
147-------------------------------------------------------------------------------
149-------------------------------------------------------------------------------
151waitWhen0 :: Int -> Socket -> IO ()
152waitWhen0 0 s = when rtsSupportsBoundThreads $
153#if MIN_VERSION_network(3,1,0)
154 withFdSocket s $ \fd -> threadWaitWrite $ fromIntegral fd
156 let fd = fdSocket s in threadWaitWrite $ fromIntegral fd
158waitWhen0 _ _ = return ()
160sendAll :: Socket -> Ptr Word8 -> Int -> IO ()
161sendAll _ _ len | len <= 0 = return ()
163 sent <- sendBuf s p len
165 -- assert (sent <= len)
166 when (sent >= 0) $ sendAll s (p `plusPtr` sent) (len - sent)
168{-# INLINABLE writeArrayWith #-}
169writeArrayWith :: Storable a
170 => (h -> Ptr Word8 -> Int -> IO ())
174writeArrayWith _ _ arr | A.length arr == 0 = return ()
175writeArrayWith f h Array{..} = withForeignPtr aStart $ \p ->
179 let p = unsafeForeignPtrToPtr aStart
182-- | Write an Array to a file handle.
185{-# INLINABLE writeArray #-}
186writeArray :: Storable a => Socket -> Array a -> IO ()
187writeArray = writeArrayWith sendAll
189-------------------------------------------------------------------------------
190-- Stream of Arrays IO
191-------------------------------------------------------------------------------
193{-# INLINABLE readArraysUptoWith #-}
194readArraysUptoWith :: (IsStream t, MonadIO m)
195 => (Int -> h -> IO (Array Word8))
196 -> Int -> h -> t m (Array Word8)
197readArraysUptoWith f size h = go
199 -- XXX use cons/nil instead
200 go = mkStream $ \_ yld _ stp -> do
201 arr <- liftIO $ f size h
206-- | @toStreamArraysOf size h@ reads a stream of arrays from file handle @h@.
207-- The maximum size of a single array is limited to @size@.
208-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'
210{-# INLINABLE toStreamArraysOf #-}
211toStreamArraysOf :: (IsStream t, MonadIO m)
212 => Int -> Socket -> t m (Array Word8)
213toStreamArraysOf = readArraysUptoWith readArrayOf
215-- XXX read 'Array a' instead of Word8
217-- | @toStreamArrays h@ reads a stream of arrays from socket handle @h@.
218-- The maximum size of a single array is limited to @defaultChunkSize@.
221{-# INLINE toStreamArrays #-}
222toStreamArrays :: (IsStream t, MonadIO m) => Socket -> t m (Array Word8)
223toStreamArrays = toStreamArraysOf A.defaultChunkSize
225-- | Unfold the tuple @(size, socket)@ into a stream of 'Word8' arrays. The
226-- stream consists of arrays representing chunks of data read from the socket.
227-- The maximum size of a single array is limited to @size@.
230{-# INLINE_NORMAL readArraysOf #-}
231readArraysOf :: MonadIO m => Unfold m (Int, Socket) (Array Word8)
232readArraysOf = Unfold step return
234 {-# INLINE_LATE step #-}
236 arr <- liftIO $ readArrayOf size h
240 _ -> D.Yield arr (size, h)
242-- | Unfolds a socket into a stream of 'Word8' arrays. The maximum size of a
243-- single array is limited to @defaultChunkSize@.
246{-# INLINE readArrays #-}
247readArrays :: MonadIO m => Unfold m Socket (Array Word8)
248readArrays = UF.supplyFirst readArraysOf A.defaultChunkSize
250-------------------------------------------------------------------------------
251-- Read File to Stream
252-------------------------------------------------------------------------------
254-- TODO for concurrent streams implement readahead IO. We can send multiple
255-- read requests at the same time. For serial case we can use async IO. We can
256-- also control the read throughput in mbps or IOPS.
259-- | @readInChunksOf chunkSize handle@ reads a byte stream from a file
260-- handle, reads are performed in chunks of up to @chunkSize@. The stream ends
261-- as soon as EOF is encountered.
263{-# INLINE readInChunksOf #-}
264readInChunksOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
265readInChunksOf chunkSize h = A.flattenArrays $ readArraysUpto chunkSize h
269-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
271-- > read = 'readByChunks' A.defaultChunkSize
272-- | Generate a stream of elements of the given type from a socket. The
273-- stream ends when EOF is encountered.
276{-# INLINE toStream #-}
277toStream :: (IsStream t, MonadIO m) => Socket -> t m Word8
278toStream = AS.concat . toStreamArrays
280-- | @readInChunksOf@ unfolds @(bufsize, socket)@ into a byte stream, reads
281-- are performed in buffers of up to @bufsize@.
284{-# INLINE readInChunksOf #-}
285readInChunksOf :: MonadIO m => Unfold m (Int, Socket) Word8
286readInChunksOf = UF.concat readArraysOf A.read
288-- | Unfolds a 'Socket' into a byte stream.
292read :: MonadIO m => Unfold m Socket Word8
293read = UF.supplyFirst readInChunksOf A.defaultChunkSize
295-------------------------------------------------------------------------------
297-------------------------------------------------------------------------------
299-- | Write a stream of arrays to a handle.
302{-# INLINE writeArraysS #-}
303writeArraysS :: (MonadIO m, Storable a)
304 => Socket -> SerialT m (Array a) -> m ()
305writeArraysS h m = S.mapM_ (liftIO . writeArray h) m
307-- | Write a stream of arrays to a socket.
310{-# INLINE writeArrays #-}
311writeArrays :: (MonadIO m, Storable a) => Socket -> Fold m (Array a) ()
312writeArrays h = FL.drainBy (liftIO . writeArray h)
314-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
317-- Note that if you use a chunk size less than 8K (GHC's default buffer
318-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
319-- do not want buffering to occur at GHC level as well. Same thing applies to
322-- | Like 'write' but provides control over the write buffer. Output will
323-- be written to the IO device as soon as we collect the specified number of
327{-# INLINE writeInChunksOfS #-}
328writeInChunksOfS :: MonadIO m => Int -> Socket -> SerialT m Word8 -> m ()
329writeInChunksOfS n h m = writeArraysS h $ AS.arraysOf n m
331-- | Write a byte stream to a socket. Accumulates the input in chunks of
332-- specified number of bytes before writing.
335{-# INLINE writeInChunksOf #-}
336writeInChunksOf :: MonadIO m => Int -> Socket -> Fold m Word8 ()
337writeInChunksOf n h = FL.lchunksOf n (A.writeNUnsafe n) (writeArrays h)
339-- > write = 'writeInChunksOf' A.defaultChunkSize
341-- | Write a byte stream to a file handle. Combines the bytes in chunks of size
342-- up to 'A.defaultChunkSize' before writing. Note that the write behavior
343-- depends on the 'IOMode' and the current seek position of the handle.
346{-# INLINE _writeS #-}
347_writeS :: MonadIO m => Socket -> SerialT m Word8 -> m ()
348_writeS = writeInChunksOfS A.defaultChunkSize
350-- | Write a byte stream to a socket. Accumulates the input in chunks of
351-- up to 'A.defaultChunkSize' bytes before writing.
354-- write = 'writeInChunksOf' 'A.defaultChunkSize'
359write :: MonadIO m => Socket -> Fold m Word8 ()
360write = writeInChunksOf A.defaultChunkSize
364write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
365write = toHandleWith A.defaultChunkSize
368-------------------------------------------------------------------------------
369-- IO with encoding/decoding Unicode characters
370-------------------------------------------------------------------------------
374-- > readUtf8 = decodeUtf8 . read
376-- Read a UTF8 encoded stream of unicode characters from a file handle.
379{-# INLINE readUtf8 #-}
380readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
381readUtf8 = decodeUtf8 . read
384-- > writeUtf8 h s = write h $ encodeUtf8 s
386-- Encode a stream of unicode characters to UTF8 and write it to the given file
387-- handle. Default block buffering applies to the writes.
390{-# INLINE writeUtf8 #-}
391writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
392writeUtf8 h s = write h $ encodeUtf8 s
394-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
395-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
396-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
397-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
401{-# INLINE writeUtf8ByLines #-}
402writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
403writeUtf8ByLines = undefined
405-- | Read UTF-8 lines from a file handle and apply the specified fold to each
406-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
409{-# INLINE readLines #-}
410readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
411readLines h f = foldLines (readUtf8 h) f
413-------------------------------------------------------------------------------
414-- Framing on a sequence
415-------------------------------------------------------------------------------
417-- | Read a stream from a file handle and split it into frames delimited by
418-- the specified sequence of elements. The supplied fold is applied on each
422{-# INLINE readFrames #-}
423readFrames :: (IsStream t, MonadIO m, Storable a)
424 => Array a -> Handle -> Fold m a b -> t m b
425readFrames = undefined -- foldFrames . read
427-- | Write a stream to the given file handle buffering up to frames separated
428-- by the given sequence or up to a maximum of @defaultChunkSize@.
431{-# INLINE writeByFrames #-}
432writeByFrames :: (IsStream t, MonadIO m, Storable a)
433 => Array a -> Handle -> t m a -> m ()
434writeByFrames = undefined
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list