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