git.iko.soy/streamly
on 6bde698 help sign in
streamly/src/Streamly/Network/Server.hs
166 lines 5.2 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.Network.Server
12-- Copyright : (c) 2019 Composewell Technologies
13--
14-- License : BSD3
15-- Maintainer : harendra.kumar@gmail.com
16-- Stability : experimental
17-- Portability : GHC
18--
19-- Combinators to build network servers.
20--
21-- > import qualified Streamly.Network.Server as Server
22--
24module Streamly.Network.Server
25 (
26 -- * Server Configuration
27 Server(..)
29 -- * TCP Servers
30 -- ** Connected sockets
31 , connections
32 , connectionsOnAddr
33 , connectionsOnAllAddrs
34 , connectionsOnLocalHost
36 {-
37 -- ** Sink Servers
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
47 , handleFrames
49 -- merge frames from all connection into a single stream. Frames can be
50 -- created by a specified fold.
51 , mergeFrames
53 -- * UDP Servers
54 , datagrams
55 , datagramsOn
56 -}
57 )
58where
60import Control.Monad.IO.Class (MonadIO(..))
61import Data.Word (Word8)
62import Network.Socket
63 (Socket, PortNumber, SocketOption(..), Family(..), SockAddr(..),
64 withSocketsDo, SocketType(..), socket, accept, bind,
65 defaultProtocol, setSocketOption, maxListenQueue, tupleToHostAddress)
66import Prelude hiding (read)
68import qualified Network.Socket as Net
70import Streamly (MonadAsync)
71import Streamly.Streams.Serial (SerialT)
73import qualified Streamly.Prelude as S
75-------------------------------------------------------------------------------
76-- Listen
77-------------------------------------------------------------------------------
79-- | Specify the configuration of a server.
80data Server = Server
81 {
82 serverAddressFamily :: !Family
83 , serverAddress :: !SockAddr
84 , serverSockOpts :: ![(SocketOption, Int)]
85 }
87initListener :: Int -> Server -> IO Socket
88initListener tcpListenQ Server{..} =
89 withSocketsDo $ do
90 sock <- socket serverAddressFamily Stream defaultProtocol
91 mapM_ (\(opt, val) -> setSocketOption sock opt val) serverSockOpts
92 bind sock serverAddress
93 Net.listen sock tcpListenQ
94 return sock
96{-# INLINE recvConnectionTuplesWith #-}
97recvConnectionTuplesWith :: MonadAsync m
98 => Int -> Server -> SerialT m (Socket, SockAddr)
99recvConnectionTuplesWith tcpListenQ opts = S.unfoldrM step Nothing
100 where
101 step Nothing = do
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.
116--
117-- @since 0.7.0
118{-# INLINE connections #-}
119connections :: MonadAsync m => Int -> Server -> SerialT m Socket
120connections tcpListenQ opts = fmap fst $
121 recvConnectionTuplesWith tcpListenQ opts
123{-# INLINE connectionTuplesOnAddr #-}
124connectionTuplesOnAddr
125 :: MonadAsync m
126 => (Word8, Word8, Word8, Word8)
127 -> PortNumber
128 -> SerialT m (Socket, SockAddr)
129connectionTuplesOnAddr addr port =
130 recvConnectionTuplesWith maxListenQueue Server
131 { serverAddressFamily = AF_INET
132 , serverAddress = SockAddrInet port (tupleToHostAddress addr)
133 , serverSockOpts = [(NoDelay,1), (ReuseAddr,1)]
134 }
136-- | Like 'connections' but binds on the specified IPv4 address of the machine
137-- and listens for TCP connections on the specified port.
138--
139-- @since 0.7.0
140{-# INLINE connectionsOnAddr #-}
141connectionsOnAddr
142 :: MonadAsync m
143 => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Socket
144connectionsOnAddr 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.
149--
150-- > connectionsOnAllAddrs = connectionsOnAddr (0,0,0,0)
151--
152-- @since 0.7.0
153{-# INLINE connectionsOnAllAddrs #-}
154connectionsOnAllAddrs :: MonadAsync m => PortNumber -> SerialT m Socket
155connectionsOnAllAddrs = 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.
160--
161-- > connectionsOnLocalHost = connectionsOnAddr (127,0,0,1)
162--
163-- @since 0.7.0
164{-# INLINE connectionsOnLocalHost #-}
165connectionsOnLocalHost :: MonadAsync m => PortNumber -> SerialT m Socket
166connectionsOnLocalHost = connectionsOnAddr (127,0,0,1)