git.iko.soy/streamly
on f8d80cc help sign in
streamly/src/Streamly/Internal/Network/Inet/TCP.hs
456 lines 15 kB history raw
1#include "inline.hs"
3-- |
4-- Module : Streamly.Internal.Network.Inet.TCP
5-- Copyright : (c) 2019 Composewell Technologies
6--
7-- License : BSD3
8-- Maintainer : streamly@composewell.com
9-- Stability : experimental
10-- Portability : GHC
11--
12-- Combinators to build Inet/TCP clients and servers.
14module Streamly.Internal.Network.Inet.TCP
15 (
16 -- * TCP Servers
17 -- ** Unfolds
18 acceptOnAddr
19 , acceptOnAddrWith
20 , acceptOnPort
21 , acceptOnPortWith
22 , acceptOnPortLocal
24 -- ** Streams
25 , connectionsOnAddr
26 , connectionsOnAddrWith
27 , connectionsOnPort
28 , connectionsOnLocalHost
30 -- * TCP clients
31 -- | IP Address based operations.
32 , connect
33 , withConnectionM
35 -- ** Unfolds
36 , usingConnection
37 , read
39 -- ** Streams
40 , withConnection
41 -- *** Source
42 , toBytes
43 -- , readUtf8
44 -- , readLines
45 -- , readFrames
46 -- , readByChunks
48 -- -- * Array Read
49 -- , readArrayUpto
50 -- , readArrayOf
52 -- , readChunksUpto
53 -- , readChunksOf
54 -- , readChunks
56 -- *** Sink
57 , write
58 -- , writeUtf8
59 -- , writeUtf8ByLines
60 -- , writeByFrames
61 , writeWithBufferOf
62 , putBytes
63 , putBytesWithBufferOf
65 -- -- * Array Write
66 -- , writeArray
67 , writeChunks
68 , putChunks
70 -- ** Transformation
71 , processBytes
72 {-
73 -- ** Sink Servers
75 -- These abstractions can be applied to any setting where we need to do a
76 -- sink processing of multiple streams e.g. output from multiple processes
77 -- or data coming from multiple files.
79 -- handle connections concurrently using a specified fold
80 -- , handleConnections
82 -- handle frames concurrently using a specified fold
83 , handleFrames
85 -- merge frames from all connection into a single stream. Frames can be
86 -- created by a specified fold.
87 , mergeFrames
89 -- * UDP Servers
90 , datagrams
91 , datagramsOn
92 -}
93 )
94where
96import Control.Exception (onException)
97import Control.Monad.Catch (MonadCatch, MonadMask, bracket)
98import Control.Monad.IO.Class (MonadIO(..))
99import Data.Word (Word8)
100import Network.Socket
101 (Socket, PortNumber, SocketOption(..), Family(..), SockAddr(..),
102 SocketType(..), defaultProtocol, maxListenQueue, tupleToHostAddress,
103 socket)
104import Prelude hiding (read)
106import Streamly.Internal.Control.Concurrent (MonadAsync, fork)
107import Streamly.Internal.Data.Array.Foreign.Type (Array(..), writeNUnsafe)
108import Streamly.Internal.Data.Fold.Type (Fold(..))
109import Streamly.Internal.Data.Stream.IsStream.Type (IsStream)
110import Streamly.Internal.Data.Stream.Serial (SerialT)
111import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
112import Streamly.Internal.Data.Unfold.Type (Unfold(..))
113import Streamly.Internal.Network.Socket (SockSpec(..), accept, connections)
114import Streamly.Internal.System.IO (defaultChunkSize)
116import qualified Control.Monad.Catch as MC
117import qualified Network.Socket as Net
119import qualified Streamly.Internal.Data.Unfold as UF
120import qualified Streamly.Internal.Data.Array.Foreign as A
121import qualified Streamly.Internal.Data.Array.Stream.Foreign as AS
122import qualified Streamly.Internal.Data.Fold.Type as FL
123import qualified Streamly.Internal.Data.Stream.IsStream as S
124import qualified Streamly.Internal.Network.Socket as ISK
126-------------------------------------------------------------------------------
127-- Accept (unfolds)
128-------------------------------------------------------------------------------
130{-# INLINE acceptOnAddrWith #-}
131acceptOnAddrWith
132 :: MonadIO m
133 => [(SocketOption, Int)]
134 -> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
135acceptOnAddrWith opts = UF.lmap f accept
136 where
137 f (addr, port) =
138 (maxListenQueue
139 , SockSpec
140 { sockFamily = AF_INET
141 , sockType = Stream
142 , sockProto = defaultProtocol -- TCP
143 , sockOpts = opts
144 }
145 , SockAddrInet port (tupleToHostAddress addr)
146 )
148-- | Unfold a tuple @(ipAddr, port)@ into a stream of connected TCP sockets.
149-- @ipAddr@ is the local IP address and @port@ is the local port on which
150-- connections are accepted.
151--
152-- @since 0.7.0
153{-# INLINE acceptOnAddr #-}
154acceptOnAddr
155 :: MonadIO m
156 => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
157acceptOnAddr = acceptOnAddrWith []
159{-# INLINE acceptOnPortWith #-}
160acceptOnPortWith :: MonadIO m
161 => [(SocketOption, Int)]
162 -> Unfold m PortNumber Socket
163acceptOnPortWith opts = UF.supplyFirst (0,0,0,0) (acceptOnAddrWith opts)
165-- | Like 'acceptOnAddr' but binds on the IPv4 address @0.0.0.0@ i.e. on all
166-- IPv4 addresses/interfaces of the machine and listens for TCP connections on
167-- the specified port.
168--
169-- > acceptOnPort = UF.supplyFirst acceptOnAddr (0,0,0,0)
170--
171-- @since 0.7.0
172{-# INLINE acceptOnPort #-}
173acceptOnPort :: MonadIO m => Unfold m PortNumber Socket
174acceptOnPort = UF.supplyFirst (0,0,0,0) acceptOnAddr
176-- | Like 'acceptOnAddr' but binds on the localhost IPv4 address @127.0.0.1@.
177-- The server can only be accessed from the local host, it cannot be accessed
178-- from other hosts on the network.
179--
180-- > acceptOnPortLocal = UF.supplyFirst acceptOnAddr (127,0,0,1)
181--
182-- @since 0.7.0
183{-# INLINE acceptOnPortLocal #-}
184acceptOnPortLocal :: MonadIO m => Unfold m PortNumber Socket
185acceptOnPortLocal = UF.supplyFirst (127,0,0,1) acceptOnAddr
187-------------------------------------------------------------------------------
188-- Accept (streams)
189-------------------------------------------------------------------------------
191{-# INLINE connectionsOnAddrWith #-}
192connectionsOnAddrWith
193 :: MonadAsync m
194 => [(SocketOption, Int)]
195 -> (Word8, Word8, Word8, Word8)
196 -> PortNumber
197 -> SerialT m Socket
198connectionsOnAddrWith opts addr port =
199 connections maxListenQueue SockSpec
200 { sockFamily = AF_INET
201 , sockType = Stream
202 , sockProto = defaultProtocol
203 , sockOpts = opts
204 }
205 (SockAddrInet port (tupleToHostAddress addr))
207-- | Like 'connections' but binds on the specified IPv4 address of the machine
208-- and listens for TCP connections on the specified port.
209--
210-- /Pre-release/
211{-# INLINE connectionsOnAddr #-}
212connectionsOnAddr
213 :: MonadAsync m
214 => (Word8, Word8, Word8, Word8)
215 -> PortNumber
216 -> SerialT m Socket
217connectionsOnAddr = connectionsOnAddrWith []
219-- | Like 'connections' but binds on the IPv4 address @0.0.0.0@ i.e. on all
220-- IPv4 addresses/interfaces of the machine and listens for TCP connections on
221-- the specified port.
222--
223-- > connectionsOnPort = connectionsOnAddr (0,0,0,0)
224--
225-- /Pre-release/
226{-# INLINE connectionsOnPort #-}
227connectionsOnPort :: MonadAsync m => PortNumber -> SerialT m Socket
228connectionsOnPort = connectionsOnAddr (0,0,0,0)
230-- | Like 'connections' but binds on the localhost IPv4 address @127.0.0.1@.
231-- The server can only be accessed from the local host, it cannot be accessed
232-- from other hosts on the network.
233--
234-- > connectionsOnLocalHost = connectionsOnAddr (127,0,0,1)
235--
236-- /Pre-release/
237{-# INLINE connectionsOnLocalHost #-}
238connectionsOnLocalHost :: MonadAsync m => PortNumber -> SerialT m Socket
239connectionsOnLocalHost = connectionsOnAddr (127,0,0,1)
241-------------------------------------------------------------------------------
242-- TCP Clients
243-------------------------------------------------------------------------------
245-- | Connect to the specified IP address and port number. Returns a connected
246-- socket or throws an exception.
247--
248-- @since 0.7.0
249connect :: (Word8, Word8, Word8, Word8) -> PortNumber -> IO Socket
250connect addr port = do
251 sock <- socket AF_INET Stream defaultProtocol
252 Net.connect sock (SockAddrInet port (Net.tupleToHostAddress addr))
253 `onException` Net.close sock
254 return sock
256-- | Connect to a remote host using IP address and port and run the supplied
257-- action on the resulting socket. 'withConnectionM' makes sure that the
258-- socket is closed on normal termination or in case of an exception. If
259-- closing the socket raises an exception, then this exception will be raised
260-- by 'withConnectionM'.
261--
262-- /Pre-release/
263{-# INLINABLE withConnectionM #-}
264withConnectionM :: (MonadMask m, MonadIO m)
265 => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> m ()) -> m ()
266withConnectionM addr port =
267 bracket (liftIO $ connect addr port) (liftIO . Net.close)
269-------------------------------------------------------------------------------
270-- Connect (unfolds)
271-------------------------------------------------------------------------------
273-- | Transform an 'Unfold' from a 'Socket' to an unfold from a remote IP
274-- address and port. The resulting unfold opens a socket, uses it using the
275-- supplied unfold and then makes sure that the socket is closed on normal
276-- termination or in case of an exception. If closing the socket raises an
277-- exception, then this exception will be raised by 'usingConnection'.
278--
279-- /Pre-release/
280{-# INLINABLE usingConnection #-}
281usingConnection :: (MonadCatch m, MonadAsync m)
282 => Unfold m Socket a
283 -> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) a
284usingConnection =
285 UF.bracket (\(addr, port) -> liftIO $ connect addr port)
286 (liftIO . Net.close)
288-------------------------------------------------------------------------------
289-- Connect (streams)
290-------------------------------------------------------------------------------
292-- | @'withConnection' addr port act@ opens a connection to the specified IPv4
293-- host address and port and passes the resulting socket handle to the
294-- computation @act@. The handle will be closed on exit from 'withConnection',
295-- whether by normal termination or by raising an exception. If closing the
296-- handle raises an exception, then this exception will be raised by
297-- 'withConnection' rather than any exception raised by 'act'.
298--
299-- /Pre-release/
300{-# INLINABLE withConnection #-}
301withConnection :: (IsStream t, MonadCatch m, MonadAsync m)
302 => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> t m a) -> t m a
303withConnection addr port =
304 S.bracket (liftIO $ connect addr port) (liftIO . Net.close)
306-------------------------------------------------------------------------------
307-- Read Addr to Stream
308-------------------------------------------------------------------------------
310-- | Read a stream from the supplied IPv4 host address and port number.
311--
312-- @since 0.7.0
313{-# INLINE read #-}
314read :: (MonadCatch m, MonadAsync m)
315 => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Word8
316read = UF.many (usingConnection ISK.readChunks) A.read
318-- | Read a stream from the supplied IPv4 host address and port number.
319--
320-- @since 0.7.0
321{-# INLINE toBytes #-}
322toBytes :: (IsStream t, MonadCatch m, MonadAsync m)
323 => (Word8, Word8, Word8, Word8) -> PortNumber -> t m Word8
324toBytes addr port = AS.concat $ withConnection addr port ISK.toChunks
326-------------------------------------------------------------------------------
327-- Writing
328-------------------------------------------------------------------------------
330-- | Write a stream of arrays to the supplied IPv4 host address and port
331-- number.
332--
333-- @since 0.7.0
334{-# INLINE putChunks #-}
335putChunks
336 :: (MonadCatch m, MonadAsync m)
337 => (Word8, Word8, Word8, Word8)
338 -> PortNumber
339 -> SerialT m (Array Word8)
340 -> m ()
341putChunks addr port xs =
342 S.drain $ withConnection addr port (\sk -> S.fromEffect $ ISK.putChunks sk xs)
344-- | Write a stream of arrays to the supplied IPv4 host address and port
345-- number.
346--
347-- @since 0.7.0
348{-# INLINE writeChunks #-}
349writeChunks
350 :: (MonadAsync m, MonadCatch m)
351 => (Word8, Word8, Word8, Word8)
352 -> PortNumber
353 -> Fold m (Array Word8) ()
354writeChunks addr port = Fold step initial extract
355 where
356 initial = do
357 skt <- liftIO (connect addr port)
358 fld <- FL.initialize (ISK.writeChunks skt)
359 `MC.onException` liftIO (Net.close skt)
360 return $ FL.Partial (Tuple' fld skt)
361 step (Tuple' fld skt) x = do
362 r <- FL.snoc fld x `MC.onException` liftIO (Net.close skt)
363 return $ FL.Partial (Tuple' r skt)
364 extract (Tuple' (Fold _ initial1 extract1) skt) = do
365 liftIO $ Net.close skt
366 res <- initial1
367 case res of
368 FL.Partial fs -> extract1 fs
369 FL.Done fb -> return fb
371-- | Like 'write' but provides control over the write buffer. Output will
372-- be written to the IO device as soon as we collect the specified number of
373-- input elements.
374--
375-- @since 0.7.0
376{-# INLINE putBytesWithBufferOf #-}
377putBytesWithBufferOf
378 :: (MonadCatch m, MonadAsync m)
379 => Int
380 -> (Word8, Word8, Word8, Word8)
381 -> PortNumber
382 -> SerialT m Word8
383 -> m ()
384putBytesWithBufferOf n addr port m = putChunks addr port $ AS.arraysOf n m
386-- | Like 'write' but provides control over the write buffer. Output will
387-- be written to the IO device as soon as we collect the specified number of
388-- input elements.
389--
390-- @since 0.7.0
391{-# INLINE writeWithBufferOf #-}
392writeWithBufferOf
393 :: (MonadAsync m, MonadCatch m)
394 => Int
395 -> (Word8, Word8, Word8, Word8)
396 -> PortNumber
397 -> Fold m Word8 ()
398writeWithBufferOf n addr port =
399 FL.chunksOf n (writeNUnsafe n) (writeChunks addr port)
401-- | Write a stream to the supplied IPv4 host address and port number.
402--
403-- @since 0.7.0
404{-# INLINE putBytes #-}
405putBytes :: (MonadCatch m, MonadAsync m)
406 => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Word8 -> m ()
407putBytes = putBytesWithBufferOf defaultChunkSize
409-- | Write a stream to the supplied IPv4 host address and port number.
410--
411-- @since 0.7.0
412{-# INLINE write #-}
413write :: (MonadAsync m, MonadCatch m)
414 => (Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()
415write = writeWithBufferOf defaultChunkSize
417-------------------------------------------------------------------------------
418-- Transformations
419-------------------------------------------------------------------------------
421{-# INLINABLE withInputConnect #-}
422withInputConnect
423 :: (IsStream t, MonadCatch m, MonadAsync m)
424 => (Word8, Word8, Word8, Word8)
425 -> PortNumber
426 -> SerialT m Word8
427 -> (Socket -> t m a)
428 -> t m a
429withInputConnect addr port input f = S.bracket pre post handler
431 where
433 pre = do
434 sk <- liftIO $ connect addr port
435 tid <- fork (ISK.putBytes sk input)
436 return (sk, tid)
438 handler (sk, _) = f sk
440 -- XXX kill the thread immediately?
441 post (sk, _) = liftIO $ Net.close sk
443-- | Send an input stream to a remote host and produce the output stream from
444-- the host. The server host just acts as a transformation function on the
445-- input stream. Both sending and receiving happen asynchronously.
446--
447-- /Pre-release/
448--
449{-# INLINABLE processBytes #-}
450processBytes
451 :: (IsStream t, MonadAsync m, MonadCatch m)
452 => (Word8, Word8, Word8, Word8)
453 -> PortNumber
454 -> SerialT m Word8
455 -> t m Word8
456processBytes addr port input = withInputConnect addr port input ISK.toBytes