1e3259c · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

use unfolds for Network.Socket APIs

1e3259cHarendra Kumar7 years ago

examples/EchoServer.hs+3−2

compare with master

unifiedsplit

@@ -1,12 +1,13 @@

1-- A concurrent TCP server that echoes everything that it receives.

3import Streamly

4+import Streamly.Internal.Network.Socket (readArrays)

4import Streamly.Network.Socket

5import Streamly.Network.Socket

5import Streamly.Network.Server

6import Streamly.Network.Server

6import qualified Streamly.Prelude as S

7import qualified Streamly.Prelude as S

8main :: IO ()

9main :: IO ()

9main = S.drain

10main = S.drain

10− $ parallely $ S.mapM (flip withSocket echo)

11+ $ parallely $ S.mapM (flip withSocketM echo)

11 $ serially $ S.unfold listenOnPort 8090

12 $ serially $ S.unfold listenOnPort 8090

12− where echo sk = S.fold (writeArrays sk) $ readArrays sk

13+ where echo sk = S.fold (writeArrays sk) $ S.unfold readArrays sk

examples/FileSinkServer.hs+2−2

@@ -24,7 +24,7 @@ main = do

24 (\src -> S.fold (FH.write src)

25 $ encodeChar8Unchecked

26 $ S.concatUnfold A.read

27− $ S.concatMapWith parallel (flip NS.withSocketS recv)

27+ $ S.concatMapWith parallel (flip NS.withSocket recv)

28 $ S.unfold NS.listenOnPort 8090)

30 where

@@ -32,4 +32,4 @@ main = do

32 recv =

33 S.splitWithSuffix (== '\n') A.write

34 . decodeChar8

35− . NS.read

35+ . S.unfold NS.read

src/Streamly/Internal/Network/Socket.hs+69−22

@@ -19,8 +19,8 @@

19module Streamly.Internal.Network.Socket

