git.iko.soy/streamly
on fa8a2cf help sign in
streamly/src/Streamly/Internal/Network/Inet/TCP.hs
453 lines 15 kB history raw
1{-# LANGUAGE CPP #-}
2{-# LANGUAGE FlexibleContexts #-}
4#include "inline.hs"
6-- |
7-- Module : Streamly.Internal.Network.Inet.TCP
8-- Copyright : (c) 2019 Composewell Technologies
9--
10-- License : BSD3
11-- Maintainer : streamly@composewell.com
12-- Stability : experimental
13-- Portability : GHC
14--
15-- Combinators to build Inet/TCP clients and servers.
17module Streamly.Internal.Network.Inet.TCP
18 (
19 -- * TCP Servers
20 -- ** Unfolds
21 acceptOnAddr
22 , acceptOnAddrWith
23 , acceptOnPort
24 , acceptOnPortWith
25 , acceptOnPortLocal
27 -- ** Streams
28 , connectionsOnAddr
29 , connectionsOnAddrWith
30 , connectionsOnPort
31 , connectionsOnLocalHost
33 -- * TCP clients
34 -- | IP Address based operations.
35 , connect
36 , withConnectionM
38 -- ** Unfolds
39 , usingConnection
40 , read
42 -- ** Streams
43 , withConnection
44 -- *** Source
45 , toBytes
46 -- , readUtf8
47 -- , readLines
48 -- , readFrames
49 -- , readByChunks
51 -- -- * Array Read
52 -- , readArrayUpto
53 -- , readArrayOf
55 -- , readChunksUpto
56 -- , readChunksOf
57 -- , readChunks
59 -- *** Sink
60 , write
61 -- , writeUtf8
62 -- , writeUtf8ByLines
63 -- , writeByFrames
64 , writeWithBufferOf
65 , fromBytes
66 , fromBytesWithBufferOf
68 -- -- * Array Write
69 -- , writeArray
70 , writeChunks
71 , fromChunks
73 -- ** Transformation
74 , transformBytesWith
75 {-
76 -- ** Sink Servers
78 -- These abstractions can be applied to any setting where we need to do a
79 -- sink processing of multiple streams e.g. output from multiple processes
80 -- or data coming from multiple files.
82 -- handle connections concurrently using a specified fold
83 -- , handleConnections
85 -- handle frames concurrently using a specified fold
86 , handleFrames
88 -- merge frames from all connection into a single stream. Frames can be
89 -- created by a specified fold.
90 , mergeFrames
92 -- * UDP Servers
93 , datagrams
94 , datagramsOn
95 -}
96 )
97where
99import Control.Monad.Catch (MonadCatch, MonadMask, bracket)
100import Control.Monad.IO.Class (MonadIO(..))
101import Data.Word (Word8)
102import Network.Socket
103 (Socket, PortNumber, SocketOption(..), Family(..), SockAddr(..),
104 SocketType(..), defaultProtocol, maxListenQueue, tupleToHostAddress,
105 socket)
106import Prelude hiding (read)
108import Streamly (MonadAsync)
109import Streamly.Internal.Data.Fold.Types (Fold(..))
110import Streamly.Internal.Data.SVar (fork)
111import Streamly.Internal.Data.Unfold.Types (Unfold(..))
112import Streamly.Internal.Network.Socket (SockSpec(..), accept, connections)
113import Streamly.Internal.Data.Stream.Serial (SerialT)
114import Streamly.Internal.Memory.Array.Types (Array(..), defaultChunkSize, writeNUnsafe)
115import Streamly.Internal.Data.Stream.StreamK.Type (IsStream)
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.Memory.Array as A
122import qualified Streamly.Internal.Memory.ArrayStream as AS
123import qualified Streamly.Internal.Data.Fold.Types as FL
124import qualified Streamly.Prelude 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 (acceptOnAddrWith opts) (0,0,0,0)
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 acceptOnAddr (0,0,0,0)
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 acceptOnAddr (127,0,0,1)
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-- /Internal/
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-- /Internal/
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-- /Internal/
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.
248--
249-- @since 0.7.0
250connect :: (Word8, Word8, Word8, Word8) -> PortNumber -> IO Socket
251connect addr port = do
252 sock <- socket AF_INET Stream defaultProtocol
253 Net.connect sock $ SockAddrInet port (Net.tupleToHostAddress addr)
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-- /Internal/
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-- /Internal/
280{-# INLINABLE usingConnection #-}
281usingConnection :: (MonadCatch m, MonadIO 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-- /Internal/
300{-# INLINABLE withConnection #-}
301withConnection :: (IsStream t, MonadCatch m, MonadIO 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, MonadIO m)
315 => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Word8
316read = UF.concat (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, MonadIO 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 fromChunks #-}
335fromChunks
336 :: (MonadCatch m, MonadAsync m)
337 => (Word8, Word8, Word8, Word8)
338 -> PortNumber
339 -> SerialT m (Array Word8)
340 -> m ()
341fromChunks addr port xs =
342 S.drain $ withConnection addr port (\sk -> S.yieldM $ ISK.fromChunks 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 (SK.writeChunks skt) `MC.onException` liftIO (Net.close skt)
359 return (fld, skt)
360 step (fld, skt) x = do
361 r <- FL.runStep fld x `MC.onException` liftIO (Net.close skt)
362 return (r, skt)
363 extract (Fold _ initial1 extract1, skt) = do
364 liftIO $ Net.close skt
365 initial1 >>= extract1
367-- | Like 'write' but provides control over the write buffer. Output will
368-- be written to the IO device as soon as we collect the specified number of
369-- input elements.
370--
371-- @since 0.7.0
372{-# INLINE fromBytesWithBufferOf #-}
373fromBytesWithBufferOf
374 :: (MonadCatch m, MonadAsync m)
375 => Int
376 -> (Word8, Word8, Word8, Word8)
377 -> PortNumber
378 -> SerialT m Word8
379 -> m ()
380fromBytesWithBufferOf n addr port m = fromChunks addr port $ AS.arraysOf n m
382-- | Like 'write' but provides control over the write buffer. Output will
383-- be written to the IO device as soon as we collect the specified number of
384-- input elements.
385--
386-- @since 0.7.0
387{-# INLINE writeWithBufferOf #-}
388writeWithBufferOf
389 :: (MonadAsync m, MonadCatch m)
390 => Int
391 -> (Word8, Word8, Word8, Word8)
392 -> PortNumber
393 -> Fold m Word8 ()
394writeWithBufferOf n addr port =
395 FL.lchunksOf n (writeNUnsafe n) (writeChunks addr port)
397-- | Write a stream to the supplied IPv4 host address and port number.
398--
399-- @since 0.7.0
400{-# INLINE fromBytes #-}
401fromBytes :: (MonadCatch m, MonadAsync m)
402 => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Word8 -> m ()
403fromBytes = fromBytesWithBufferOf defaultChunkSize
405-- | Write a stream to the supplied IPv4 host address and port number.
406--
407-- @since 0.7.0
408{-# INLINE write #-}
409write :: (MonadAsync m, MonadCatch m)
410 => (Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()
411write = writeWithBufferOf defaultChunkSize
413-------------------------------------------------------------------------------
414-- Transformations
415-------------------------------------------------------------------------------
417{-# INLINABLE withInputConnect #-}
418withInputConnect
419 :: (IsStream t, MonadCatch m, MonadAsync m)
420 => (Word8, Word8, Word8, Word8)
421 -> PortNumber
422 -> SerialT m Word8
423 -> (Socket -> t m a)
424 -> t m a
425withInputConnect addr port input f = S.bracket pre post handler
427 where
429 pre = do
430 sk <- liftIO $ connect addr port
431 tid <- fork (ISK.fromBytes sk input)
432 return (sk, tid)
434 handler (sk, _) = f sk
436 -- XXX kill the thread immediately?
437 post (sk, _) = liftIO $ Net.close sk
439-- | Send an input stream to a remote host and produce the output stream from
440-- the host. The server host just acts as a transformation function on the
441-- input stream. Both sending and receiving happen asynchronously.
442--
443-- /Internal/
444--
445{-# INLINABLE transformBytesWith #-}
446transformBytesWith
447 :: (IsStream t, MonadAsync m, MonadCatch m)
448 => (Word8, Word8, Word8, Word8)
449 -> PortNumber
450 -> SerialT m Word8
451 -> t m Word8
452transformBytesWith addr port input =
453 withInputConnect addr port input ISK.toBytes