@@ -0,0 +1,248 @@
1+{-# LANGUAGE CPP #-}
2+{-# LANGUAGE BangPatterns #-}
3+{-# LANGUAGE FlexibleContexts #-}
4+{-# LANGUAGE MagicHash #-}
5+{-# LANGUAGE RecordWildCards #-}
6+{-# LANGUAGE UnboxedTuples #-}
7+
8+#include "inline.hs"
9+
10+-- |
11+-- Module : Streamly.Internal.Network.Server
12+-- Copyright : (c) 2019 Composewell Technologies
13+--
14+-- License : BSD3
15+-- Maintainer : streamly@composewell.com
16+-- Stability : experimental
17+-- Portability : GHC
18+--
19+-- Combinators to build network servers.
20+--
21+-- > import qualified Streamly.Internal.Network.Server as Server
22+--
23+
24+module Streamly.Internal.Network.Server
25+ (
26+ -- * Server Configuration
27+ Server(..)
28+
29+ -- * TCP Servers
30+ -- ** Connected sockets
31+ -- *** Unfolds
32+ , listen
33+ , listenOnAddr
34+ , listenOnPort
35+ , listenOnPortLocal
36+
37+ -- *** Streams
38+ , connections
39+ , connectionsOnAddr
40+ , connectionsOnAllAddrs
41+ , connectionsOnLocalHost
42+
43+ {-
44+ -- ** Sink Servers
45+
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.
49+
50+ -- handle connections concurrently using a specified fold
51+ -- , handleConnections
52+
53+ -- handle frames concurrently using a specified fold
54+ , handleFrames
55+
56+ -- merge frames from all connection into a single stream. Frames can be
57+ -- created by a specified fold.
58+ , mergeFrames
59+
60+ -- * UDP Servers
61+ , datagrams
62+ , datagramsOn
63+ -}
64+ )
65+where
66+
67+import Control.Monad.IO.Class (MonadIO(..))
68+import Data.Word (Word8)
69+import Network.Socket
70+ (Socket, PortNumber, SocketOption(..), Family(..), SockAddr(..),
71+ withSocketsDo, SocketType(..), socket, accept, bind,
72+ defaultProtocol, setSocketOption, maxListenQueue, tupleToHostAddress)
73+import Prelude hiding (read)
74+
75+import qualified Network.Socket as Net
76+
77+import Streamly (MonadAsync)
78+import Streamly.Internal.Data.Unfold.Types (Unfold(..))
79+import Streamly.Streams.Serial (SerialT)
80+
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
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.
142+-- @ipAddr@ is the local IP address and @port@ is the local port on which
143+-- connections are accepted.
144+--
145+-- @since 0.7.0
146+{-# INLINE listenOnAddr #-}
147+listenOnAddr
148+ :: MonadIO m
149+ => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
150+listenOnAddr = UF.map fst listenTuplesOnAddr
151+
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.
155+--
156+-- > listenOnPort = UF.first listenOnAddr (0,0,0,0)
157+--
158+-- @since 0.7.0
159+{-# INLINE listenOnPort #-}
160+listenOnPort :: MonadIO m => Unfold m PortNumber Socket
161+listenOnPort = UF.first listenOnAddr (0,0,0,0)
162+
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.
166+--
167+-- > listenOnPortLocal = UF.first listenOnAddr (127,0,0,1)
168+--
169+-- @since 0.7.0
170+{-# INLINE listenOnPortLocal #-}
171+listenOnPortLocal :: MonadIO m => Unfold m PortNumber Socket
172+listenOnPortLocal = UF.first listenOnAddr (127,0,0,1)
173+
174+-------------------------------------------------------------------------------
175+-- Listen
176+-------------------------------------------------------------------------------
177+
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
219+-- and listens for TCP connections on the specified port.
220+--
221+-- /Internal/
222+{-# INLINE connectionsOnAddr #-}
223+connectionsOnAddr
224+ :: MonadAsync m
225+ => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Socket
226+connectionsOnAddr addr port = fmap fst $ connectionTuplesOnAddr addr port
227+
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.
231+--
232+-- > connectionsOnAllAddrs = connectionsOnAddr (0,0,0,0)
233+--
234+-- /Internal/
235+{-# INLINE connectionsOnAllAddrs #-}
236+connectionsOnAllAddrs :: MonadAsync m => PortNumber -> SerialT m Socket
237+connectionsOnAllAddrs = connectionsOnAddr (0,0,0,0)
238+
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.
242+--
243+-- > connectionsOnLocalHost = connectionsOnAddr (127,0,0,1)
244+--
245+-- /Internal/
246+{-# INLINE connectionsOnLocalHost #-}
247+connectionsOnLocalHost :: MonadAsync m => PortNumber -> SerialT m Socket
248+connectionsOnLocalHost = connectionsOnAddr (127,0,0,1)