git.iko.soy/streamly
on 32eace2 help sign in
streamly/src/Streamly/Internal/Network/Inet/TCP.hs
457 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.Data.Array.Foreign.Type (Array(..), writeNUnsafe)
107import Streamly.Internal.Data.Fold.Type (Fold(..))
108import Streamly.Internal.Data.Stream.IsStream (MonadAsync)
109import Streamly.Internal.Data.SVar (fork)
110import Streamly.Internal.Data.Stream.Serial (SerialT)
111import Streamly.Internal.Data.Stream.StreamK.Type (IsStream)
112import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
113import Streamly.Internal.Data.Unfold.Type (Unfold(..))
114import Streamly.Internal.Network.Socket (SockSpec(..), accept, connections)
115import Streamly.Internal.System.IO (defaultChunkSize)
117import qualified Control.Monad.Catch as MC
118import qualified Network.Socket as Net
120import qualified Streamly.Internal.Data.Unfold as UF
121import qualified Streamly.Internal.Data.Array.Foreign as A
122import qualified Streamly.Internal.Data.Array.Stream.Foreign as AS
123import qualified Streamly.Internal.Data.Fold.Type as FL
124import qualified Streamly.Internal.Data.Stream.IsStream as S
125import qualified Streamly.Network.Socket as SK
126import qualified Streamly.Internal.Network.Socket as ISK
128-------------------------------------------------------------------------------
129-- Accept (unfolds)
130-------------------------------------------------------------------------------
132{-# INLINE acceptOnAddrWith #-}
133acceptOnAddrWith
134 :: MonadIO m
135 => [(SocketOption, Int)]
136 -> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
137acceptOnAddrWith opts = UF.lmap f accept
138 where
139 f (addr, port) =
140 (maxListenQueue
141 , SockSpec
142 { sockFamily = AF_INET
143 , sockType = Stream
144 , sockProto = defaultProtocol -- TCP
145 , sockOpts = opts
146 }
147 , SockAddrInet port (tupleToHostAddress addr)
148 )
150-- | Unfold a tuple @(ipAddr, port)@ into a stream of connected TCP sockets.
151-- @ipAddr@ is the local IP address and @port@ is the local port on which
152-- connections are accepted.
153--
154-- @since 0.7.0
155{-# INLINE acceptOnAddr #-}
156acceptOnAddr
157 :: MonadIO m
158 => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
159acceptOnAddr = acceptOnAddrWith []
161{-# INLINE acceptOnPortWith #-}
162acceptOnPortWith :: MonadIO m
163 => [(SocketOption, Int)]
164 -> Unfold m PortNumber Socket
165acceptOnPortWith opts = UF.supplyFirst (0,0,0,0) (acceptOnAddrWith opts)
167-- | Like 'acceptOnAddr' but binds on the IPv4 address @0.0.0.0@ i.e. on all
168-- IPv4 addresses/interfaces of the machine and listens for TCP connections on
169-- the specified port.
170--
171-- > acceptOnPort = UF.supplyFirst acceptOnAddr (0,0,0,0)
172--
173-- @since 0.7.0
174{-# INLINE acceptOnPort #-}
175acceptOnPort :: MonadIO m => Unfold m PortNumber Socket
176acceptOnPort = UF.supplyFirst (0,0,0,0) acceptOnAddr
178-- | Like 'acceptOnAddr' but binds on the localhost IPv4 address @127.0.0.1@.
179-- The server can only be accessed from the local host, it cannot be accessed
180-- from other hosts on the network.
181--
182-- > acceptOnPortLocal = UF.supplyFirst acceptOnAddr (127,0,0,1)
183--
184-- @since 0.7.0
185{-# INLINE acceptOnPortLocal #-}
186acceptOnPortLocal :: MonadIO m => Unfold m PortNumber Socket
187acceptOnPortLocal = UF.supplyFirst (127,0,0,1) acceptOnAddr
189-------------------------------------------------------------------------------
190-- Accept (streams)
191-------------------------------------------------------------------------------
193{-# INLINE connectionsOnAddrWith #-}
194connectionsOnAddrWith
195 :: MonadAsync m
196 => [(SocketOption, Int)]
197 -> (Word8, Word8, Word8, Word8)
198 -> PortNumber
199 -> SerialT m Socket
200connectionsOnAddrWith opts addr port =
201 connections maxListenQueue SockSpec
202 { sockFamily = AF_INET
203 , sockType = Stream
204 , sockProto = defaultProtocol
205 , sockOpts = opts
206 }
207 (SockAddrInet port (tupleToHostAddress addr))
209-- | Like 'connections' but binds on the specified IPv4 address of the machine
210-- and listens for TCP connections on the specified port.
211--
212-- /Pre-release/
213{-# INLINE connectionsOnAddr #-}
214connectionsOnAddr
215 :: MonadAsync m
216 => (Word8, Word8, Word8, Word8)
217 -> PortNumber
218 -> SerialT m Socket
219connectionsOnAddr = connectionsOnAddrWith []
221-- | Like 'connections' but binds on the IPv4 address @0.0.0.0@ i.e. on all
222-- IPv4 addresses/interfaces of the machine and listens for TCP connections on
223-- the specified port.
224--
225-- > connectionsOnPort = connectionsOnAddr (0,0,0,0)
226--
227-- /Pre-release/
228{-# INLINE connectionsOnPort #-}
229connectionsOnPort :: MonadAsync m => PortNumber -> SerialT m Socket
230connectionsOnPort = connectionsOnAddr (0,0,0,0)
232-- | Like 'connections' but binds on the localhost IPv4 address @127.0.0.1@.
233-- The server can only be accessed from the local host, it cannot be accessed
234-- from other hosts on the network.
235--
236-- > connectionsOnLocalHost = connectionsOnAddr (127,0,0,1)
237--
238-- /Pre-release/
239{-# INLINE connectionsOnLocalHost #-}
240connectionsOnLocalHost :: MonadAsync m => PortNumber -> SerialT m Socket
241connectionsOnLocalHost = connectionsOnAddr (127,0,0,1)
243-------------------------------------------------------------------------------
244-- TCP Clients
245-------------------------------------------------------------------------------
247-- | Connect to the specified IP address and port number. Returns a connected
248-- socket or throws an exception.
249--
250-- @since 0.7.0
251connect :: (Word8, Word8, Word8, Word8) -> PortNumber -> IO Socket
252connect addr port = do
253 sock <- socket AF_INET Stream defaultProtocol
254 Net.connect sock (SockAddrInet port (Net.tupleToHostAddress addr))
255 `onException` Net.close sock
256 return sock
258-- | Connect to a remote host using IP address and port and run the supplied
259-- action on the resulting socket. 'withConnectionM' makes sure that the
260-- socket is closed on normal termination or in case of an exception. If
261-- closing the socket raises an exception, then this exception will be raised
262-- by 'withConnectionM'.
263--
264-- /Pre-release/
265{-# INLINABLE withConnectionM #-}
266withConnectionM :: (MonadMask m, MonadIO m)
267 => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> m ()) -> m ()
268withConnectionM addr port =
269 bracket (liftIO $ connect addr port) (liftIO . Net.close)
271-------------------------------------------------------------------------------
272-- Connect (unfolds)
273-------------------------------------------------------------------------------
275-- | Transform an 'Unfold' from a 'Socket' to an unfold from a remote IP
276-- address and port. The resulting unfold opens a socket, uses it using the
277-- supplied unfold and then makes sure that the socket is closed on normal
278-- termination or in case of an exception. If closing the socket raises an
279-- exception, then this exception will be raised by 'usingConnection'.
280--
281-- /Pre-release/
282{-# INLINABLE usingConnection #-}
283usingConnection :: (MonadCatch m, MonadAsync m)
284 => Unfold m Socket a
285 -> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) a
286usingConnection =
287 UF.bracket (\(addr, port) -> liftIO $ connect addr port)
288 (liftIO . Net.close)
290-------------------------------------------------------------------------------
291-- Connect (streams)
292-------------------------------------------------------------------------------
294-- | @'withConnection' addr port act@ opens a connection to the specified IPv4
295-- host address and port and passes the resulting socket handle to the
296-- computation @act@. The handle will be closed on exit from 'withConnection',
297-- whether by normal termination or by raising an exception. If closing the
298-- handle raises an exception, then this exception will be raised by
299-- 'withConnection' rather than any exception raised by 'act'.
300--
301-- /Pre-release/
302{-# INLINABLE withConnection #-}
303withConnection :: (IsStream t, MonadCatch m, MonadAsync m)
304 => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> t m a) -> t m a
305withConnection addr port =
306 S.bracket (liftIO $ connect addr port) (liftIO . Net.close)
308-------------------------------------------------------------------------------
309-- Read Addr to Stream
310-------------------------------------------------------------------------------
312-- | Read a stream from the supplied IPv4 host address and port number.
313--
314-- @since 0.7.0
315{-# INLINE read #-}
316read :: (MonadCatch m, MonadAsync m)
317 => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Word8
318read = UF.many (usingConnection ISK.readChunks) A.read
320-- | Read a stream from the supplied IPv4 host address and port number.
321--
322-- @since 0.7.0
323{-# INLINE toBytes #-}
324toBytes :: (IsStream t, MonadCatch m, MonadAsync m)
325 => (Word8, Word8, Word8, Word8) -> PortNumber -> t m Word8
326toBytes addr port = AS.concat $ withConnection addr port ISK.toChunks
328-------------------------------------------------------------------------------
329-- Writing
330-------------------------------------------------------------------------------
332-- | Write a stream of arrays to the supplied IPv4 host address and port
333-- number.
334--
335-- @since 0.7.0
336{-# INLINE putChunks #-}
337putChunks
338 :: (MonadCatch m, MonadAsync m)
339 => (Word8, Word8, Word8, Word8)
340 -> PortNumber
341 -> SerialT m (Array Word8)
342 -> m ()
343putChunks addr port xs =
344 S.drain $ withConnection addr port (\sk -> S.fromEffect $ ISK.putChunks sk xs)
346-- | Write a stream of arrays to the supplied IPv4 host address and port
347-- number.
348--
349-- @since 0.7.0
350{-# INLINE writeChunks #-}
351writeChunks
352 :: (MonadAsync m, MonadCatch m)
353 => (Word8, Word8, Word8, Word8)
354 -> PortNumber
355 -> Fold m (Array Word8) ()
356writeChunks addr port = Fold step initial extract
357 where
358 initial = do
359 skt <- liftIO (connect addr port)
360 fld <- FL.initialize (SK.writeChunks skt) `MC.onException` liftIO (Net.close skt)
361 return $ FL.Partial (Tuple' fld skt)
362 step (Tuple' fld skt) x = do
363 r <- FL.runStep fld x `MC.onException` liftIO (Net.close skt)
364 return $ FL.Partial (Tuple' r skt)
365 extract (Tuple' (Fold _ initial1 extract1) skt) = do
366 liftIO $ Net.close skt
367 res <- initial1
368 case res of
369 FL.Partial fs -> extract1 fs
370 FL.Done fb -> return fb
372-- | Like 'write' but provides control over the write buffer. Output will
373-- be written to the IO device as soon as we collect the specified number of
374-- input elements.
375--
376-- @since 0.7.0
377{-# INLINE putBytesWithBufferOf #-}
378putBytesWithBufferOf
379 :: (MonadCatch m, MonadAsync m)
380 => Int
381 -> (Word8, Word8, Word8, Word8)
382 -> PortNumber
383 -> SerialT m Word8
384 -> m ()
385putBytesWithBufferOf n addr port m = putChunks addr port $ AS.arraysOf n m
387-- | Like 'write' but provides control over the write buffer. Output will
388-- be written to the IO device as soon as we collect the specified number of
389-- input elements.
390--
391-- @since 0.7.0
392{-# INLINE writeWithBufferOf #-}
393writeWithBufferOf
394 :: (MonadAsync m, MonadCatch m)
395 => Int
396 -> (Word8, Word8, Word8, Word8)
397 -> PortNumber
398 -> Fold m Word8 ()
399writeWithBufferOf n addr port =
400 FL.chunksOf n (writeNUnsafe n) (writeChunks addr port)
402-- | Write a stream to the supplied IPv4 host address and port number.
403--
404-- @since 0.7.0
405{-# INLINE putBytes #-}
406putBytes :: (MonadCatch m, MonadAsync m)
407 => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Word8 -> m ()
408putBytes = putBytesWithBufferOf defaultChunkSize
410-- | Write a stream to the supplied IPv4 host address and port number.
411--
412-- @since 0.7.0
413{-# INLINE write #-}
414write :: (MonadAsync m, MonadCatch m)
415 => (Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()
416write = writeWithBufferOf defaultChunkSize
418-------------------------------------------------------------------------------
419-- Transformations
420-------------------------------------------------------------------------------
422{-# INLINABLE withInputConnect #-}
423withInputConnect
424 :: (IsStream t, MonadCatch m, MonadAsync m)
425 => (Word8, Word8, Word8, Word8)
426 -> PortNumber
427 -> SerialT m Word8
428 -> (Socket -> t m a)
429 -> t m a
430withInputConnect addr port input f = S.bracket pre post handler
432 where
434 pre = do
435 sk <- liftIO $ connect addr port
436 tid <- fork (ISK.putBytes sk input)
437 return (sk, tid)
439 handler (sk, _) = f sk
441 -- XXX kill the thread immediately?
442 post (sk, _) = liftIO $ Net.close sk
444-- | Send an input stream to a remote host and produce the output stream from
445-- the host. The server host just acts as a transformation function on the
446-- input stream. Both sending and receiving happen asynchronously.
447--
448-- /Pre-release/
449--
450{-# INLINABLE processBytes #-}
451processBytes
452 :: (IsStream t, MonadAsync m, MonadCatch m)
453 => (Word8, Word8, Word8, Word8)
454 -> PortNumber
455 -> SerialT m Word8
456 -> t m Word8
457processBytes addr port input = withInputConnect addr port input ISK.toBytes