git.iko.soy/streamly
on 3de89fe help sign in
streamly/src/Streamly/Internal/Network/Server.hs
248 lines 8.1 kB history raw
1{-# LANGUAGE CPP #-}
2{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE MagicHash #-}
5{-# LANGUAGE RecordWildCards #-}
6{-# LANGUAGE UnboxedTuples #-}
8#include "inline.hs"
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--
24module Streamly.Internal.Network.Server
25 (
26 -- * Server Configuration
27 Server(..)
29 -- * TCP Servers
30 -- ** Connected sockets
31 -- *** Unfolds
32 , listen
33 , listenOnAddr
34 , listenOnPort
35 , listenOnPortLocal
37 -- *** Streams
38 , connections
39 , connectionsOnAddr
40 , connectionsOnAllAddrs
41 , connectionsOnLocalHost
43 {-
44 -- ** Sink Servers
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
54 , handleFrames
56 -- merge frames from all connection into a single stream. Frames can be
57 -- created by a specified fold.
58 , mergeFrames
60 -- * UDP Servers
61 , datagrams
62 , datagramsOn
63 -}
64 )
65where
67import Control.Monad.IO.Class (MonadIO(..))
68import Data.Word (Word8)
69import Network.Socket
70 (Socket, PortNumber, SocketOption(..), Family(..), SockAddr(..),
71 withSocketsDo, SocketType(..), socket, accept, bind,
72 defaultProtocol, setSocketOption, maxListenQueue, tupleToHostAddress)
73import Prelude hiding (read)
75import qualified Network.Socket as Net
77import Streamly (MonadAsync)
78import Streamly.Internal.Data.Unfold.Types (Unfold(..))
79import Streamly.Streams.Serial (SerialT)
81import qualified Streamly.Internal.Data.Unfold as UF
82import qualified Streamly.Prelude as S
83import qualified Streamly.Streams.StreamD.Type as D
85-------------------------------------------------------------------------------
86-- Listen
87-------------------------------------------------------------------------------
89-- | Specify the configuration of a server.
90data Server = Server
91 {
92 serverAddressFamily :: !Family
93 , serverAddress :: !SockAddr
94 , serverSockOpts :: ![(SocketOption, Int)]
95 }
97initListener :: Int -> Server -> IO Socket
98initListener 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
106{-# INLINE listenTuples #-}
107listenTuples :: MonadIO m => Unfold m (Int, Server) (Socket, SockAddr)
108listenTuples = Unfold step inject
109 where
110 inject (tcpListenQ, opts) = do
111 listener <- liftIO $ initListener tcpListenQ opts
112 return listener
114 step listener = do
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).
123--
124-- @since 0.7.0
125{-# INLINE listen #-}
126listen :: MonadIO m => Unfold m (Int, Server) Socket
127listen = UF.map fst listenTuples
129{-# INLINE listenTuplesOnAddr #-}
130listenTuplesOnAddr
131 :: MonadIO m
132 => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) (Socket, SockAddr)
133listenTuplesOnAddr = 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 })
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 #-}
147listenOnAddr
148 :: MonadIO m
149 => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
150listenOnAddr = 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.
155--
156-- > listenOnPort = UF.first listenOnAddr (0,0,0,0)
157--
158-- @since 0.7.0
159{-# INLINE listenOnPort #-}
160listenOnPort :: MonadIO m => Unfold m PortNumber Socket
161listenOnPort = 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.
166--
167-- > listenOnPortLocal = UF.first listenOnAddr (127,0,0,1)
168--
169-- @since 0.7.0
170{-# INLINE listenOnPortLocal #-}
171listenOnPortLocal :: MonadIO m => Unfold m PortNumber Socket
172listenOnPortLocal = UF.first listenOnAddr (127,0,0,1)
174-------------------------------------------------------------------------------
175-- Listen
176-------------------------------------------------------------------------------
178{-# INLINE recvConnectionTuplesWith #-}
179recvConnectionTuplesWith :: MonadAsync m
180 => Int -> Server -> SerialT m (Socket, SockAddr)
181recvConnectionTuplesWith 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)
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.
198--
199-- /Internal/
200{-# INLINE connections #-}
201connections :: MonadAsync m => Int -> Server -> SerialT m Socket
202connections tcpListenQ opts = fmap fst $
203 recvConnectionTuplesWith tcpListenQ opts
205{-# INLINE connectionTuplesOnAddr #-}
206connectionTuplesOnAddr
207 :: MonadAsync m
208 => (Word8, Word8, Word8, Word8)
209 -> PortNumber
210 -> SerialT m (Socket, SockAddr)
211connectionTuplesOnAddr addr port =
212 recvConnectionTuplesWith maxListenQueue Server
213 { serverAddressFamily = AF_INET
214 , serverAddress = SockAddrInet port (tupleToHostAddress addr)
215 , serverSockOpts = [(NoDelay,1), (ReuseAddr,1)]
216 }
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 #-}
223connectionsOnAddr
224 :: MonadAsync m
225 => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Socket
226connectionsOnAddr 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.
231--
232-- > connectionsOnAllAddrs = connectionsOnAddr (0,0,0,0)
233--
234-- /Internal/
235{-# INLINE connectionsOnAllAddrs #-}
236connectionsOnAllAddrs :: MonadAsync m => PortNumber -> SerialT m Socket
237connectionsOnAllAddrs = 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.
242--
243-- > connectionsOnLocalHost = connectionsOnAddr (127,0,0,1)
244--
245-- /Internal/
246{-# INLINE connectionsOnLocalHost #-}
247connectionsOnLocalHost :: MonadAsync m => PortNumber -> SerialT m Socket
248connectionsOnLocalHost = connectionsOnAddr (127,0,0,1)