@@ -66,10 +66,11 @@ import Prelude hiding (read)
6666import qualified Network.Socket as Net
6767
6868import Streamly (MonadAsync)
69−import Streamly.Mem.Array.Types (Array(..))
69+import Streamly.Mem.Array.Types (Array(..), defaultChunkSize)
7070import Streamly.Streams.Serial (SerialT)
7171import Streamly.Streams.StreamK.Type (IsStream)
7272
73+import qualified Streamly.Mem.Array as A
7374import qualified Streamly.Prelude as S
7475import qualified Streamly.Network.Socket as SK
7576
@@ -91,6 +92,7 @@ openConnection addr port = do
9192-- 'withConnection' rather than any exception raised by 'act'.
9293--
9394-- @since 0.7.0
95+{-# INLINABLE withConnection #-}
9496withConnection :: (IsStream t, MonadCatch m, MonadIO m)
9597 => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> t m a) -> t m a
9698withConnection addr port =
@@ -103,26 +105,20 @@ withConnection addr port =
103105-- | Read a stream from the supplied IPv4 host address and port number.
104106--
105107-- @since 0.7.0
108+{-# INLINE read #-}
106109read :: (IsStream t, MonadCatch m, MonadIO m)
107110 => (Word8, Word8, Word8, Word8) -> PortNumber -> t m Word8
108−read addr port = withConnection addr port SK.read
111+read addr port = A.flattenArrays $ withConnection addr port SK.readArrays
109112
110113-------------------------------------------------------------------------------
111114-- Writing
112115-------------------------------------------------------------------------------
113116
114−-- | Write a stream to the supplied IPv4 host address and port number.
115−--
116−-- @since 0.7.0
117−write :: (MonadCatch m, MonadIO m)
118− => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Word8 -> m ()
119−write addr port xs =
120− S.drain $ withConnection addr port (\sk -> S.yieldM $ SK.write sk xs)
121−
122117-- | Write a stream of arrays to the supplied IPv4 host address and port
123118-- number.
124119--
125120-- @since 0.7.0
121+{-# INLINE writeArrays #-}
126122writeArrays
127123 :: (MonadCatch m, MonadAsync m)
128124 => (Word8, Word8, Word8, Word8)
@@ -131,3 +127,26 @@ writeArrays
131127 -> m ()
132128writeArrays addr port xs =
133129 S.drain $ withConnection addr port (\sk -> S.yieldM $ SK.writeArrays sk xs)
130+
131+-- | Like 'write' but provides control over the write buffer. Output will
132+-- be written to the IO device as soon as we collect the specified number of
133+-- input elements.
134+--
135+-- @since 0.7.0
136+{-# INLINE writeByChunks #-}
137+writeByChunks
138+ :: (MonadCatch m, MonadAsync m)
139+ => Int
140+ -> (Word8, Word8, Word8, Word8)
141+ -> PortNumber
142+ -> SerialT m Word8
143+ -> m ()
144+writeByChunks n addr port m = writeArrays addr port $ A.arraysOf n m
145+
146+-- | Write a stream to the supplied IPv4 host address and port number.
147+--
148+-- @since 0.7.0
149+{-# INLINE write #-}
150+write :: (MonadCatch m, MonadAsync m)
151+ => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Word8 -> m ()
152+write = writeByChunks defaultChunkSize