@@ -0,0 +1,248 @@
2+{-# LANGUAGE BangPatterns #-}
3+{-# LANGUAGE FlexibleContexts #-}
4+{-# LANGUAGE MagicHash #-}
5+{-# LANGUAGE RecordWildCards #-}
6+{-# LANGUAGE UnboxedTuples #-}
11+-- Module : Streamly.Internal.Network.Server
12+-- Copyright : (c) 2019 Composewell Technologies
15+-- Maintainer : streamly@composewell.com
16+-- Stability : experimental
19+-- Combinators to build network servers.
21+-- > import qualified Streamly.Internal.Network.Server as Server
24+module Streamly.Internal.Network.Server
26+ -- * Server Configuration
30+ -- ** Connected sockets
40+ , connectionsOnAllAddrs
41+ , connectionsOnLocalHost
46+ -- These abstractions can be applied to any setting where we need to do a
47+ -- sink processing of multiple streams e.g. output from multiple processes
48+ -- or data coming from multiple files.
50+ -- handle connections concurrently using a specified fold
51+ -- , handleConnections
53+ -- handle frames concurrently using a specified fold
56+ -- merge frames from all connection into a single stream. Frames can be
57+ -- created by a specified fold.
67+import Control.Monad.IO.Class (MonadIO(..))
68+import Data.Word (Word8)
70+ (Socket, PortNumber, SocketOption(..), Family(..), SockAddr(..),
71+ withSocketsDo, SocketType(..), socket, accept, bind,
72+ defaultProtocol, setSocketOption, maxListenQueue, tupleToHostAddress)
73+import Prelude hiding (read)
75+import qualified Network.Socket as Net
77+import Streamly (MonadAsync)
78+import Streamly.Internal.Data.Unfold.Types (Unfold(..))
79+import Streamly.Streams.Serial (SerialT)
81+import qualified Streamly.Internal.Data.Unfold as UF
82+import qualified Streamly.Prelude as S
83+import qualified Streamly.Streams.StreamD.Type as D
85+-------------------------------------------------------------------------------
87+-------------------------------------------------------------------------------
89+-- | Specify the configuration of a server.
92+ serverAddressFamily :: !Family
93+ , serverAddress :: !SockAddr
94+ , serverSockOpts :: ![(SocketOption, Int)]
97+initListener :: Int -> Server -> IO Socket
98+initListener tcpListenQ Server{..} =
100+ sock <- socket serverAddressFamily Stream defaultProtocol
101+ mapM_ (\(opt, val) -> setSocketOption sock opt val) serverSockOpts
102+ bind sock serverAddress
103+ Net.listen sock tcpListenQ
106+{-# INLINE listenTuples #-}
107+listenTuples :: MonadIO m => Unfold m (Int, Server) (Socket, SockAddr)
108+listenTuples = Unfold step inject
110+ inject (tcpListenQ, opts) = do
111+ listener <- liftIO $ initListener tcpListenQ opts
115+ r <- liftIO $ accept listener
116+ -- XXX error handling
117+ return $ D.Yield r listener
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).
125+{-# INLINE listen #-}
126+listen :: MonadIO m => Unfold m (Int, Server) Socket
127+listen = UF.map fst listenTuples
129+{-# INLINE listenTuplesOnAddr #-}
132+ => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) (Socket, SockAddr)
133+listenTuplesOnAddr = UF.lmap f listenTuples
135+ f (addr, port) = (maxListenQueue, Server
136+ { serverAddressFamily = AF_INET
137+ , serverAddress = SockAddrInet port (tupleToHostAddress addr)
138+ , serverSockOpts = [(NoDelay,1), (ReuseAddr,1)]
141+-- | 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
143+-- connections are accepted.
146+{-# INLINE listenOnAddr #-}
149+ => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
150+listenOnAddr = UF.map fst listenTuplesOnAddr
152+-- | 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
154+-- the specified port.
156+-- > listenOnPort = UF.first listenOnAddr (0,0,0,0)
159+{-# INLINE listenOnPort #-}
160+listenOnPort :: MonadIO m => Unfold m PortNumber Socket
161+listenOnPort = UF.first listenOnAddr (0,0,0,0)
163+-- | Like 'listenOnAddr' but binds on the localhost IPv4 address @127.0.0.1@.
164+-- The server can only be accessed from the local host, it cannot be accessed
165+-- from other hosts on the network.
167+-- > listenOnPortLocal = UF.first listenOnAddr (127,0,0,1)
170+{-# INLINE listenOnPortLocal #-}
171+listenOnPortLocal :: MonadIO m => Unfold m PortNumber Socket
172+listenOnPortLocal = UF.first listenOnAddr (127,0,0,1)
174+-------------------------------------------------------------------------------
176+-------------------------------------------------------------------------------
178+{-# INLINE recvConnectionTuplesWith #-}
179+recvConnectionTuplesWith :: MonadAsync m
180+ => Int -> Server -> SerialT m (Socket, SockAddr)
181+recvConnectionTuplesWith tcpListenQ opts = S.unfoldrM step Nothing
184+ listener <- liftIO $ initListener tcpListenQ opts
185+ r <- liftIO $ accept listener
186+ -- XXX error handling
187+ return $ Just (r, Just listener)
189+ step (Just listener) = do
190+ r <- liftIO $ accept listener
191+ -- XXX error handling
192+ return $ Just (r, Just listener)
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.
200+{-# INLINE connections #-}
201+connections :: MonadAsync m => Int -> Server -> SerialT m Socket
202+connections tcpListenQ opts = fmap fst $
203+ recvConnectionTuplesWith tcpListenQ opts
205+{-# INLINE connectionTuplesOnAddr #-}
206+connectionTuplesOnAddr
208+ => (Word8, Word8, Word8, Word8)
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)]
218+-- | Like 'connections' but binds on the specified IPv4 address of the machine
219+-- and listens for TCP connections on the specified port.
222+{-# INLINE connectionsOnAddr #-}
225+ => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Socket
226+connectionsOnAddr addr port = fmap fst $ connectionTuplesOnAddr addr port
228+-- | 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
230+-- the specified port.
232+-- > connectionsOnAllAddrs = connectionsOnAddr (0,0,0,0)
235+{-# INLINE connectionsOnAllAddrs #-}
236+connectionsOnAllAddrs :: MonadAsync m => PortNumber -> SerialT m Socket
237+connectionsOnAllAddrs = connectionsOnAddr (0,0,0,0)
239+-- | Like 'connections' but binds on the localhost IPv4 address @127.0.0.1@.
240+-- The server can only be accessed from the local host, it cannot be accessed
241+-- from other hosts on the network.
243+-- > connectionsOnLocalHost = connectionsOnAddr (127,0,0,1)
246+{-# INLINE connectionsOnLocalHost #-}
247+connectionsOnLocalHost :: MonadAsync m => PortNumber -> SerialT m Socket
248+connectionsOnLocalHost = connectionsOnAddr (127,0,0,1)