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