diff --git a/NEWS b/NEWS --- a/NEWS +++ b/NEWS @@ -1,3 +1,12 @@ +# Version HEAD + +* Significantly upgraded the API and renamed functions to play well with + pipes-4.0.0, pipes-safe-2.0.0 and network-simple-0.3.0. + +* Throw `IOError` in `IO` in order to report timeout errors. Delete + the `Timeout` data-type. + + # Version 0.5.1.0 * Re-export `Network.Socket.withSocketsDo`. diff --git a/PEOPLE b/PEOPLE --- a/PEOPLE +++ b/PEOPLE @@ -6,3 +6,4 @@ Gabriel Gonzalez Paolo Capriotti Marius Ghita +Kyle Van Berendonck diff --git a/pipes-network.cabal b/pipes-network.cabal --- a/pipes-network.cabal +++ b/pipes-network.cabal @@ -1,5 +1,5 @@ name: pipes-network -version: 0.5.1.0 +version: 0.6.0 license: BSD3 license-file: LICENSE copyright: Copyright (c) Renzo Carbonara 2012-2013, Paolo Capriotti 2012-2012. @@ -19,13 +19,14 @@ . This package is organized using the following namespaces: . - * "Control.Proxy.TCP" exports 'Control.Proxy.Proxy's and functions for - establishing and using TCP connections. + * "Pipes.Network.TCP" exports pipes and utilities for using TCP connections in + a streaming fashion. . - * "Control.Proxy.TCP.Safe" is similar to "Control.Proxy.TCP", except - the exported 'Control.Proxy.Proxy's themselves can obtain new network - resources safely by using the facilities providied by the @pipes-safe@ - package. + * "Pipes.Network.TCP.Safe" subsumes "Pipes.Network.TCP", exporting pipes and + functions that allow you to safely establish new TCP connections within a + pipeline using the @pipes-safe@ facilities. You only need to use this module + if you want to safely acquire and release operating system resources within a + pipeline. . See the @NEWS@ file in the source distribution to learn about any important changes between version. @@ -40,27 +41,12 @@ base (==4.*), bytestring (>=0.9.2.1), network, - network-simple (>=0.2.1 && <0.3), - pipes (>=3.2 && <3.4), - pipes-safe (>=1.1 && <1.3), + network-simple (>=0.3 && <0.4), + pipes (>=4.0 && <4.1), + pipes-safe (>=2.0 && <2.1), transformers (>=0.2 && <0.4) exposed-modules: - Control.Proxy.TCP - Control.Proxy.TCP.Safe - ghc-options: -Wall -fno-warn-unused-do-bind + Pipes.Network.TCP + Pipes.Network.TCP.Safe + ghc-options: -Wall -O2 -test-suite simple - hs-source-dirs: tests - main-is: Simple.hs - type: exitcode-stdio-1.0 - build-depends: - base < 5, - bytestring, - HUnit, - network, - pipes-network, - pipes, - pipes-safe, - transformers, - test-framework, - test-framework-hunit diff --git a/src/Control/Proxy/TCP.hs b/src/Control/Proxy/TCP.hs deleted file mode 100644 --- a/src/Control/Proxy/TCP.hs +++ /dev/null @@ -1,250 +0,0 @@ -{-# LANGUAGE DeriveDataTypeable #-} - --- | This module exports functions that allow you to safely use 'NS.Socket' --- resources acquired and released outside a 'P.Proxy' pipeline. --- --- Instead, if want to safely acquire and release resources within the --- pipeline itself, then you should use the functions exported by --- "Control.Proxy.TCP.Safe". --- --- This module re-exports many functions from "Network.Simple.TCP" --- module in the @network-simple@ package. You might refer to that --- module for more documentation. - - -module Control.Proxy.TCP ( - -- * Client side - -- $client-side - S.connect - - -- * Server side - -- $server-side - , S.serve - -- ** Listening - , S.listen - -- ** Accepting - , S.accept - , S.acceptFork - - -- * Socket streams - -- $socket-streaming - , socketReadS - , nsocketReadS - , socketWriteD - -- ** Timeouts - -- $socket-streaming-timeout - , socketReadTimeoutS - , nsocketReadTimeoutS - , socketWriteTimeoutD - - -- * Note to Windows users - -- $windows-users - , NS.withSocketsDo - - -- * Types - , S.HostPreference(..) - , Timeout(..) - ) where - -import qualified Control.Exception as E -import Control.Monad.Trans.Class -import qualified Control.Proxy as P -import qualified Control.Proxy.Trans.Either as PE -import qualified Data.ByteString as B -import Data.Data (Data,Typeable) -import Data.Monoid -import qualified Network.Socket as NS -import qualified Network.Simple.TCP as S -import System.Timeout (timeout) - - --------------------------------------------------------------------------------- - --- $windows-users --- --- If you are running Windows, then you /must/ call 'NS.withSocketsDo', just --- once, right at the beginning of your program. That is, change your program's --- 'main' function from: --- --- @ --- main = do --- print \"Hello world\" --- -- rest of the program... --- @ --- --- To: --- --- @ --- main = 'NS.withSocketsDo' $ do --- print \"Hello world\" --- -- rest of the program... --- @ --- --- If you don't do this, your networking code won't work and you will get many --- unexpected errors at runtime. If you use an operating system other than --- Windows then you don't need to do this, but it is harmless to do it, so it's --- recommended that you do for portability reasons. - --------------------------------------------------------------------------------- - --- $client-side --- --- Here's how you could run a TCP client: --- --- @ --- 'S.connect' \"www.example.org\" \"80\" $ \(connectionSocket, remoteAddr) -> do --- putStrLn $ \"Connection established to \" ++ show remoteAddr --- -- Now you may use connectionSocket as you please within this scope, --- -- possibly using 'socketReadS', 'socketWriteD' or similar proxies --- -- explained below. --- @ - --------------------------------------------------------------------------------- - --- $server-side --- --- Here's how you can run a TCP server that handles in different threads each --- incoming connection to port @8000@ at IPv4 address @127.0.0.1@: --- --- @ --- 'S.serve' ('S.Host' \"127.0.0.1\") \"8000\" $ \(connectionSocket, remoteAddr) -> do --- putStrLn $ \"TCP connection established from \" ++ show remoteAddr --- -- Now you may use connectionSocket as you please within this scope, --- -- possibly using 'socketReadS', 'socketWriteD' or similar proxies --- -- explained below. --- @ --- --- If you need more control on the way your server runs, then you can use more --- advanced functions such as 'listen', 'accept' and 'acceptFork'. - --------------------------------------------------------------------------------- - --- $socket-streaming --- --- Once you have a connected 'NS.Socket', you can use the following 'P.Proxy's --- to interact with the other connection end using streams. - --- | Receives bytes from the remote end sends them downstream. --- --- Less than the specified maximum number of bytes might be received at once. --- --- This proxy returns if the remote peer closes its side of the connection or --- EOF is received. -socketReadS - :: P.Proxy p - => Int -- ^Maximum number of bytes to receive and send - -- dowstream at once. Any positive value is fine, the - -- optimal value depends on how you deal with the - -- received data. Try using @4096@ if you don't care. - -> NS.Socket -- ^Connected socket. - -> () -> P.Producer p B.ByteString IO () -socketReadS nbytes sock () = P.runIdentityP loop where - loop = do - mbs <- lift (S.recv sock nbytes) - case mbs of - Just bs -> P.respond bs >> loop - Nothing -> return () -{-# INLINABLE socketReadS #-} - --- | Just like 'socketReadS', except each request from downstream specifies the --- maximum number of bytes to receive. -nsocketReadS - :: P.Proxy p - => NS.Socket -- ^Connected socket. - -> Int -> P.Server p Int B.ByteString IO () -nsocketReadS sock = P.runIdentityK loop where - loop nbytes = do - mbs <- lift (S.recv sock nbytes) - case mbs of - Just bs -> P.respond bs >>= loop - Nothing -> return () -{-# INLINABLE nsocketReadS #-} - --- | Sends to the remote end the bytes received from upstream, then forwards --- such same bytes downstream. --- --- Requests from downstream are forwarded upstream. -socketWriteD - :: P.Proxy p - => NS.Socket -- ^Connected socket. - -> x -> p x B.ByteString x B.ByteString IO r -socketWriteD sock = P.runIdentityK loop where - loop x = do - a <- P.request x - lift (S.send sock a) - P.respond a >>= loop -{-# INLINABLE socketWriteD #-} - --------------------------------------------------------------------------------- - --- $socket-streaming-timeout --- --- These proxies behave like the similarly named ones above, except support for --- timing out the interaction with the remote end is added. - --- | Like 'socketReadS', except it throws a 'Timeout' exception in the --- 'PE.EitherP' proxy transformer if receiving data from the remote end takes --- more time than specified. -socketReadTimeoutS - :: P.Proxy p - => Int -- ^Timeout in microseconds (1/10^6 seconds). - -> Int -- ^Maximum number of bytes to receive and send - -- dowstream at once. Any positive value is fine, the - -- optimal value depends on how you deal with the - -- received data. Try using @4096@ if you don't care. - -> NS.Socket -- ^Connected socket. - -> () -> P.Producer (PE.EitherP Timeout p) B.ByteString IO () -socketReadTimeoutS wait nbytes sock () = loop where - loop = do - mmbs <- lift (timeout wait (S.recv sock nbytes)) - case mmbs of - Just (Just bs) -> P.respond bs >> loop - Just Nothing -> return () - Nothing -> PE.throw ex - ex = Timeout $ "socketReadTimeoutS: " <> show wait <> " microseconds." -{-# INLINABLE socketReadTimeoutS #-} - --- | Like 'nsocketReadS', except it throws a 'Timeout' exception in the --- 'PE.EitherP' proxy transformer if receiving data from the remote end takes --- more time than specified. -nsocketReadTimeoutS - :: P.Proxy p - => Int -- ^Timeout in microseconds (1/10^6 seconds). - -> NS.Socket -- ^Connected socket. - -> Int -> P.Server (PE.EitherP Timeout p) Int B.ByteString IO () -nsocketReadTimeoutS wait sock = loop where - loop nbytes = do - mmbs <- lift (timeout wait (S.recv sock nbytes)) - case mmbs of - Just (Just bs) -> P.respond bs >>= loop - Just Nothing -> return () - Nothing -> PE.throw ex - ex = Timeout $ "nsocketReadTimeoutS: " <> show wait <> " microseconds." -{-# INLINABLE nsocketReadTimeoutS #-} - --- | Like 'socketWriteD', except it throws a 'Timeout' exception in the --- 'PE.EitherP' proxy transformer if sending data to the remote end takes --- more time than specified. -socketWriteTimeoutD - :: P.Proxy p - => Int -- ^Timeout in microseconds (1/10^6 seconds). - -> NS.Socket -- ^Connected socket. - -> x -> (PE.EitherP Timeout p) x B.ByteString x B.ByteString IO r -socketWriteTimeoutD wait sock = loop where - loop x = do - a <- P.request x - m <- lift (timeout wait (S.send sock a)) - case m of - Just () -> P.respond a >>= loop - Nothing -> PE.throw ex - ex = Timeout $ "socketWriteTimeoutD: " <> show wait <> " microseconds." -{-# INLINABLE socketWriteTimeoutD #-} - --------------------------------------------------------------------------------- - --- |Exception thrown when a time limit has elapsed. -data Timeout - = Timeout String -- ^Timeouted with an additional explanatory message. - deriving (Eq, Show, Data, Typeable) - -instance E.Exception Timeout where diff --git a/src/Control/Proxy/TCP/Safe.hs b/src/Control/Proxy/TCP/Safe.hs deleted file mode 100644 --- a/src/Control/Proxy/TCP/Safe.hs +++ /dev/null @@ -1,468 +0,0 @@ -{-# LANGUAGE Rank2Types #-} - --- | This module exports functions that allow you to safely use 'NS.Socket' --- resources within a 'P.Proxy' pipeline, possibly acquiring and releasing such --- resources within the pipeline itself, using the facilities provided by --- 'P.ExceptionP' from the @pipes-safe@ library. --- --- Instead, if just want to use resources already acquired or released outside --- the pipeline, then you could use the simpler functions exported by --- "Control.Proxy.TCP". - -module Control.Proxy.TCP.Safe ( - -- * Client side - -- $client-side - connect, - -- ** Streaming - -- $client-streaming - connectReadS, - connectWriteD, - - -- * Server side - -- $server-side - serve, - -- ** Listening - listen, - -- ** Accepting - accept, - acceptFork, - -- ** Streaming - -- $server-streaming - serveReadS, - serveWriteD, - - -- * Socket streams - -- $socket-streaming - socketReadS, - nsocketReadS, - socketWriteD, - - -- * Note to Windows users - -- $windows-users - NS.withSocketsDo, - - - -- * Exports - S.HostPreference(..), - Timeout(..) - ) where - -import Control.Concurrent (ThreadId) -import Control.Monad -import qualified Control.Proxy as P -import qualified Control.Proxy.Safe as P -import Control.Proxy.TCP (Timeout(..)) -import qualified Data.ByteString as B -import Data.Monoid -import qualified Network.Socket as NS -import qualified Network.Simple.TCP as S -import System.Timeout (timeout) - --------------------------------------------------------------------------------- - --- $windows-users --- --- If you are running Windows, then you /must/ call 'NS.withSocketsDo', just --- once, right at the beginning of your program. That is, change your program's --- 'main' function from: --- --- @ --- main = do --- print \"Hello world\" --- -- rest of the program... --- @ --- --- To: --- --- @ --- main = 'NS.withSocketsDo' $ do --- print \"Hello world\" --- -- rest of the program... --- @ --- --- If you don't do this, your networking code won't work and you will get many --- unexpected errors at runtime. If you use an operating system other than --- Windows then you don't need to do this, but it is harmless to do it, so it's --- recommended that you do for portability reasons. - --------------------------------------------------------------------------------- - --- $client-side --- --- Here's how you could run a TCP client: --- --- @ --- 'connect' \"www.example.org\" \"80\" $ \(connectionSocket, remoteAddr) -> do --- putStrLn $ \"Connection established to \" ++ show remoteAddr --- -- Now you may use connectionSocket as you please within this scope, --- -- possibly using 'socketReadS', 'socketWriteD' or similar proxies --- -- explained below. --- @ --- --- You might instead prefer the simpler but less general solutions offered by --- 'connectReadS' and 'connectWriteD', so check those too. - --------------------------------------------------------------------------------- - --- | Connect to a TCP server and use the connection. --- --- The connection socket is closed when done or in case of exceptions. --- --- If you prefer to acquire close the socket yourself, then use --- 'S.connectSock' and the 'NS.sClose' from "Network.Socket" instead. -connect - :: (P.Proxy p, Monad m) - => (forall x. P.SafeIO x -> m x) -- ^Monad morphism. - -> NS.HostName -- ^Server hostname. - -> NS.ServiceName -- ^Server service port. - -> ((NS.Socket, NS.SockAddr) -> P.ExceptionP p a' a b' b m r) - -- ^Computation taking the - -- communication socket and the server - -- address. - -> P.ExceptionP p a' a b' b m r -connect morph host port = - P.bracket morph (S.connectSock host port) (NS.sClose . fst) - --------------------------------------------------------------------------------- - --- $client-streaming --- --- The following proxies allow you to easily connect to a TCP server and --- immediately interact with it using streams, all at once, instead of --- having to perform the individual steps separately. - --------------------------------------------------------------------------------- - --- | Connect to a TCP server and send downstream the bytes received from the --- remote end. --- --- If an optional timeout is given and receiveing data from the remote end takes --- more time that such timeout, then throw a 'Timeout' exception in the --- 'P.ExceptionP' proxy transformer. --- --- The connection socket is closed when done or in case of exceptions. --- --- Using this proxy you can write straightforward code like the following, which --- prints whatever is received from a single TCP connection to a given server --- listening locally on port 9000, in chunks of up to 4096 bytes: --- --- >>> runSafeIO . runProxy . runEitherK $ connectReadS Nothing 4096 "127.0.0.1" "9000" >-> tryK printD -connectReadS - :: P.Proxy p - => Maybe Int -- ^Optional timeout in microseconds (1/10^6 seconds). - -> Int -- ^Maximum number of bytes to receive and send - -- dowstream at once. Any positive value is fine, the - -- optimal value depends on how you deal with the - -- received data. Try using @4096@ if you don't care. - -> NS.HostName -- ^Server host name. - -> NS.ServiceName -- ^Server service port. - -> () -> P.Producer (P.ExceptionP p) B.ByteString P.SafeIO () -connectReadS mwait nbytes host port () = do - connect id host port $ \(csock,_) -> do - socketReadS mwait nbytes csock () - --- | Connects to a TCP server, sends to the remote end the bytes received from --- upstream, then forwards such same bytes downstream. --- --- Requests from downstream are forwarded upstream. --- --- If an optional timeout is given and sending data to the remote end takes --- more time that such timeout, then throw a 'Timeout' exception in the --- 'P.ExceptionP' proxy transformer. --- --- The connection socket is closed when done or in case of exceptions. --- --- Using this proxy you can write straightforward code like the following, which --- greets a TCP client listening locally at port 9000: --- --- >>> :set -XOverloadedStrings --- >>> runSafeIO . runProxy . runEitherK $ fromListS ["He","llo\r\n"] >-> connectWriteD Nothing "127.0.0.1" "9000" -connectWriteD - :: P.Proxy p - => Maybe Int -- ^Optional timeout in microseconds (1/10^6 seconds). - -> NS.HostName -- ^Server host name. - -> NS.ServiceName -- ^Server service port. - -> x -> (P.ExceptionP p) x B.ByteString x B.ByteString P.SafeIO r -connectWriteD mwait hp port x = do - connect id hp port $ \(csock,_) -> - socketWriteD mwait csock x - --------------------------------------------------------------------------------- - --- $server-side --- --- Here's how you can run a TCP server that handles in different threads each --- incoming connection to port @8000@ at IPv4 address @127.0.0.1@: --- --- @ --- 'serve' ('Host' \"127.0.0.1\") \"8000\" $ \(connectionSocket, remoteAddr) -> do --- putStrLn $ \"TCP connection established from \" ++ show remoteAddr --- -- Now you may use connectionSocket as you please within this scope, --- -- possibly using 'socketReadS', 'socketWriteD' or similar proxies --- -- explained below. --- @ --- --- You might instead prefer the simpler but less general solutions offered by --- 'serveReadS' and 'serveWriteD', so check those too. On the other hand, --- if you need more control on the way your server runs, then you can use more --- advanced functions such as 'listen', 'accept' and 'acceptFork'. - --------------------------------------------------------------------------------- - --- | Start a TCP server that accepts incoming connections and handles each of --- them concurrently in different threads. --- --- Any acquired network resources are properly closed and discarded when done or --- in case of exceptions. --- --- Note: This function performs 'listen' and 'acceptFork', so you don't need to --- perform those manually. -serve - :: (P.Proxy p, Monad m) - => (forall x. P.SafeIO x -> m x) -- ^Monad morphism. - -> S.HostPreference -- ^Preferred host to bind. - -> NS.ServiceName -- ^Service port to bind. - -> ((NS.Socket, NS.SockAddr) -> IO ()) - -- ^Computation to run in a different thread - -- once an incoming connection is accepted. - -- Takes the connection socket and remote end - -- address. - -> P.ExceptionP p a' a b' b m r -serve morph hp port k = do - listen morph hp port $ \(lsock,_) -> do - forever $ acceptFork morph lsock k - --------------------------------------------------------------------------------- - --- | Bind a TCP listening socket and use it. --- --- The listening socket is closed when done or in case of exceptions. --- --- If you prefer to acquire and close the socket yourself, then use --- 'S.bindSock' and the 'NS.listen' and 'NS.sClose' functions from --- "Network.Socket" instead. --- --- Note: 'N.maxListenQueue' is tipically 128, which is too small for high --- performance servers. So, we use the maximum between 'N.maxListenQueue' and --- 2048 as the default size of the listening queue. -listen - :: (P.Proxy p, Monad m) - => (forall x. P.SafeIO x -> m x) -- ^Monad morphism. - -> S.HostPreference -- ^Preferred host to bind. - -> NS.ServiceName -- ^Service port to bind. - -> ((NS.Socket, NS.SockAddr) -> P.ExceptionP p a' a b' b m r) - -- ^Computation taking the listening - -- socket and the address it's bound to. - -> P.ExceptionP p a' a b' b m r -listen morph hp port = P.bracket morph listen' (NS.sClose . fst) - where - listen' = do x@(bsock,_) <- S.bindSock hp port - NS.listen bsock $ max 2048 NS.maxListenQueue - return x - --------------------------------------------------------------------------------- - --- | Accept a single incoming connection and use it. --- --- The connection socket is closed when done or in case of exceptions. -accept - :: (P.Proxy p, Monad m) - => (forall x. P.SafeIO x -> m x) -- ^Monad morphism. - -> NS.Socket -- ^Listening and bound socket. - -> ((NS.Socket, NS.SockAddr) -> P.ExceptionP p a' a b' b m r) - -- ^Computation to run once an incoming - -- connection is accepted. Takes the - -- connection socket and remote end address. - -> P.ExceptionP p a' a b' b m r -accept morph lsock k = do - conn@(csock,_) <- P.hoist morph . P.tryIO $ NS.accept lsock - P.finally morph (NS.sClose csock) (k conn) -{-# INLINABLE accept #-} - --- | Accept a single incoming connection and use it in a different thread. --- --- The connection socket is closed when done or in case of exceptions. -acceptFork - :: (P.Proxy p, Monad m) - => (forall x. P.SafeIO x -> m x) -- ^Monad morphism. - -> NS.Socket -- ^Listening and bound socket. - -> ((NS.Socket, NS.SockAddr) -> IO ()) - -- ^Computation to run in a different thread - -- once an incoming connection is accepted. - -- Takes the connection socket and remote end - -- address. - -> P.ExceptionP p a' a b' b m ThreadId -acceptFork morph lsock k = P.hoist morph . P.tryIO $ S.acceptFork lsock k -{-# INLINABLE acceptFork #-} - --------------------------------------------------------------------------------- - --- $server-streaming --- --- The following proxies allow you to easily run a TCP server and immediately --- interact with incoming connections using streams, all at once, instead of --- having to perform the individual steps separately. - --------------------------------------------------------------------------------- - --- | Binds a listening socket, accepts a single connection and sends downstream --- any bytes received from the remote end. --- --- If an optional timeout is given and receiveing data from the remote end takes --- more time that such timeout, then throw a 'Timeout' exception in the --- 'P.ExceptionP' proxy transformer. --- --- Less than the specified maximum number of bytes might be received at once. --- --- This proxy returns if the remote peer closes its side of the connection or --- EOF is received. --- --- Both the listening and connection sockets are closed when done or in case of --- exceptions. --- --- Using this proxy you can write straightforward code like the following, which --- prints whatever is received from a single TCP connection to port 9000, in --- chunks of up to 4096 bytes. --- --- >>> :set -XOverloadedStrings --- >>> runSafeIO . runProxy . runEitherK $ serveReadS Nothing 4096 "127.0.0.1" "9000" >-> tryK printD -serveReadS - :: P.Proxy p - => Maybe Int -- ^Optional timeout in microseconds (1/10^6 seconds). - -> Int -- ^Maximum number of bytes to receive and send - -- dowstream at once. Any positive value is fine, the - -- optimal value depends on how you deal with the - -- received data. Try using @4096@ if you don't care. - -> S.HostPreference -- ^Preferred host to bind. - -> NS.ServiceName -- ^Service port to bind. - -> () -> P.Producer (P.ExceptionP p) B.ByteString P.SafeIO () -serveReadS mwait nbytes hp port () = do - listen id hp port $ \(lsock,_) -> do - accept id lsock $ \(csock,_) -> do - socketReadS mwait nbytes csock () - --- | Binds a listening socket, accepts a single connection, sends to the remote --- end the bytes received from upstream, then forwards such sames bytes --- downstream. --- --- Requests from downstream are forwarded upstream. --- --- If an optional timeout is given and sending data to the remote end takes --- more time that such timeout, then throw a 'Timeout' exception in the --- 'P.ExceptionP' proxy transformer. --- --- Both the listening and connection sockets are closed when done or in case of --- exceptions. --- --- Using this proxy you can write straightforward code like the following, which --- greets a TCP client connecting to port 9000: --- --- >>> :set -XOverloadedStrings --- >>> runSafeIO . runProxy . runEitherK $ fromListS ["He","llo\r\n"] >-> serveWriteD Nothing "127.0.0.1" "9000" -serveWriteD - :: P.Proxy p - => Maybe Int -- ^Optional timeout in microseconds (1/10^6 seconds). - -> S.HostPreference -- ^Preferred host to bind. - -> NS.ServiceName -- ^Service port to bind. - -> x -> (P.ExceptionP p) x B.ByteString x B.ByteString P.SafeIO r -serveWriteD mwait hp port x = do - listen id hp port $ \(lsock,_) -> do - accept id lsock $ \(csock,_) -> do - socketWriteD mwait csock x - --------------------------------------------------------------------------------- - --- $socket-streaming --- --- Once you have a connected 'NS.Socket', you can use the following 'P.Proxy's --- to interact with the other connection end using streams. - --------------------------------------------------------------------------------- - --- | Receives bytes from the remote end and sends them downstream. --- --- If an optional timeout is given and receiveing data from the remote end takes --- more time that such timeout, then throw a 'Timeout' exception in the --- 'P.ExceptionP' proxy transformer. --- --- Less than the specified maximum number of bytes might be received at once. --- --- This proxy returns if the remote peer closes its side of the connection or --- EOF is received. -socketReadS - :: P.Proxy p - => Maybe Int -- ^Optional timeout in microseconds (1/10^6 seconds). - -> Int -- ^Maximum number of bytes to receive and send - -- dowstream at once. Any positive value is fine, the - -- optimal value depends on how you deal with the - -- received data. Try using @4096@ if you don't care. - -> NS.Socket -- ^Connected socket. - -> () -> P.Producer (P.ExceptionP p) B.ByteString P.SafeIO () -socketReadS Nothing nbytes sock () = loop where - loop = do - mbs <- P.tryIO (S.recv sock nbytes) - case mbs of - Just bs -> P.respond bs >> loop - Nothing -> return () -socketReadS (Just wait) nbytes sock () = loop where - loop = do - mmbs <- P.tryIO (timeout wait (S.recv sock nbytes)) - case mmbs of - Just (Just bs) -> P.respond bs >> loop - Just Nothing -> return () - Nothing -> P.throw ex - ex = Timeout $ "socketReadS: " <> show wait <> " microseconds." -{-# INLINABLE socketReadS #-} - --- | Just like 'socketReadS', except each request from downstream specifies the --- maximum number of bytes to receive. -nsocketReadS - :: P.Proxy p - => Maybe Int -- ^Optional timeout in microseconds (1/10^6 seconds). - -> NS.Socket -- ^Connected socket. - -> Int -> P.Server (P.ExceptionP p) Int B.ByteString P.SafeIO () -nsocketReadS Nothing sock = loop where - loop nbytes = do - mbs <- P.tryIO (S.recv sock nbytes) - case mbs of - Just bs -> P.respond bs >>= loop - Nothing -> return () -nsocketReadS (Just wait) sock = loop where - loop nbytes = do - mbs <- P.tryIO (timeout wait (S.recv sock nbytes)) - case mbs of - Just (Just bs) -> P.respond bs >>= loop - Just Nothing -> return () - Nothing -> P.throw ex - ex = Timeout $ "nsocketReadS: " <> show wait <> " microseconds." -{-# INLINABLE nsocketReadS #-} - --- | Sends to the remote end the bytes received from upstream, then forwards --- such same bytes downstream. --- --- If an optional timeout is given and sending data to the remote end takes --- more time that such timeout, then throw a 'Timeout' exception in the --- 'P.ExceptionP' proxy transformer. --- --- Requests from downstream are forwarded upstream. -socketWriteD - :: P.Proxy p - => Maybe Int -- ^Optional timeout in microseconds (1/10^6 seconds). - -> NS.Socket -- ^Connected socket. - -> x -> (P.ExceptionP p) x B.ByteString x B.ByteString P.SafeIO r -socketWriteD Nothing sock = loop where - loop x = do - a <- P.request x - P.tryIO (S.send sock a) - P.respond a >>= loop -socketWriteD (Just wait) sock = loop where - loop x = do - a <- P.request x - m <- P.tryIO (timeout wait (S.send sock a)) - case m of - Just () -> P.respond a >>= loop - Nothing -> P.throw ex - ex = Timeout $ "socketWriteD: " <> show wait <> " microseconds." -{-# INLINABLE socketWriteD #-} - - diff --git a/src/Pipes/Network/TCP.hs b/src/Pipes/Network/TCP.hs new file mode 100644 --- /dev/null +++ b/src/Pipes/Network/TCP.hs @@ -0,0 +1,194 @@ +{-# LANGUAGE RankNTypes #-} + +-- | This minimal module exports facilities that ease the usage of TCP +-- 'Socket's in the /Pipes ecosystem/. It is meant to be used together with +-- the "Network.Simple.TCP" module from the @network-simple@ package, which is +-- completely re-exported from this module. +-- +-- This module /does not/ export facilities that would allow you to acquire new +-- 'Socket's within a pipeline. If you need to do so, then you should use +-- "Pipes.Network.TCP.Safe" instead, which exports a similar API to the one +-- exported by this module. However, don't be confused by the word “safe” in +-- that module; this module is equally safe to use as long as you don't try to +-- acquire resources within the pipeline. + +module Pipes.Network.TCP ( + -- * Receiving + -- $receiving + fromSocket + , fromSocketTimeout + -- ** Bidirectional pipes + -- $bidirectional + , fromSocketN + , fromSocketTimeoutN + -- * Sending + -- $sending + , toSocket + , toSocketTimeout + -- * Exports + -- $exports + , module Network.Simple.TCP + ) where + +import qualified Data.ByteString as B +import Foreign.C.Error (errnoToIOError, eTIMEDOUT) +import qualified Network.Socket.ByteString as NSB +import Network.Simple.TCP + (connect, serve, listen, accept, acceptFork, + bindSock, connectSock, recv, send, withSocketsDo, + HostName, HostPreference(HostAny, HostIPv4, HostIPv6, Host), + ServiceName, SockAddr, Socket) +import Pipes +import Pipes.Core +import System.Timeout (timeout) + +-------------------------------------------------------------------------------- + +-- $receiving +-- +-- The following pipes allow you to receive bytes from the remote end. +-- +-- Besides the pipes below, you might want to use "Network.Simple.TCP"'s +-- 'Network.Simple.TCP.recv', which happens to be an 'Effect'': +-- +-- @ +-- 'Network.Simple.TCP.recv' :: 'MonadIO' m => 'Socket' -> 'Int' -> 'Effect'' m ('Maybe' 'B.ByteString') +-- @ + +-------------------------------------------------------------------------------- + +-- | Receives bytes from the remote end sends them downstream. +-- +-- The number of bytes received at once is always in the interval +-- /[1 .. specified maximum]/. +-- +-- This 'Producer'' returns if the remote peer closes its side of the connection +-- or EOF is received. +fromSocket + :: MonadIO m + => Socket -- ^Connected socket. + -> Int -- ^Maximum number of bytes to receive and send + -- dowstream at once. Any positive value is fine, the + -- optimal value depends on how you deal with the + -- received data. Try using @4096@ if you don't care. + -> Producer' B.ByteString m () +fromSocket sock nbytes = loop where + loop = do + bs <- liftIO (NSB.recv sock nbytes) + if B.null bs + then return () + else yield bs >> loop +{-# INLINABLE fromSocket #-} + +-- | Like 'fromSocket', except with the first 'Int' argument you can specify +-- the maximum time that each interaction with the remote end can take. If such +-- time elapses before the interaction finishes, then an 'IOError' exception is +-- thrown. The time is specified in microseconds (10e6). +fromSocketTimeout + :: MonadIO m + => Int -> Socket -> Int -> Producer' B.ByteString m () +fromSocketTimeout wait sock nbytes = loop where + loop = do + mbs <- liftIO (timeout wait (NSB.recv sock nbytes)) + case mbs of + Just bs -> yield bs >> loop + Nothing -> liftIO $ ioError $ errnoToIOError + "Pipes.Network.TCP.fromSocketTimeout" eTIMEDOUT Nothing Nothing +{-# INLINABLE fromSocketTimeout #-} + +-------------------------------------------------------------------------------- + +-- $bidirectional +-- +-- The following pipes are bidirectional, which means useful data can flow +-- through them upstream and downstream. If you don't care about bidirectional +-- pipes, just skip this section. + +-------------------------------------------------------------------------------- + +-- | Like 'fromSocket', except the downstream pipe can specify the maximum +-- number of bytes to receive at once using 'request'. +fromSocketN :: MonadIO m => Socket -> Int -> Server' Int B.ByteString m () +fromSocketN sock = loop where + loop = \nbytes -> do + bs <- liftIO (NSB.recv sock nbytes) + if B.null bs + then return () + else respond bs >>= loop +{-# INLINABLE fromSocketN #-} + + +-- | Like 'fromSocketN', except with the first 'Int' argument you can specify +-- the maximum time that each interaction with the remote end can take. If such +-- time elapses before the interaction finishes, then an 'IOError' exception is +-- thrown. The time is specified in microseconds (10e6). +fromSocketTimeoutN + :: MonadIO m + => Int -> Socket -> Int -> Server' Int B.ByteString m () +fromSocketTimeoutN wait sock = loop where + loop = \nbytes -> do + mbs <- liftIO (timeout wait (NSB.recv sock nbytes)) + case mbs of + Just bs -> respond bs >>= loop + Nothing -> liftIO $ ioError $ errnoToIOError + "Pipes.Network.TCP.fromSocketTimeoutN" eTIMEDOUT Nothing Nothing +{-# INLINABLE fromSocketTimeoutN #-} + +-------------------------------------------------------------------------------- + +-- $sending +-- +-- The following pipes allow you to send bytes to the remote end. +-- +-- Besides the pipes below, you might want to use "Network.Simple.TCP"'s +-- 'Network.Simple.TCP.send', which happens to be an 'Effect'': +-- +-- @ +-- 'Network.Simple.TCP.send' :: 'MonadIO' m => 'Socket' -> 'B.ByteString' -> 'Effect'' m () +-- @ + +-- | Sends to the remote end each 'B.ByteString' received from upstream. +toSocket + :: MonadIO m + => Socket -- ^Connected socket. + -> Consumer' B.ByteString m r +toSocket sock = for cat (\a -> send sock a) +{-# INLINABLE toSocket #-} + +-- | Like 'toSocket', except with the first 'Int' argument you can specify +-- the maximum time that each interaction with the remote end can take. If such +-- time elapses before the interaction finishes, then an 'IOError' exception is +-- thrown. The time is specified in microseconds (10e6). +toSocketTimeout + :: MonadIO m + => Int -> Socket -> Consumer' B.ByteString m r +toSocketTimeout wait sock = for cat $ \a -> do + mu <- liftIO (timeout wait (NSB.sendAll sock a)) + case mu of + Just () -> return () + Nothing -> liftIO $ ioError $ errnoToIOError + "Pipes.Network.TCP.toSocketTimeout" eTIMEDOUT Nothing Nothing +{-# INLINABLE toSocketTimeout #-} + +-------------------------------------------------------------------------------- + +-- $exports +-- +-- [From "Network.Simple.TCP"] +-- 'accept', +-- 'acceptFork', +-- 'bindSock', +-- 'connect', +-- 'connectSock', +-- 'HostPreference'('HostAny','HostIPv4','HostIPv6','Host'), +-- 'listen', +-- 'recv', +-- 'send', +-- 'serve'. +-- +-- [From "Network.Socket"] +-- 'HostName', +-- 'ServiceName', +-- 'SockAddr', +-- 'Socket', +-- 'withSocketsDo'. diff --git a/src/Pipes/Network/TCP/Safe.hs b/src/Pipes/Network/TCP/Safe.hs new file mode 100644 --- /dev/null +++ b/src/Pipes/Network/TCP/Safe.hs @@ -0,0 +1,246 @@ +{-# LANGUAGE Rank2Types, TypeFamilies #-} + +-- | This module exports facilities allowing you to safely obtain, use and +-- release 'Socket' resources within a /Pipes/ pipeline, by relying on +-- @pipes-safe@. +-- +-- This module is meant to be used as a replacement of "Pipes.Network.TCP", +-- and as such it overrides some functions from "Network.Simple.TCP" so that +-- they use 'Ps.MonadSafe' instead of 'IO' as their base monad. Additionally, +-- It also exports pipes that establish a TCP connection and interact with +-- it in a streaming fashion at once. +-- +-- If you just want to use 'Socket' obtained outside the /Pipes/ pipeline, +-- then you can just ignore this module and use the simpler module +-- "Pipes.Network.TCP" instead. + +module Pipes.Network.TCP.Safe ( + -- * @MonadSafe@-compatible upgrades + -- $network-simple-upgrades + connect + , serve + , listen + , accept + -- * Streaming + -- ** Client side + -- $client-streaming + , fromConnect + , toConnect + -- ** Server side + -- $server-streaming + , fromServe + , toServe + -- * Exports + -- $exports + , module Pipes.Network.TCP + , module Network.Simple.TCP + , module Pipes.Safe + ) where + +import Control.Monad +import qualified Data.ByteString as B +import Network.Simple.TCP + (acceptFork, bindSock, connectSock, recv, send, withSocketsDo, + HostName, HostPreference(HostAny, HostIPv4, HostIPv6, Host), + ServiceName, SockAddr, Socket) +import qualified Network.Socket as NS +import Pipes +import Pipes.Network.TCP + (fromSocket, fromSocketTimeout, fromSocketN, + fromSocketTimeoutN, toSocket, toSocketTimeout) +import qualified Pipes.Safe as Ps +import Pipes.Safe (runSafeT) + +-------------------------------------------------------------------------------- + +-- $network-simple-upgrades +-- +-- The following functions are analogous versions of those exported by +-- "Network.Simple.TCP", but compatible with 'Ps.MonadSafe'. + +-- | Like 'Network.Simple.TCP.connect' from "Network.Simple.TCP", but compatible +-- with 'Ps.MonadSafe'. +connect + :: (Ps.MonadSafe m, Ps.Base m ~ IO) + => HostName -> ServiceName -> ((Socket, SockAddr) -> m r) -> m r +connect host port = Ps.bracket (connectSock host port) + (NS.sClose . fst) + +-- | Like 'Network.Simple.TCP.serve' from "Network.Simple.TCP", but compatible +-- with 'Ps.MonadSafe'. +serve + :: (Ps.MonadSafe m, Ps.Base m ~ IO) + => HostPreference -> ServiceName -> ((Socket, SockAddr) -> IO ()) -> m r +serve hp port k = do + listen hp port $ \(lsock,_) -> do + forever $ acceptFork lsock k + +-- | Like 'Network.Simple.TCP.listen' from "Network.Simple.TCP", but compatible +-- with 'Ps.MonadSafe'. +listen + :: (Ps.MonadSafe m, Ps.Base m ~ IO) + => HostPreference -> ServiceName -> ((Socket, SockAddr) -> m r) -> m r +listen hp port = Ps.bracket listen' (NS.sClose . fst) + where + listen' = do x@(bsock,_) <- bindSock hp port + NS.listen bsock (max 2048 NS.maxListenQueue) + return x + +-- | Like 'Network.Simple.TCP.accept' from "Network.Simple.TCP", but compatible +-- with 'Ps.MonadSafe'. +accept + :: (Ps.MonadSafe m, Ps.Base m ~ IO) + => Socket -> ((Socket, SockAddr) -> m r) -> m r +accept lsock k = do + conn@(csock,_) <- liftIO (NS.accept lsock) + Ps.finally (k conn) (NS.sClose csock) +{-# INLINABLE accept #-} + +-------------------------------------------------------------------------------- + +-- $client-streaming +-- +-- The following pipes allow you to easily connect to a TCP server and +-- immediately interact with it in a streaming fashion, all at once, instead of +-- having to perform the individual steps separately. However, keep +-- in mind that you'll be able to interact with the remote end in only one +-- direction, that is, you'll either send or receive data, but not both. + +-------------------------------------------------------------------------------- + +-- | Connect to a TCP server and send downstream the bytes received from the +-- remote end. +-- +-- The connection socket is closed when done or in case of exceptions. +-- +-- Using this 'Producer'' you can write straightforward code like the following, +-- which prints whatever is received from a single TCP connection to a given +-- server listening locally on port 9000, in chunks of up to 4096 bytes: +-- +-- >>> runSafeT . runEffect $ fromConnect 4096 "127.0.0.1" "9000" >-> P.print +fromConnect + :: (Ps.MonadSafe m, Ps.Base m ~ IO) + => Int -- ^Maximum number of bytes to receive and send + -- dowstream at once. Any positive value is fine, the + -- optimal value depends on how you deal with the + -- received data. Try using @4096@ if you don't care. + -> HostName -- ^Server host name. + -> ServiceName -- ^Server service port. + -> Producer' B.ByteString m () +fromConnect nbytes host port = do + connect host port $ \(csock,_) -> do + fromSocket csock nbytes + +-- | Connects to a TCP server, sends to the remote end the bytes received from +-- upstream. +-- +-- The connection socket is closed in case of exceptions. +-- +-- Using this 'Consumer'' you can write straightforward code like the following, +-- which greets a TCP client listening locally at port 9000: +-- +-- >>> :set -XOverloadedStrings +-- >>> runSafeT . runEffect $ each ["He","llo\r\n"] >-> toConnect "127.0.0.1" "9000" +toConnect + :: (Ps.MonadSafe m, Ps.Base m ~ IO) + => HostName -- ^Server host name. + -> ServiceName -- ^Server service port. + -> Consumer' B.ByteString m r +toConnect hp port = do + connect hp port $ \(csock,_) -> do + toSocket csock + +-------------------------------------------------------------------------------- + +-- $server-streaming +-- +-- The following pipes allow you to easily run a TCP server and immediately +-- interact with incoming connections in a streaming fashion, all at once, +-- instead of having to perform the individual steps separately. However, keep +-- in mind that you'll be able to interact with the remote end in only one +-- direction, that is, you'll either send or receive data, but not both. + +-------------------------------------------------------------------------------- + +-- | Binds a listening socket, accepts a single connection and sends downstream +-- any bytes received from the remote end. +-- +-- Less than the specified maximum number of bytes might be received at once. +-- +-- This 'Producer'' returns if the remote peer closes its side of the connection +-- or EOF is received. +-- +-- Both the listening and connection sockets are closed when done or in case of +-- exceptions. +-- +-- Using this 'Producer'' you can write straightforward code like the following, +-- which prints whatever is received from a single TCP connection to port 9000, +-- in chunks of up to 4096 bytes. +-- +-- >>> :set -XOverloadedStrings +-- >>> runSafeT . runEffect $ fromServe 4096 "127.0.0.1" "9000" >-> P.print +fromServe + :: (Ps.MonadSafe m, Ps.Base m ~ IO) + => Int -- ^Maximum number of bytes to receive and send + -- dowstream at once. Any positive value is fine, the + -- optimal value depends on how you deal with the + -- received data. Try using @4096@ if you don't care. + -> HostPreference -- ^Preferred host to bind. + -> ServiceName -- ^Service port to bind. + -> Producer' B.ByteString m () +fromServe nbytes hp port = do + listen hp port $ \(lsock,_) -> do + accept lsock $ \(csock,_) -> do + fromSocket csock nbytes + +-- | Binds a listening socket, accepts a single connection, sends to the remote +-- end the bytes received from upstream. +-- +-- Both the listening and connection sockets are closed when done or in case of +-- exceptions. +-- +-- Using this 'Consumer'' you can write straightforward code like the following, +-- which greets a TCP client connecting to port 9000: +-- +-- >>> :set -XOverloadedStrings +-- >>> runSafeT . runEffect $ each ["He","llo\r\n"] >-> toServe "127.0.0.1" "9000" +toServe + :: (Ps.MonadSafe m, Ps.Base m ~ IO) + => HostPreference -- ^Preferred host to bind. + -> ServiceName -- ^Service port to bind. + -> Consumer' B.ByteString m r +toServe hp port = do + listen hp port $ \(lsock,_) -> do + accept lsock $ \(csock,_) -> do + toSocket csock + +-------------------------------------------------------------------------------- + +-- $exports +-- +-- [From "Pipes.Network.TCP"] +-- 'fromSocket', +-- 'fromSocketN', +-- 'fromSocketTimeout', +-- 'fromSocketTimeoutN', +-- 'toSocket', +-- 'toSocketTimeout'. +-- +-- [From "Network.Simple.TCP"] +-- 'acceptFork', +-- 'bindSock', +-- 'connectSock', +-- 'HostPreference'('HostAny','HostIPv4','HostIPv6','Host'), +-- 'recv', +-- 'send'. +-- +-- [From "Network.Socket"] +-- 'HostName', +-- 'ServiceName', +-- 'SockAddr', +-- 'Socket', +-- 'withSocketsDo'. +-- +-- [From "Pipes.Safe"] +-- 'Ps.runSafeT'. + diff --git a/tests/Simple.hs b/tests/Simple.hs deleted file mode 100644 --- a/tests/Simple.hs +++ /dev/null @@ -1,146 +0,0 @@ -{-# LANGUAGE OverloadedStrings #-} -{-# OPTIONS_GHC -fno-warn-unused-do-bind -fno-warn-missing-signatures #-} - -module Main where - -import Control.Concurrent (forkIO, threadDelay) -import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) -import qualified Control.Exception as E -import qualified Data.ByteString.Char8 as B -import qualified Network.Socket as NS -import Test.Framework (Test, defaultMain, testGroup) -import Test.Framework.Providers.HUnit (testCase) -import Test.HUnit (Assertion, (@=?)) -import Control.Proxy ((>->)) -import qualified Control.Proxy as P -import qualified Control.Proxy.Safe as P -import qualified Control.Proxy.TCP as T -import qualified Control.Proxy.TCP.Safe as T' - -host1 = "127.0.0.1" :: NS.HostName -host1p = T.Host host1 :: T.HostPreference -ports = fmap show [14000..14010] :: [NS.ServiceName] -msg1 = take 1000 $ cycle ["Hell","o\r\n"] :: [B.ByteString] -msg1b = B.concat msg1 :: B.ByteString - - --- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- --- The following 4 IO actions are used throughout the various tests as the --- default implementations for reading/writing a server/client. They themselves --- are also tested below. - --- tested by 'test_listen_accept_socketWriteD_then_connect_socketReadD' -connectAndRead :: NS.HostName -> NS.ServiceName -> IO [B.ByteString] -connectAndRead host port = do - T.connect host port $ \(csock, _caddr) -> do - let p = P.raiseK (T.socketReadS 4096 csock) >-> P.toListD - fmap snd $ P.runWriterT . P.runProxy $ p --- let p = P.raiseK (T'.connectReadS Nothing 4096 host port) >-> P.toListD --- (eex,out) <- P.trySafeIO . P.runWriterT . P.runProxy .P.runEitherK $ p --- case eex of --- Left ex -> E.throw ex --- Right () -> return out - --- tested by 'test_listen_accept_socketReadS_then_connect_socketWriteD' -connectAndWrite :: NS.HostName -> NS.ServiceName -> [B.ByteString] -> IO () -connectAndWrite host port msg = do - T.connect host port $ \(csock, _caddr) -> do - P.runProxy $ P.fromListS msg >-> T.socketWriteD csock --- let p = P.fromListS msg >-> T'.connectWriteD Nothing host1 port --- P.runSafeIO . P.runProxy .P.runEitherK $ p - --- tested by 'test_listen_accept_socketWriteD_then_connect_socketReadD' -serveOnceAndRead :: T.HostPreference -> NS.ServiceName -> IO [B.ByteString] -serveOnceAndRead hp port = do - T.listen hp port $ \(lsock, _laddr) -> do - T.accept lsock $ \(csock, _caddr) -> do - let p = P.raiseK (T.socketReadS 4096 csock) >-> P.toListD - fmap snd $ P.runWriterT . P.runProxy $ p --- let p = P.raiseK (T'.serveReadS Nothing 4096 host1p port) >-> P.toListD --- (eex,out) <- P.trySafeIO . P.runWriterT . P.runProxy .P.runEitherK $ p --- case eex of --- Left ex -> E.throw ex --- Right () -> return out - --- tested by 'test_listen_accept_socketWriteD_then_connect_socketReadD' -serveOnceAndWrite :: T.HostPreference -> NS.ServiceName -> [B.ByteString] -> IO () -serveOnceAndWrite hp port msg = do - T.listen hp port $ \(lsock, _laddr) -> do - T.accept lsock $ \(csock, _caddr) -> do - P.runProxy $ P.fromListS msg >-> T.socketWriteD csock --- let p = P.fromListS msg >-> T'.serveWriteD Nothing host1p port --- P.runSafeIO . P.runProxy .P.runEitherK $ p --- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- - --- Note: In all the tests below we wait a bit before starting the --- client, hoping that by then the server has already started. --- Yes, I know, it's not the best approach. Hopefully it will be enough. -waitTime :: Int -- in microseconds (1e6) -waitTime = 200000 - -test_listen_accept_socketReadS_then_connect_socketWriteD :: Assertion -test_listen_accept_socketReadS_then_connect_socketWriteD = do - let port = ports !! 0 - mvout <- newEmptyMVar - forkIO $ putMVar mvout =<< serveOnceAndRead host1p port - threadDelay waitTime - connectAndWrite host1 port msg1 - out <- takeMVar mvout - B.concat out @=? msg1b - -test_listen_accept_socketWriteD_then_connect_socketReadD :: Assertion -test_listen_accept_socketWriteD_then_connect_socketReadD = do - let port = ports !! 1 - forkIO $ serveOnceAndWrite host1p port msg1 - threadDelay waitTime - out <- connectAndRead host1 port - B.concat out @=? msg1b - -test_safe_serveWriteD :: Assertion -test_safe_serveWriteD = do - let port = ports !! 2 - serveOnceAndWrite' = do - let p = P.fromListS msg1 >-> T'.serveWriteD Nothing host1p port - P.runSafeIO . P.runProxy .P.runEitherK $ p - forkIO serveOnceAndWrite' - threadDelay waitTime - out <- connectAndRead host1 port - B.concat out @=? msg1b - -test_safe_serveReadS :: Assertion -test_safe_serveReadS = do - let port = ports !! 3 - serveOnceAndRead' = do - let p = P.raiseK (T'.serveReadS Nothing 4096 host1p port) >-> P.toListD - (eex,out) <- P.trySafeIO . P.runWriterT . P.runProxy .P.runEitherK $ p - case eex of - Left ex -> E.throw ex - Right () -> return out - mvout <- newEmptyMVar - forkIO $ putMVar mvout =<< serveOnceAndRead' - threadDelay waitTime - connectAndWrite host1 port msg1 - out <- takeMVar mvout - B.concat out @=? msg1b - - -tests :: [Test] -tests = - [ testGroup "TCP" - [ testGroup "{listen*accept,connect}*{socketReadS,socketWriteD}" - [ testCase "test_listen_accept_socketReadS_then_connect_socketWriteD" - test_listen_accept_socketReadS_then_connect_socketWriteD - , testCase "test_listen_accept_socketWriteD_then_connect_socketReadD" - test_listen_accept_socketWriteD_then_connect_socketReadD - ] - ] - , testGroup "TCP.Safe" - [ testGroup "{serve,connect}{WriteD,ReadS}" - [ testCase "test_safe_serveWriteD" test_safe_serveWriteD - , testCase "test_safe_serveReadS" test_safe_serveReadS - ] - ] - ] - -main :: IO () -main = NS.withSocketsDo $ defaultMain tests