@@ -0,0 +1,166 @@
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.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+--
23+
24+module Streamly.Network.Server
25+ (
26+ -- * Server Configuration
27+ Server(..)
28+
29+ -- * TCP Servers
30+ -- ** Connected sockets
31+ , connections
32+ , connectionsOnAddr
33+ , connectionsOnAllAddrs
34+ , connectionsOnLocalHost
35+
36+ {-
37+ -- ** Sink Servers
38+
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.
42+
43+ -- handle connections concurrently using a specified fold
44+ -- , handleConnections
45+
46+ -- handle frames concurrently using a specified fold
47+ , handleFrames
48+
49+ -- merge frames from all connection into a single stream. Frames can be
50+ -- created by a specified fold.
51+ , mergeFrames
52+
53+ -- * UDP Servers
54+ , datagrams
55+ , datagramsOn
56+ -}
57+ )
58+where
59+
60+import Control.Monad.IO.Class (MonadIO(..))
61+import Data.Word (Word8)
62+import Network.Socket
63+ (Socket, PortNumber, SocketOption(..), Family(..), SockAddr(..),
64+ withSocketsDo, SocketType(..), socket, accept, bind,
65+ defaultProtocol, setSocketOption, maxListenQueue, tupleToHostAddress)
66+import Prelude hiding (read)
67+
68+import qualified Network.Socket as Net
69+
70+import Streamly (MonadAsync)
71+import Streamly.Streams.Serial (SerialT)
72+
73+import qualified Streamly.Prelude as S
74+
75+-------------------------------------------------------------------------------
76+-- Listen
77+-------------------------------------------------------------------------------
78+
79+-- | Specify the configuration of a server.
80+data Server = Server
81+ {
82+ serverAddressFamily :: !Family
83+ , serverAddress :: !SockAddr
84+ , serverSockOpts :: ![(SocketOption, Int)]
85+ }
86+
87+initListener :: Int -> Server -> IO Socket
88+initListener 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
95+
96+{-# INLINE recvConnectionTuplesWith #-}
97+recvConnectionTuplesWith :: MonadAsync m
98+ => Int -> Server -> SerialT m (Socket, SockAddr)
99+recvConnectionTuplesWith 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)
106+
107+ step (Just listener) = do
108+ r <- liftIO $ accept listener
109+ -- XXX error handling
110+ return $ Just (r, Just listener)
111+
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 #-}
119+connections :: MonadAsync m => Int -> Server -> SerialT m Socket
120+connections tcpListenQ opts = fmap fst $
121+ recvConnectionTuplesWith tcpListenQ opts
122+
123+{-# INLINE connectionTuplesOnAddr #-}
124+connectionTuplesOnAddr
125+ :: MonadAsync m
126+ => (Word8, Word8, Word8, Word8)
127+ -> PortNumber
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)]
134+ }
135+
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 #-}
141+connectionsOnAddr
142+ :: MonadAsync m
143+ => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Socket
144+connectionsOnAddr addr port = fmap fst $ connectionTuplesOnAddr addr port
145+
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 #-}
154+connectionsOnAllAddrs :: MonadAsync m => PortNumber -> SerialT m Socket
155+connectionsOnAllAddrs = connectionsOnAddr (0,0,0,0)
156+
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 #-}
165+connectionsOnLocalHost :: MonadAsync m => PortNumber -> SerialT m Socket
166+connectionsOnLocalHost = connectionsOnAddr (127,0,0,1)