@@ -23,14 +23,10 @@
2323
2424module Streamly.Internal.Network.Server
2525 (
26− -- * Server Configuration
27− Server(..)
28−
2926 -- * TCP Servers
3027 -- ** Connected sockets
3128 -- *** Unfolds
32− , listen
33− , listenOnAddr
29+ listenOnAddr
3430 , listenOnPort
3531 , listenOnPortLocal
3632
@@ -68,75 +64,15 @@ import Control.Monad.IO.Class (MonadIO(..))
6864import Data.Word (Word8)
6965import Network.Socket
7066 (Socket, PortNumber, SocketOption(..), Family(..), SockAddr(..),
71− withSocketsDo, SocketType(..), socket, accept, bind,
72− defaultProtocol, setSocketOption, maxListenQueue, tupleToHostAddress)
67+ SocketType(..), defaultProtocol, maxListenQueue, tupleToHostAddress)
7368import Prelude hiding (read)
7469
75−import qualified Network.Socket as Net
76−
7770import Streamly (MonadAsync)
7871import Streamly.Internal.Data.Unfold.Types (Unfold(..))
72+import Streamly.Internal.Network.Socket (SockSpec(..), listen, connections)
7973import Streamly.Streams.Serial (SerialT)
8074
8175import 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− })
14076
14177-- | Unfold a tuple @(ipAddr, port)@ into a stream of connected TCP sockets.
14278-- @ipAddr@ is the local IP address and @port@ is the local port on which
@@ -147,7 +83,18 @@ listenTuplesOnAddr = UF.lmap f listenTuples
14783listenOnAddr
14884 :: MonadIO m
14985 => 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+ )
15198
15299-- | Like 'listenOnAddr' but binds on the IPv4 address @0.0.0.0@ i.e. on all
153100-- IPv4 addresses/interfaces of the machine and listens for TCP connections on
@@ -175,46 +122,6 @@ listenOnPortLocal = UF.supplyFirst listenOnAddr (127,0,0,1)
175122-- Listen
176123-------------------------------------------------------------------------------
177124
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−
218125-- | Like 'connections' but binds on the specified IPv4 address of the machine
219126-- and listens for TCP connections on the specified port.
220127--
@@ -222,8 +129,17 @@ connectionTuplesOnAddr addr port =
222129{-# INLINE connectionsOnAddr #-}
223130connectionsOnAddr
224131 :: MonadAsync m
225− => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Socket
226−connectionsOnAddr addr port = fmap fst $ connectionTuplesOnAddr addr port
132+ => (Word8, Word8, Word8, Word8)
133+ -> PortNumber
134+ -> SerialT m Socket
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))
227143
228144-- | Like 'connections' but binds on the IPv4 address @0.0.0.0@ i.e. on all
229145-- IPv4 addresses/interfaces of the machine and listens for TCP connections on