git.iko.soy/streamly
on ef9447d help sign in
streamly/src/Streamly/Internal/Network/Inet/TCP.hs
388 lines 13 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
74 {-
75 -- ** Sink Servers
77 -- These abstractions can be applied to any setting where we need to do a
78 -- sink processing of multiple streams e.g. output from multiple processes
79 -- or data coming from multiple files.
81 -- handle connections concurrently using a specified fold
82 -- , handleConnections
84 -- handle frames concurrently using a specified fold
85 , handleFrames
87 -- merge frames from all connection into a single stream. Frames can be
88 -- created by a specified fold.
89 , mergeFrames
91 -- * UDP Servers
92 , datagrams
93 , datagramsOn
94 -}
95 )
96where
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 (MonadAsync)
108import Streamly.Internal.Data.Fold.Types (Fold(..))
109import Streamly.Internal.Data.Unfold.Types (Unfold(..))
110import Streamly.Internal.Network.Socket (SockSpec(..), accept, connections)
111import Streamly.Streams.Serial (SerialT)
112import Streamly.Internal.Memory.Array.Types (Array(..), defaultChunkSize, writeNUnsafe)
113import Streamly.Streams.StreamK.Type (IsStream)
115import qualified Control.Monad.Catch as MC
116import qualified Network.Socket as Net
118import qualified Streamly.Internal.Data.Unfold as UF
119import qualified Streamly.Internal.Memory.Array as A
120import qualified Streamly.Internal.Memory.ArrayStream as AS
121import qualified Streamly.Internal.Data.Fold.Types as FL
122import qualified Streamly.Prelude as S
123import qualified Streamly.Network.Socket as SK
124import qualified Streamly.Internal.Network.Socket as ISK
126-------------------------------------------------------------------------------
127-- Accept (unfolds)
128-------------------------------------------------------------------------------
130-- | Unfold a tuple @(ipAddr, port)@ into a stream of connected TCP sockets.
131-- @ipAddr@ is the local IP address and @port@ is the local port on which
132-- connections are accepted.
133--
134-- @since 0.7.0
135{-# INLINE acceptOnAddr #-}
136acceptOnAddr
137 :: MonadIO m
138 => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
139acceptOnAddr = UF.lmap f accept
140 where
141 f (addr, port) =
142 (maxListenQueue
143 , SockSpec
144 { sockFamily = AF_INET
145 , sockType = Stream
146 , sockProto = defaultProtocol -- TCP
147 , sockOpts = [(NoDelay,1), (ReuseAddr,1)]
148 }
149 , SockAddrInet port (tupleToHostAddress addr)
150 )
152-- | Like 'acceptOnAddr' but binds on the IPv4 address @0.0.0.0@ i.e. on all
153-- IPv4 addresses/interfaces of the machine and listens for TCP connections on
154-- the specified port.
155--
156-- > acceptOnPort = UF.supplyFirst acceptOnAddr (0,0,0,0)
157--
158-- @since 0.7.0
159{-# INLINE acceptOnPort #-}
160acceptOnPort :: MonadIO m => Unfold m PortNumber Socket
161acceptOnPort = UF.supplyFirst acceptOnAddr (0,0,0,0)
163-- | Like 'acceptOnAddr' but binds on the localhost IPv4 address @127.0.0.1@.
164-- The server can only be accessed from the local host, it cannot be accessed
165-- from other hosts on the network.
166--
167-- > acceptOnPortLocal = UF.supplyFirst acceptOnAddr (127,0,0,1)
168--
169-- @since 0.7.0
170{-# INLINE acceptOnPortLocal #-}
171acceptOnPortLocal :: MonadIO m => Unfold m PortNumber Socket
172acceptOnPortLocal = UF.supplyFirst acceptOnAddr (127,0,0,1)
174-------------------------------------------------------------------------------
175-- Accept (streams)
176-------------------------------------------------------------------------------
178-- | Like 'connections' but binds on the specified IPv4 address of the machine
179-- and listens for TCP connections on the specified port.
180--
181-- /Internal/
182{-# INLINE connectionsOnAddr #-}
183connectionsOnAddr
184 :: MonadAsync m
185 => (Word8, Word8, Word8, Word8)
186 -> PortNumber
187 -> SerialT m Socket
188connectionsOnAddr addr port =
189 connections maxListenQueue SockSpec
190 { sockFamily = AF_INET
191 , sockType = Stream
192 , sockProto = defaultProtocol
193 , sockOpts = [(NoDelay,1), (ReuseAddr,1)]
194 }
195 (SockAddrInet port (tupleToHostAddress addr))
197-- | Like 'connections' but binds on the IPv4 address @0.0.0.0@ i.e. on all
198-- IPv4 addresses/interfaces of the machine and listens for TCP connections on
199-- the specified port.
200--
201-- > connectionsOnPort = connectionsOnAddr (0,0,0,0)
202--
203-- /Internal/
204{-# INLINE connectionsOnPort #-}
205connectionsOnPort :: MonadAsync m => PortNumber -> SerialT m Socket
206connectionsOnPort = connectionsOnAddr (0,0,0,0)
208-- | Like 'connections' but binds on the localhost IPv4 address @127.0.0.1@.
209-- The server can only be accessed from the local host, it cannot be accessed
210-- from other hosts on the network.
211--
212-- > connectionsOnLocalHost = connectionsOnAddr (127,0,0,1)
213--
214-- /Internal/
215{-# INLINE connectionsOnLocalHost #-}
216connectionsOnLocalHost :: MonadAsync m => PortNumber -> SerialT m Socket
217connectionsOnLocalHost = connectionsOnAddr (127,0,0,1)
219-------------------------------------------------------------------------------
220-- TCP Clients
221-------------------------------------------------------------------------------
223-- | Connect to the specified IP address and port number.
224--
225-- @since 0.7.0
226connect :: (Word8, Word8, Word8, Word8) -> PortNumber -> IO Socket
227connect addr port = do
228 sock <- socket AF_INET Stream defaultProtocol
229 Net.connect sock $ SockAddrInet port (Net.tupleToHostAddress addr)
230 return sock
232-- | Connect to a remote host using IP address and port and run the supplied
233-- action on the resulting socket. 'withConnectionM' makes sure that the
234-- socket is closed on normal termination or in case of an exception. If
235-- closing the socket raises an exception, then this exception will be raised
236-- by 'withConnectionM'.
237--
238-- /Internal/
239{-# INLINABLE withConnectionM #-}
240withConnectionM :: (MonadMask m, MonadIO m)
241 => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> m ()) -> m ()
242withConnectionM addr port =
243 bracket (liftIO $ connect addr port) (liftIO . Net.close)
245-------------------------------------------------------------------------------
246-- Connect (unfolds)
247-------------------------------------------------------------------------------
249-- | Transform an 'Unfold' from a 'Socket' to an unfold from a remote IP
250-- address and port. The resulting unfold opens a socket, uses it using the
251-- supplied unfold and then makes sure that the socket is closed on normal
252-- termination or in case of an exception. If closing the socket raises an
253-- exception, then this exception will be raised by 'usingConnection'.
254--
255-- /Internal/
256{-# INLINABLE usingConnection #-}
257usingConnection :: (MonadCatch m, MonadIO m)
258 => Unfold m Socket a
259 -> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) a
260usingConnection =
261 UF.bracket (\(addr, port) -> liftIO $ connect addr port)
262 (liftIO . Net.close)
264-------------------------------------------------------------------------------
265-- Connect (streams)
266-------------------------------------------------------------------------------
268-- | @'withConnection' addr port act@ opens a connection to the specified IPv4
269-- host address and port and passes the resulting socket handle to the
270-- computation @act@. The handle will be closed on exit from 'withConnection',
271-- whether by normal termination or by raising an exception. If closing the
272-- handle raises an exception, then this exception will be raised by
273-- 'withConnection' rather than any exception raised by 'act'.
274--
275-- /Internal/
276{-# INLINABLE withConnection #-}
277withConnection :: (IsStream t, MonadCatch m, MonadIO m)
278 => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> t m a) -> t m a
279withConnection addr port =
280 S.bracket (liftIO $ connect addr port) (liftIO . Net.close)
282-------------------------------------------------------------------------------
283-- Read Addr to Stream
284-------------------------------------------------------------------------------
286-- | Read a stream from the supplied IPv4 host address and port number.
287--
288-- @since 0.7.0
289{-# INLINE read #-}
290read :: (MonadCatch m, MonadIO m)
291 => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Word8
292read = UF.concat (usingConnection ISK.readChunks) A.read
294-- | Read a stream from the supplied IPv4 host address and port number.
295--
296-- @since 0.7.0
297{-# INLINE toBytes #-}
298toBytes :: (IsStream t, MonadCatch m, MonadIO m)
299 => (Word8, Word8, Word8, Word8) -> PortNumber -> t m Word8
300toBytes addr port = AS.concat $ withConnection addr port ISK.toChunks
302-------------------------------------------------------------------------------
303-- Writing
304-------------------------------------------------------------------------------
306-- | Write a stream of arrays to the supplied IPv4 host address and port
307-- number.
308--
309-- @since 0.7.0
310{-# INLINE fromChunks #-}
311fromChunks
312 :: (MonadCatch m, MonadAsync m)
313 => (Word8, Word8, Word8, Word8)
314 -> PortNumber
315 -> SerialT m (Array Word8)
316 -> m ()
317fromChunks addr port xs =
318 S.drain $ withConnection addr port (\sk -> S.yieldM $ ISK.fromChunks sk xs)
320-- | Write a stream of arrays to the supplied IPv4 host address and port
321-- number.
322--
323-- @since 0.7.0
324{-# INLINE writeChunks #-}
325writeChunks
326 :: (MonadAsync m, MonadCatch m)
327 => (Word8, Word8, Word8, Word8)
328 -> PortNumber
329 -> Fold m (Array Word8) ()
330writeChunks addr port = Fold step initial extract
331 where
332 initial = do
333 skt <- liftIO (connect addr port)
334 fld <- FL.initialize (SK.writeChunks skt)
335 `MC.onException` (liftIO $ Net.close skt)
336 return (fld, skt)
337 step (fld, skt) x = do
338 r <- FL.runStep fld x `MC.onException` (liftIO $ Net.close skt)
339 return (r, skt)
340 extract ((Fold _ initial1 extract1), skt) = do
341 liftIO $ Net.close skt
342 initial1 >>= extract1
344-- | Like 'write' but provides control over the write buffer. Output will
345-- be written to the IO device as soon as we collect the specified number of
346-- input elements.
347--
348-- @since 0.7.0
349{-# INLINE fromBytesWithBufferOf #-}
350fromBytesWithBufferOf
351 :: (MonadCatch m, MonadAsync m)
352 => Int
353 -> (Word8, Word8, Word8, Word8)
354 -> PortNumber
355 -> SerialT m Word8
356 -> m ()
357fromBytesWithBufferOf n addr port m = fromChunks addr port $ AS.arraysOf n m
359-- | Like 'write' but provides control over the write buffer. Output will
360-- be written to the IO device as soon as we collect the specified number of
361-- input elements.
362--
363-- @since 0.7.0
364{-# INLINE writeWithBufferOf #-}
365writeWithBufferOf
366 :: (MonadAsync m, MonadCatch m)
367 => Int
368 -> (Word8, Word8, Word8, Word8)
369 -> PortNumber
370 -> Fold m Word8 ()
371writeWithBufferOf n addr port =
372 FL.lchunksOf n (writeNUnsafe n) (writeChunks addr port)
374-- | Write a stream to the supplied IPv4 host address and port number.
375--
376-- @since 0.7.0
377{-# INLINE fromBytes #-}
378fromBytes :: (MonadCatch m, MonadAsync m)
379 => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Word8 -> m ()
380fromBytes = fromBytesWithBufferOf defaultChunkSize
382-- | Write a stream to the supplied IPv4 host address and port number.
383--
384-- @since 0.7.0
385{-# INLINE write #-}
386write :: (MonadAsync m, MonadCatch m)
387 => (Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()
388write = writeWithBufferOf defaultChunkSize