ae1ffbf · streamly · git.iko.soy

src/Streamly/Internal/Network/Inet/TCP.hs+4−5

compare with master

unifiedsplit

@@ -68,7 +68,7 @@ module Streamly.Internal.Network.Inet.TCP

68 , fromChunks

69

70 -- ** Transformation

71− , transformBytesWith

71+ , processBytes

72 {-

73 -- ** Sink Servers

74

@@ -446,12 +446,11 @@ withInputConnect addr port input f = S.bracket pre post handler

446--

447-- /Pre-release/

448--

449−{-# INLINABLE transformBytesWith #-}

449+{-# INLINABLE processBytes #-}

450−transformBytesWith

450+processBytes

451 :: (IsStream t, MonadAsync m, MonadCatch m)

452 => (Word8, Word8, Word8, Word8)

453 -> PortNumber

454 -> SerialT m Word8

455 -> t m Word8

456−transformBytesWith addr port input =

457− withInputConnect addr port input ISK.toBytes

456+processBytes addr port input = withInputConnect addr port input ISK.toBytes

src/Streamly/Internal/Network/Socket.hs+18−16

@@ -13,8 +13,8 @@ module Streamly.Internal.Network.Socket

13 (

14 SockSpec (..)

15 -- * Use a socket

16− , handleWithM

16+ , forSocketM

17− , handleWith

17+ , withSocket

18

19 -- * Accept connections

20 , accept

@@ -105,25 +105,27 @@ import qualified Streamly.Internal.Data.Array.Foreign.Type as A

105import qualified Streamly.Internal.Data.Stream.IsStream as S

106import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

107

108−-- | @'handleWithM' socket act@ runs the monadic computation @act@ passing the

108+-- | @'forSocketM' action socket@ runs the monadic computation @action@ passing

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

109+-- the socket handle to it. The handle will be closed on exit from

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

110+-- 'forSocketM', whether by normal termination or by raising an exception. If

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

111+-- closing the handle raises an exception, then this exception will be raised

112−-- 'handleWithM' rather than any exception raised by 'act'.

112+-- by 'forSocketM' rather than any exception raised by 'action'.

113--

114−-- @since 0.7.0

114+-- @since 0.8.0

115−{-# INLINE handleWithM #-}

115+{-# INLINE forSocketM #-}

116−handleWithM :: (MonadMask m, MonadIO m) => (Socket -> m ()) -> Socket -> m ()

116+forSocketM :: (MonadMask m, MonadIO m) => (Socket -> m ()) -> Socket -> m ()

117−handleWithM f sk = finally (f sk) (liftIO (Net.close sk))

117+forSocketM f sk = finally (f sk) (liftIO (Net.close sk))

118

119−-- | Like 'handleWithM' but runs a streaming computation instead of a monadic

119+-- | Like 'forSocketM' but runs a streaming computation instead of a monadic

120-- computation.

121--

122−-- @since 0.7.0

122+-- /Inhibits stream fusion/

123−{-# INLINE handleWith #-}

123+--

124+-- /Internal/

125+{-# INLINE withSocket #-}

124−handleWith :: (IsStream t, MonadAsync m, MonadCatch m)

126+withSocket :: (IsStream t, MonadAsync m, MonadCatch m)

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

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

126−handleWith sk f = S.finally (liftIO $ Net.close sk) (f sk)

128+withSocket sk f = S.finally (liftIO $ Net.close sk) (f sk)

127

129

128-------------------------------------------------------------------------------

130-------------------------------------------------------------------------------

129-- Accept (Unfolds)

131-- Accept (Unfolds)

src/Streamly/Network/Inet/TCP.hs+4

@@ -22,6 +22,10 @@ module Streamly.Network.Inet.TCP

22 -- * Connect to Servers

23 , connect

24

25+ -- XXX Expose this as a pipe when we have pipes.

26+ -- * Transformation

27+ -- , processBytes

28+

25 {-

29 {-

26 -- ** Sink Servers

30 -- ** Sink Servers

27

31

src/Streamly/Network/Socket.hs+9−7

@@ -2,7 +2,7 @@

2-- Module : Streamly.Network.Socket

4--

5−-- License : BSD3

5+-- License : BSD-3-Clause

6-- Maintainer : streamly@composewell.com

7-- Stability : experimental

8-- Portability : GHC

@@ -32,7 +32,6 @@

32--

33-- import Data.Function ((&))

34-- import Network.Socket

35−-- import Streamly.Internal.Network.Socket (handleWithM)

36-- import Streamly.Network.Socket (SockSpec(..))

35-- import Streamly.Network.Socket (SockSpec(..))

37--

36--

38-- import qualified Streamly.Prelude as Stream

37-- import qualified Streamly.Prelude as Stream

@@ -52,8 +51,8 @@

52--

51--

53-- server spec addr =

52-- server spec addr =

54-- Stream.unfold Socket.accept (maxListenQueue, spec, addr) -- ParallelT IO Socket

53-- Stream.unfold Socket.accept (maxListenQueue, spec, addr) -- ParallelT IO Socket

55−-- & Stream.mapM (handleWithM echo) -- ParallelT IO ()

54+-- & Stream.mapM (Socket.forSocketM echo) -- ParallelT IO ()

56−-- & fromParallel -- SerialT IO ()

55+-- & Stream.fromParallel -- SerialT IO ()

57-- & Stream.drain -- IO ()

56-- & Stream.drain -- IO ()

58--

57--

59-- echo sk =

58-- echo sk =

@@ -64,12 +63,12 @@

64-- = Programmer Notes

63-- = Programmer Notes

65--

64--

66-- Read IO requests to connected stream sockets are performed in chunks of

65-- Read IO requests to connected stream sockets are performed in chunks of

67−-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'. Unless specified

66+-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'. Unless

68−-- otherwise in the API, writes are collected into chunks of

67+-- specified otherwise in the API, writes are collected into chunks of

69-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize' before they are

68-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize' before they are

70-- written to the socket. APIs are provided to control the chunking behavior.

69-- written to the socket. APIs are provided to control the chunking behavior.

71--

70--

72−-- > import qualified Streamly.Network.Socket as SK

71+-- > import qualified Streamly.Network.Socket as Socket

73--

72--

74-- = See Also

73-- = See Also

75--

74--

@@ -113,6 +112,9 @@ module Streamly.Network.Socket

113 , writeChunks

112 , writeChunks

114 , writeChunksWithBufferOf

113 , writeChunksWithBufferOf

115 , writeChunk

114 , writeChunk

115+

116+ -- * Exceptions

117+ , forSocketM

116 )

118 )

117where

119where

118

120

test/Streamly/Test/Network/Inet/TCP.hs+2−2

@@ -72,7 +72,7 @@ server

72server listener port sem handler = do

73 putMVar sem ()

74 Stream.fromSerial (Stream.unfold listener port)

75− & (Stream.fromAsync . Stream.mapM (Socket.handleWithM handler))

75+ & (Stream.fromAsync . Stream.mapM (Socket.forSocketM handler))

76 & Stream.drain

77

78remoteAddr :: (Word8,Word8,Word8,Word8)

@@ -85,7 +85,7 @@ sender port sem = do

85 Stream.replicate 1000 testData -- SerialT IO String

86 & Stream.concatMap Stream.fromList -- SerialT IO Char

87 & Unicode.encodeLatin1 -- SerialT IO Word8

88− & TCP.transformBytesWith remoteAddr port -- SerialT IO Word8

88+ & TCP.processBytes remoteAddr port -- SerialT IO Word8

89 & Unicode.decodeLatin1 -- SerialT IO Char

90

91execute

test/Streamly/Test/Network/Socket.hs+2−2

@@ -83,7 +83,7 @@ server :: PortNumber -> MVar () -> (Socket -> IO ()) -> IO ()

83server port sem handler = do

84 putMVar sem ()

85 Stream.fromSerial (Stream.unfold TCP.acceptOnPort port)

86− & Stream.fromAsync . Stream.mapM (Socket.handleWithM handler)

86+ & Stream.fromAsync . Stream.mapM (Socket.forSocketM handler)

87 & Stream.drain

88

89remoteAddr :: (Word8,Word8,Word8,Word8)

@@ -96,7 +96,7 @@ sender port sem = do

96 Stream.replicate 1000 testData -- SerialT IO String

97 & Stream.concatMap Stream.fromList -- SerialT IO Char

98 & Unicode.encodeLatin1 -- SerialT IO Word8

99− & TCP.transformBytesWith remoteAddr port -- SerialT IO Word8

99+ & TCP.processBytes remoteAddr port -- SerialT IO Word8

100 & Unicode.decodeLatin1 -- SerialT IO Char

101

102execute :: PortNumber -> Int -> (Socket -> IO ()) -> IO (SerialT IO Char)