48fc6bf · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

move generic socket listening APIs to socket module

48fc6bfHarendra Kumar7 years ago

src/Streamly/Internal/Network/Server.hs+26−110

compare with master

unifiedsplit

@@ -23,14 +23,10 @@

24module Streamly.Internal.Network.Server

25 (

26− -- * Server Configuration

27− Server(..)

28−

29 -- * TCP Servers

26 -- * TCP Servers

30 -- ** Connected sockets

27 -- ** Connected sockets

31 -- *** Unfolds

28 -- *** Unfolds

32− , listen

33− , listenOnAddr

29+ listenOnAddr

34 , listenOnPort

30 , listenOnPort

35 , listenOnPortLocal

31 , listenOnPortLocal

@@ -68,75 +64,15 @@ import Control.Monad.IO.Class (MonadIO(..))

68import Data.Word (Word8)

64import Data.Word (Word8)

69import Network.Socket

65import Network.Socket

70 (Socket, PortNumber, SocketOption(..), Family(..), SockAddr(..),

66 (Socket, PortNumber, SocketOption(..), Family(..), SockAddr(..),

71− withSocketsDo, SocketType(..), socket, accept, bind,

72− defaultProtocol, setSocketOption, maxListenQueue, tupleToHostAddress)

67+ SocketType(..), defaultProtocol, maxListenQueue, tupleToHostAddress)

73import Prelude hiding (read)

68import Prelude hiding (read)

75−import qualified Network.Socket as Net

76−

77import Streamly (MonadAsync)

70import Streamly (MonadAsync)

78import Streamly.Internal.Data.Unfold.Types (Unfold(..))

71import Streamly.Internal.Data.Unfold.Types (Unfold(..))

72+import Streamly.Internal.Network.Socket (SockSpec(..), listen, connections)

79import Streamly.Streams.Serial (SerialT)

73import Streamly.Streams.Serial (SerialT)

81import qualified Streamly.Internal.Data.Unfold as UF

75import qualified Streamly.Internal.Data.Unfold as UF

82−import qualified Streamly.Prelude as S

83−import qualified Streamly.Streams.StreamD.Type as D

84−

85−-------------------------------------------------------------------------------

86−-- Listen

87−-------------------------------------------------------------------------------

88−

89−-- | Specify the configuration of a server.

90−data Server = Server

91− {

92− serverAddressFamily :: !Family

93− , serverAddress :: !SockAddr

94− , serverSockOpts :: ![(SocketOption, Int)]

95− }

96−

97−initListener :: Int -> Server -> IO Socket

98−initListener tcpListenQ Server{..} =

99− withSocketsDo $ do

100− sock <- socket serverAddressFamily Stream defaultProtocol

101− mapM_ (\(opt, val) -> setSocketOption sock opt val) serverSockOpts

102− bind sock serverAddress

103− Net.listen sock tcpListenQ

104− return sock

105−

106−{-# INLINE listenTuples #-}

107−listenTuples :: MonadIO m => Unfold m (Int, Server) (Socket, SockAddr)

108−listenTuples = Unfold step inject

109− where

110− inject (tcpListenQ, opts) = do

111− listener <- liftIO $ initListener tcpListenQ opts

112− return listener

113−

114− step listener = do

115− r <- liftIO $ accept listener

116− -- XXX error handling

117− return $ D.Yield r listener

118−

119−-- | Unfold a tuple @(listenQueue, address)@ into a stream of connected

120−-- TCP sockets. @listenQueue@ is the maximum number of pending connections in

121−-- the backlog. @address@ is the server address specification (address

122−-- family, local interface IP address and port).

123−--

124−-- @since 0.7.0

125−{-# INLINE listen #-}

126−listen :: MonadIO m => Unfold m (Int, Server) Socket

127−listen = UF.map fst listenTuples

128−

129−{-# INLINE listenTuplesOnAddr #-}

130−listenTuplesOnAddr

131− :: MonadIO m

132− => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) (Socket, SockAddr)

133−listenTuplesOnAddr = UF.lmap f listenTuples

134− where

135− f (addr, port) = (maxListenQueue, Server

136− { serverAddressFamily = AF_INET

137− , serverAddress = SockAddrInet port (tupleToHostAddress addr)

138− , serverSockOpts = [(NoDelay,1), (ReuseAddr,1)]

139− })

140

141-- | Unfold a tuple @(ipAddr, port)@ into a stream of connected TCP sockets.

77-- | Unfold a tuple @(ipAddr, port)@ into a stream of connected TCP sockets.

142-- @ipAddr@ is the local IP address and @port@ is the local port on which

78-- @ipAddr@ is the local IP address and @port@ is the local port on which

@@ -147,7 +83,18 @@ listenTuplesOnAddr = UF.lmap f listenTuples

147listenOnAddr

83listenOnAddr

148 :: MonadIO m

84 :: MonadIO m

149 => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket

85 => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket

150−listenOnAddr = UF.map fst listenTuplesOnAddr

86+listenOnAddr = UF.lmap f listen

87+ where

88+ f (addr, port) =

89+ (maxListenQueue

90+ , SockSpec

91+ { sockFamily = AF_INET

92+ , sockType = Stream

93+ , sockProto = defaultProtocol -- TCP

94+ , sockOpts = [(NoDelay,1), (ReuseAddr,1)]

95+ }

96+ , SockAddrInet port (tupleToHostAddress addr)

97+ )

151

152-- | Like 'listenOnAddr' but binds on the IPv4 address @0.0.0.0@ i.e. on all

99-- | Like 'listenOnAddr' but binds on the IPv4 address @0.0.0.0@ i.e. on all

153-- IPv4 addresses/interfaces of the machine and listens for TCP connections on

100-- IPv4 addresses/interfaces of the machine and listens for TCP connections on

@@ -175,46 +122,6 @@ listenOnPortLocal = UF.supplyFirst listenOnAddr (127,0,0,1)

175-- Listen

122-- Listen

176-------------------------------------------------------------------------------

123-------------------------------------------------------------------------------

177

124

178−{-# INLINE recvConnectionTuplesWith #-}

179−recvConnectionTuplesWith :: MonadAsync m

180− => Int -> Server -> SerialT m (Socket, SockAddr)

181−recvConnectionTuplesWith tcpListenQ opts = S.unfoldrM step Nothing

182− where

183− step Nothing = do

184− listener <- liftIO $ initListener tcpListenQ opts

185− r <- liftIO $ accept listener

186− -- XXX error handling

187− return $ Just (r, Just listener)

188−

189− step (Just listener) = do

190− r <- liftIO $ accept listener

191− -- XXX error handling

192− return $ Just (r, Just listener)

193−

194−-- | Start a TCP stream server that listens for connections on the supplied

195−-- server address specification (address family, local interface IP address and

196−-- port). The server generates a stream of connected sockets. The first

197−-- argument is the maximum number of pending connections in the backlog.

198−--

199−-- /Internal/

200−{-# INLINE connections #-}

201−connections :: MonadAsync m => Int -> Server -> SerialT m Socket

202−connections tcpListenQ opts = fmap fst $

203− recvConnectionTuplesWith tcpListenQ opts

204−

205−{-# INLINE connectionTuplesOnAddr #-}

206−connectionTuplesOnAddr

207− :: MonadAsync m

208− => (Word8, Word8, Word8, Word8)

209− -> PortNumber

210− -> SerialT m (Socket, SockAddr)

211−connectionTuplesOnAddr addr port =

212− recvConnectionTuplesWith maxListenQueue Server

213− { serverAddressFamily = AF_INET

214− , serverAddress = SockAddrInet port (tupleToHostAddress addr)

215− , serverSockOpts = [(NoDelay,1), (ReuseAddr,1)]

216− }

217−

218-- | Like 'connections' but binds on the specified IPv4 address of the machine

125-- | Like 'connections' but binds on the specified IPv4 address of the machine

219-- and listens for TCP connections on the specified port.

126-- and listens for TCP connections on the specified port.

220--

127--

@@ -222,8 +129,17 @@ connectionTuplesOnAddr addr port =

222{-# INLINE connectionsOnAddr #-}

129{-# INLINE connectionsOnAddr #-}

223connectionsOnAddr

130connectionsOnAddr

224 :: MonadAsync m

131 :: MonadAsync m

225− => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Socket

132+ => (Word8, Word8, Word8, Word8)

133+ -> PortNumber

134+ -> SerialT m Socket

226−connectionsOnAddr addr port = fmap fst $ connectionTuplesOnAddr addr port

135+connectionsOnAddr addr port =

136+ connections maxListenQueue SockSpec

137+ { sockFamily = AF_INET

138+ , sockType = Stream

139+ , sockProto = defaultProtocol

140+ , sockOpts = [(NoDelay,1), (ReuseAddr,1)]

141+ }

142+ (SockAddrInet port (tupleToHostAddress addr))

227

143

228-- | Like 'connections' but binds on the IPv4 address @0.0.0.0@ i.e. on all

144-- | Like 'connections' but binds on the IPv4 address @0.0.0.0@ i.e. on all

229-- IPv4 addresses/interfaces of the machine and listens for TCP connections on

145-- IPv4 addresses/interfaces of the machine and listens for TCP connections on

src/Streamly/Internal/Network/Socket.hs+94−2

@@ -18,10 +18,15 @@

18--

19module Streamly.Internal.Network.Socket

20 (

21+ SockSpec (..)

21 -- * Use a socket

22 -- * Use a socket

22− useSocketM

23+ , useSocketM

23 , useSocket

24 , useSocket

26+ -- * Listen for incoming connections

27+ , listen

28+ , connections

29+

25 -- * Read from connection

30 -- * Read from connection

26 , read

31 , read

27 -- , readUtf8

32 -- , readUtf8

@@ -66,7 +71,12 @@ import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

66import Foreign.Ptr (minusPtr, plusPtr, Ptr, castPtr)

71import Foreign.Ptr (minusPtr, plusPtr, Ptr, castPtr)

67import Foreign.Storable (Storable(..))

72import Foreign.Storable (Storable(..))

68import GHC.ForeignPtr (mallocPlainForeignPtrBytes)

73import GHC.ForeignPtr (mallocPlainForeignPtrBytes)

69−import Network.Socket (Socket, sendBuf, recvBuf)

74+import Network.Socket (sendBuf, recvBuf)

75+import Network.Socket

76+ (Socket, PortNumber, SocketOption(..), Family(..), SockAddr(..),

77+ ProtocolNumber,

78+ withSocketsDo, SocketType(..), socket, accept, bind,

79+ defaultProtocol, setSocketOption, maxListenQueue, tupleToHostAddress)

70#if MIN_VERSION_network(3,1,0)

80#if MIN_VERSION_network(3,1,0)

71import Network.Socket (withFdSocket)

81import Network.Socket (withFdSocket)

72#else

82#else

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

77import qualified Network.Socket as Net

87import qualified Network.Socket as Net

89+import Streamly (MonadAsync)

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

90import Streamly.Internal.Data.Unfold.Types (Unfold(..))

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

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

81import Streamly.Streams.Serial (SerialT)

92import Streamly.Streams.Serial (SerialT)

@@ -112,6 +123,87 @@ useSocket :: (IsStream t, MonadCatch m, MonadIO m)

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

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

113useSocket sk f = S.finally (liftIO $ Net.close sk) (f sk)

124useSocket sk f = S.finally (liftIO $ Net.close sk) (f sk)

114

125

126+-------------------------------------------------------------------------------

127+-- Listen (Unfolds)

128+-------------------------------------------------------------------------------

129+

130+-- XXX Protocol specific socket options should be separated from socket level

131+-- options.

132+--

133+-- | Specify the socket protocol details.

134+data SockSpec = SockSpec

135+ {

136+ sockFamily :: !Family

137+ , sockType :: !SocketType

138+ , sockProto :: !ProtocolNumber

139+ , sockOpts :: ![(SocketOption, Int)]

140+ }

141+

142+initListener :: Int -> SockSpec -> SockAddr -> IO Socket

143+initListener listenQLen SockSpec{..} addr =

144+ withSocketsDo $ do

145+ sock <- socket sockFamily sockType sockProto

146+ mapM_ (\(opt, val) -> setSocketOption sock opt val) sockOpts

147+ bind sock addr

148+ Net.listen sock listenQLen

149+ return sock

150+

151+{-# INLINE listenTuples #-}

152+listenTuples :: MonadIO m

153+ => Unfold m (Int, SockSpec, SockAddr) (Socket, SockAddr)

154+listenTuples = Unfold step inject

155+ where

156+ inject (listenQLen, spec, addr) = do

157+ listener <- liftIO $ initListener listenQLen spec addr

158+ return listener

159+

160+ step listener = do

161+ r <- liftIO $ accept listener

162+ -- XXX error handling

163+ return $ D.Yield r listener

164+

165+-- | Unfold a three tuple @(listenQLen, spec, addr)@ into a stream of connected

166+-- protocol sockets. @listenQLen@ is the maximum number of pending connections

167+-- in the backlog. @spec@ is the socket protocol and options specification and

168+-- @addr@ is the protocol address where the server listens for incoming

169+-- connections.

170+--

171+-- @since 0.7.0

172+{-# INLINE listen #-}

173+listen :: MonadIO m => Unfold m (Int, SockSpec, SockAddr) Socket

174+listen = UF.map fst listenTuples

175+

176+-------------------------------------------------------------------------------

177+-- Listen (Streams)

178+-------------------------------------------------------------------------------

179+

180+{-# INLINE recvConnectionTuplesWith #-}

181+recvConnectionTuplesWith :: MonadAsync m

182+ => Int -> SockSpec -> SockAddr -> SerialT m (Socket, SockAddr)

183+recvConnectionTuplesWith tcpListenQ spec addr = S.unfoldrM step Nothing

184+ where

185+ step Nothing = do

186+ listener <- liftIO $ initListener tcpListenQ spec addr

187+ r <- liftIO $ accept listener

188+ -- XXX error handling

189+ return $ Just (r, Just listener)

190+

191+ step (Just listener) = do

192+ r <- liftIO $ accept listener

193+ -- XXX error handling

194+ return $ Just (r, Just listener)

195+

196+-- | Start a TCP stream server that listens for connections on the supplied

197+-- server address specification (address family, local interface IP address and

198+-- port). The server generates a stream of connected sockets. The first

199+-- argument is the maximum number of pending connections in the backlog.

200+--

201+-- /Internal/

202+{-# INLINE connections #-}

203+connections :: MonadAsync m => Int -> SockSpec -> SockAddr -> SerialT m Socket

204+connections tcpListenQ spec addr = fmap fst $

205+ recvConnectionTuplesWith tcpListenQ spec addr

206+

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

207-------------------------------------------------------------------------------

116-- Array IO (Input)

208-- Array IO (Input)

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

209-------------------------------------------------------------------------------

src/Streamly/Network/Server.hs+2−7

@@ -14,13 +14,8 @@

15module Streamly.Network.Server

16 (

17− -- * Server Configuration

17+ -- * Accept Connections

18− Server(..)

19−

20− -- * TCP Servers

21− -- ** Connected sockets

22− , listen

23− , listenOnAddr

18+ listenOnAddr

24 , listenOnPort

19 , listenOnPort

25 , listenOnPortLocal

20 , listenOnPortLocal

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

@@ -31,8 +31,14 @@

31--

32module Streamly.Network.Socket

33 (

34+ -- * Socket Specification

35+ SockSpec(..)

36+

37+ -- * Accept Connections

38+ , listen

39+

34 -- * Read from connection

40 -- * Read from connection

35− read

41+ , read

36 -- , readUtf8

42 -- , readUtf8

37 -- , readLines

43 -- , readLines

38 -- , readFrames

44 -- , readFrames