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