20 (

21 -- * Use a socket

22− withSocket

22+ withSocketM

23− , withSocketS

23+ , withSocket

25 -- * Read from connection

26 , read

@@ -28,15 +28,19 @@ module Streamly.Internal.Network.Socket

28 -- , readLines

29 -- , readFrames

30 -- , readByChunks

31+ , toStream

32 -- -- * Array Read

33 -- -- * Array Read

33 -- , readArrayUpto

34 -- , readArrayUpto

34 -- , readArrayOf

35 -- , readArrayOf

36 -- , readArraysUpto

37 -- , readArraysUpto

37− -- , readArraysOf

38+ , readArraysOf

38 , readArrays

39 , readArrays

41+ , toStreamArraysOf

42+ , toStreamArrays

43+

40 -- * Write to connection

44 -- * Write to connection

41 , write

45 , write

42 -- , writeUtf8

46 -- , writeUtf8

@@ -72,6 +76,7 @@ import Prelude hiding (read)

73import qualified Network.Socket as Net

77import qualified Network.Socket as Net

79+import Streamly.Internal.Data.Unfold.Types (Unfold(..))

75import Streamly.Internal.Memory.Array.Types (Array(..))

80import Streamly.Internal.Memory.Array.Types (Array(..))

76import Streamly.Streams.Serial (SerialT)

81import Streamly.Streams.Serial (SerialT)

77import Streamly.Streams.StreamK.Type (IsStream, mkStream)

82import Streamly.Streams.StreamK.Type (IsStream, mkStream)

@@ -80,21 +85,23 @@ import Streamly.Data.Fold (Fold)

81import qualified Streamly.Data.Fold as FL

86import qualified Streamly.Data.Fold as FL

82import qualified Streamly.Internal.Data.Fold.Types as FL

87import qualified Streamly.Internal.Data.Fold.Types as FL

88+import qualified Streamly.Internal.Data.Unfold as UF

83import qualified Streamly.Memory.Array as A

89import qualified Streamly.Memory.Array as A

84import qualified Streamly.Internal.Memory.ArrayStream as AS

90import qualified Streamly.Internal.Memory.ArrayStream as AS

85import qualified Streamly.Internal.Memory.Array.Types as A

91import qualified Streamly.Internal.Memory.Array.Types as A

86import qualified Streamly.Prelude as S

92import qualified Streamly.Prelude as S

93+import qualified Streamly.Streams.StreamD.Type as D

88−-- | @'withSocket' socket act@ runs the monadic computation @act@ passing the

95+-- | @'withSocketM' socket act@ runs the monadic computation @act@ passing the

89−-- socket handle to it. The handle will be closed on exit from 'withSocket',

96+-- socket handle to it. The handle will be closed on exit from 'withSocketM',

90-- whether by normal termination or by raising an exception. If closing the

97-- whether by normal termination or by raising an exception. If closing the

91-- handle raises an exception, then this exception will be raised by

98-- handle raises an exception, then this exception will be raised by

92−-- 'withSocket' rather than any exception raised by 'act'.

99+-- 'withSocketM' rather than any exception raised by 'act'.

93--

100--

94-- @since 0.7.0

101-- @since 0.7.0

95−{-# INLINE withSocket #-}

102+{-# INLINE withSocketM #-}

96−withSocket :: (MonadCatch m, MonadIO m) => Socket -> (Socket -> m ()) -> m ()

103+withSocketM :: (MonadCatch m, MonadIO m) => Socket -> (Socket -> m ()) -> m ()

97−withSocket sk f = do

104+withSocketM sk f = do

98 f sk `onException` liftIO (Net.close sk)

105 f sk `onException` liftIO (Net.close sk)

99 liftIO (Net.close sk)

106 liftIO (Net.close sk)

100

107

@@ -105,14 +112,14 @@ withSocket sk f = do

105-- withSocketS sk = A.flattenArrays $

112-- withSocketS sk = A.flattenArrays $

106-- S.finally (liftIO (Net.close sk)) (readArrays sk)

113-- S.finally (liftIO (Net.close sk)) (readArrays sk)

107

114

108−-- | Like 'withSocket' but runs a streaming computation instead of a monadic

115+-- | Like 'withSocketM' but runs a streaming computation instead of a monadic

109-- computation.

116-- computation.

110--

117--

111-- @since 0.7.0

118-- @since 0.7.0

112−{-# INLINE withSocketS #-}

119+{-# INLINE withSocket #-}

113−withSocketS :: (IsStream t, MonadCatch m, MonadIO m)

120+withSocket :: (IsStream t, MonadCatch m, MonadIO m)

114 => Socket -> (Socket -> t m a) -> t m a

121 => Socket -> (Socket -> t m a) -> t m a

115−withSocketS sk = S.bracket (return sk) (liftIO . Net.close)

122+withSocket sk = S.bracket (return sk) (liftIO . Net.close)

116

123

117-------------------------------------------------------------------------------

124-------------------------------------------------------------------------------

118-- Array IO (Input)

125-- Array IO (Input)

@@ -205,24 +212,49 @@ readArraysUptoWith f size h = go

205 then stp

212 then stp

206 else yld arr go

213 else yld arr go

207

214

208−-- | @readArraysOf size h@ reads a stream of arrays from file handle @h@.

215+-- | @toStreamArraysOf size h@ reads a stream of arrays from file handle @h@.

209-- The maximum size of a single array is limited to @size@.

216-- The maximum size of a single array is limited to @size@.

210-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'

217-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'

211-- on the 'Handle'.

218-- on the 'Handle'.

212−{-# INLINABLE readArraysOf #-}

219+{-# INLINABLE toStreamArraysOf #-}

213−readArraysOf :: (IsStream t, MonadIO m)

220+toStreamArraysOf :: (IsStream t, MonadIO m)

214 => Int -> Socket -> t m (Array Word8)

221 => Int -> Socket -> t m (Array Word8)

215−readArraysOf = readArraysUptoWith readArrayOf

222+toStreamArraysOf = readArraysUptoWith readArrayOf

216

223

217-- XXX read 'Array a' instead of Word8

224-- XXX read 'Array a' instead of Word8

218--

225--

219−-- | @readArrays h@ reads a stream of arrays from socket handle @h@.

226+-- | @toStreamArrays h@ reads a stream of arrays from socket handle @h@.

220-- The maximum size of a single array is limited to @defaultChunkSize@.

227-- The maximum size of a single array is limited to @defaultChunkSize@.

221--

228--

222-- @since 0.7.0

229-- @since 0.7.0

230+{-# INLINE toStreamArrays #-}

231+toStreamArrays :: (IsStream t, MonadIO m) => Socket -> t m (Array Word8)

232+toStreamArrays = toStreamArraysOf A.defaultChunkSize

233+

234+-- | Unfold the tuple @(size, socket)@ into a stream of 'Word8' arrays. The

235+-- stream consists of arrays representing chunks of data read from the socket.

236+-- The maximum size of a single array is limited to @size@.

237+--

238+-- @since 0.7.0

239+{-# INLINE_NORMAL readArraysOf #-}

240+readArraysOf :: MonadIO m => Unfold m (Int, Socket) (Array Word8)

241+readArraysOf = Unfold step return

242+ where

243+ {-# INLINE_LATE step #-}

244+ step (size, h) = do

245+ arr <- liftIO $ readArrayOf size h

246+ return $

247+ case A.length arr of

248+ 0 -> D.Stop

249+ _ -> D.Yield arr (size, h)

250+

251+-- | Unfolds a socket into a stream of 'Word8' arrays. The maximum size of a

252+-- single array is limited to @defaultChunkSize@.

253+--

254+-- @since 0.7.0

223{-# INLINE readArrays #-}

255{-# INLINE readArrays #-}

224−readArrays :: (IsStream t, MonadIO m) => Socket -> t m (Array Word8)

256+readArrays :: MonadIO m => Unfold m Socket (Array Word8)

225−readArrays = readArraysOf A.defaultChunkSize

257+readArrays = UF.first readArraysOf A.defaultChunkSize

226

258

227-------------------------------------------------------------------------------

259-------------------------------------------------------------------------------

228-- Read File to Stream

260-- Read File to Stream

@@ -250,9 +282,24 @@ readInChunksOf chunkSize h = A.flattenArrays $ readArraysUpto chunkSize h

250-- stream ends when EOF is encountered.

282-- stream ends when EOF is encountered.

251--

283--

252-- @since 0.7.0

284-- @since 0.7.0

285+{-# INLINE toStream #-}

286+toStream :: (IsStream t, MonadIO m) => Socket -> t m Word8

287+toStream = AS.concat . toStreamArrays

288+

289+-- | @readInChunksOf@ unfolds @(bufsize, socket)@ into a byte stream, reads

290+-- are performed in buffers of up to @bufsize@.

291+--

292+-- @since 0.7.0

293+{-# INLINE readInChunksOf #-}

294+readInChunksOf :: MonadIO m => Unfold m (Int, Socket) Word8

295+readInChunksOf = UF.concat readArraysOf A.read

296+

297+-- | Unfolds a 'Socket' into a byte stream.

298+--

299+-- @since 0.7.0

253{-# INLINE read #-}

300{-# INLINE read #-}

254−read :: (IsStream t, MonadIO m) => Socket -> t m Word8

301+read :: MonadIO m => Unfold m Socket Word8

255−read = AS.concat . readArrays

302+read = UF.first readInChunksOf A.defaultChunkSize

256

303

257-------------------------------------------------------------------------------

304-------------------------------------------------------------------------------

258-- Writing

305-- Writing

src/Streamly/Network/Client.hs+2−1

@@ -75,6 +75,7 @@ import qualified Streamly.Internal.Memory.ArrayStream as AS

75import qualified Streamly.Internal.Data.Fold.Types as FL

76import qualified Streamly.Prelude as S

77import qualified Streamly.Network.Socket as SK

78+import qualified Streamly.Internal.Network.Socket as ISK

79-------------------------------------------------------------------------------

80-------------------------------------------------------------------------------

80-- Connect

81-- Connect

@@ -110,7 +111,7 @@ withConnection addr port =

110{-# INLINE read #-}

111{-# INLINE read #-}

111read :: (IsStream t, MonadCatch m, MonadIO m)

112read :: (IsStream t, MonadCatch m, MonadIO m)

112 => (Word8, Word8, Word8, Word8) -> PortNumber -> t m Word8

113 => (Word8, Word8, Word8, Word8) -> PortNumber -> t m Word8

113−read addr port = AS.concat $ withConnection addr port SK.readArrays

114+read addr port = AS.concat $ withConnection addr port ISK.toStreamArrays

114

115

115-------------------------------------------------------------------------------

116-------------------------------------------------------------------------------

116-- Writing

117-- Writing

src/Streamly/Network/Socket.hs+6−6

@@ -24,8 +24,8 @@

24module Streamly.Network.Socket

25 (

26 -- * Use a socket

27− withSocket

27+ withSocketM

28− , withSocketS

28+ , withSocket

30 -- * Read from connection

31 , read

@@ -39,18 +39,18 @@ module Streamly.Network.Socket

39 -- , readArrayOf

41 -- , readArraysUpto

42− -- , readArraysOf

42+ , readArraysOf

43− , readArrays

43+ -- , readArrays

45 -- * Write to connection

46 , write

47 -- , writeUtf8

48 -- , writeUtf8ByLines

49 -- , writeByFrames

50− , writeInChunksOf

50+ -- , writeInChunksOf

52 -- -- * Array Write

53− , writeArray

53+ -- , writeArray

54 , writeArrays

56 -- reading/writing datagrams