@@ -0,0 +1,166 @@
2+{-# LANGUAGE BangPatterns #-}
3+{-# LANGUAGE FlexibleContexts #-}
4+{-# LANGUAGE MagicHash #-}
5+{-# LANGUAGE RecordWildCards #-}
6+{-# LANGUAGE UnboxedTuples #-}
11+-- Module : Streamly.Network.Server
12+-- Copyright : (c) 2019 Composewell Technologies
15+-- Maintainer : harendra.kumar@gmail.com
16+-- Stability : experimental
19+-- Combinators to build network servers.
21+-- > import qualified Streamly.Network.Server as Server
24+module Streamly.Network.Server
26+ -- * Server Configuration
30+ -- ** Connected sockets
33+ , connectionsOnAllAddrs
34+ , connectionsOnLocalHost
39+ -- These abstractions can be applied to any setting where we need to do a
40+ -- sink processing of multiple streams e.g. output from multiple processes
41+ -- or data coming from multiple files.
43+ -- handle connections concurrently using a specified fold
44+ -- , handleConnections
46+ -- handle frames concurrently using a specified fold
49+ -- merge frames from all connection into a single stream. Frames can be
50+ -- created by a specified fold.
60+import Control.Monad.IO.Class (MonadIO(..))
61+import Data.Word (Word8)
63+ (Socket, PortNumber, SocketOption(..), Family(..), SockAddr(..),
64+ withSocketsDo, SocketType(..), socket, accept, bind,
65+ defaultProtocol, setSocketOption, maxListenQueue, tupleToHostAddress)
66+import Prelude hiding (read)
68+import qualified Network.Socket as Net
70+import Streamly (MonadAsync)
71+import Streamly.Streams.Serial (SerialT)
73+import qualified Streamly.Prelude as S
75+-------------------------------------------------------------------------------
77+-------------------------------------------------------------------------------
79+-- | Specify the configuration of a server.
82+ serverAddressFamily :: !Family
83+ , serverAddress :: !SockAddr
84+ , serverSockOpts :: ![(SocketOption, Int)]
87+initListener :: Int -> Server -> IO Socket
88+initListener tcpListenQ Server{..} =
90+ sock <- socket serverAddressFamily Stream defaultProtocol
91+ mapM_ (\(opt, val) -> setSocketOption sock opt val) serverSockOpts
92+ bind sock serverAddress
93+ Net.listen sock tcpListenQ
96+{-# INLINE recvConnectionTuplesWith #-}
97+recvConnectionTuplesWith :: MonadAsync m
98+ => Int -> Server -> SerialT m (Socket, SockAddr)
99+recvConnectionTuplesWith tcpListenQ opts = S.unfoldrM step Nothing
102+ listener <- liftIO $ initListener tcpListenQ opts
103+ r <- liftIO $ accept listener
104+ -- XXX error handling
105+ return $ Just (r, Just listener)
107+ step (Just listener) = do
108+ r <- liftIO $ accept listener
109+ -- XXX error handling
110+ return $ Just (r, Just listener)
112+-- | Start a TCP stream server that listens for connections on the supplied
113+-- server address specification (address family, local interface IP address and
114+-- port). The server generates a stream of connected sockets. The first
115+-- argument is the maximum number of pending connections in the backlog.
118+{-# INLINE connections #-}
119+connections :: MonadAsync m => Int -> Server -> SerialT m Socket
120+connections tcpListenQ opts = fmap fst $
121+ recvConnectionTuplesWith tcpListenQ opts
123+{-# INLINE connectionTuplesOnAddr #-}
124+connectionTuplesOnAddr
126+ => (Word8, Word8, Word8, Word8)
128+ -> SerialT m (Socket, SockAddr)
129+connectionTuplesOnAddr addr port =
130+ recvConnectionTuplesWith maxListenQueue Server
131+ { serverAddressFamily = AF_INET
132+ , serverAddress = SockAddrInet port (tupleToHostAddress addr)
133+ , serverSockOpts = [(NoDelay,1), (ReuseAddr,1)]
136+-- | Like 'connections' but binds on the specified IPv4 address of the machine
137+-- and listens for TCP connections on the specified port.
140+{-# INLINE connectionsOnAddr #-}
143+ => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Socket
144+connectionsOnAddr addr port = fmap fst $ connectionTuplesOnAddr addr port
146+-- | Like 'connections' but binds on the IPv4 address @0.0.0.0@ i.e. on all
147+-- IPv4 addresses/interfaces of the machine and listens for TCP connections on
148+-- the specified port.
150+-- > connectionsOnAllAddrs = connectionsOnAddr (0,0,0,0)
153+{-# INLINE connectionsOnAllAddrs #-}
154+connectionsOnAllAddrs :: MonadAsync m => PortNumber -> SerialT m Socket
155+connectionsOnAllAddrs = connectionsOnAddr (0,0,0,0)
157+-- | Like 'connections' but binds on the localhost IPv4 address @127.0.0.1@.
158+-- The server can only be accessed from the local host, it cannot be accessed
159+-- from other hosts on the network.
161+-- > connectionsOnLocalHost = connectionsOnAddr (127,0,0,1)
164+{-# INLINE connectionsOnLocalHost #-}
165+connectionsOnLocalHost :: MonadAsync m => PortNumber -> SerialT m Socket
166+connectionsOnLocalHost = connectionsOnAddr (127,0,0,1)