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