pipes-network 0.5.1.0 → 0.6.0
raw patch · 8 files changed
+464/−892 lines, 8 filesdep −HUnitdep −pipes-networkdep −test-frameworkdep ~basedep ~bytestringdep ~network-simplePVP ok
version bump matches the API change (PVP)
Dependencies removed: HUnit, pipes-network, test-framework, test-framework-hunit
Dependency ranges changed: base, bytestring, network-simple, pipes, pipes-safe, transformers
API changes (from Hackage documentation)
- Control.Proxy.TCP: Host :: HostName -> HostPreference
- Control.Proxy.TCP: HostAny :: HostPreference
- Control.Proxy.TCP: HostIPv4 :: HostPreference
- Control.Proxy.TCP: HostIPv6 :: HostPreference
- Control.Proxy.TCP: Timeout :: String -> Timeout
- Control.Proxy.TCP: accept :: Socket -> ((Socket, SockAddr) -> IO b) -> IO b
- Control.Proxy.TCP: acceptFork :: Socket -> ((Socket, SockAddr) -> IO ()) -> IO ThreadId
- Control.Proxy.TCP: connect :: HostName -> ServiceName -> ((Socket, SockAddr) -> IO r) -> IO r
- Control.Proxy.TCP: data HostPreference :: *
- Control.Proxy.TCP: data Timeout
- Control.Proxy.TCP: instance Data Timeout
- Control.Proxy.TCP: instance Eq Timeout
- Control.Proxy.TCP: instance Exception Timeout
- Control.Proxy.TCP: instance Show Timeout
- Control.Proxy.TCP: instance Typeable Timeout
- Control.Proxy.TCP: listen :: HostPreference -> ServiceName -> ((Socket, SockAddr) -> IO r) -> IO r
- Control.Proxy.TCP: nsocketReadS :: Proxy p => Socket -> Int -> Server p Int ByteString IO ()
- Control.Proxy.TCP: nsocketReadTimeoutS :: Proxy p => Int -> Socket -> Int -> Server (EitherP Timeout p) Int ByteString IO ()
- Control.Proxy.TCP: serve :: HostPreference -> ServiceName -> ((Socket, SockAddr) -> IO ()) -> IO ()
- Control.Proxy.TCP: socketReadS :: Proxy p => Int -> Socket -> () -> Producer p ByteString IO ()
- Control.Proxy.TCP: socketReadTimeoutS :: Proxy p => Int -> Int -> Socket -> () -> Producer (EitherP Timeout p) ByteString IO ()
- Control.Proxy.TCP: socketWriteD :: Proxy p => Socket -> x -> p x ByteString x ByteString IO r
- Control.Proxy.TCP: socketWriteTimeoutD :: Proxy p => Int -> Socket -> x -> (EitherP Timeout p) x ByteString x ByteString IO r
- Control.Proxy.TCP: withSocketsDo :: IO a -> IO a
- Control.Proxy.TCP.Safe: Host :: HostName -> HostPreference
- Control.Proxy.TCP.Safe: HostAny :: HostPreference
- Control.Proxy.TCP.Safe: HostIPv4 :: HostPreference
- Control.Proxy.TCP.Safe: HostIPv6 :: HostPreference
- Control.Proxy.TCP.Safe: Timeout :: String -> Timeout
- Control.Proxy.TCP.Safe: accept :: (Proxy p, Monad m) => (forall x. SafeIO x -> m x) -> Socket -> ((Socket, SockAddr) -> ExceptionP p a' a b' b m r) -> ExceptionP p a' a b' b m r
- Control.Proxy.TCP.Safe: acceptFork :: (Proxy p, Monad m) => (forall x. SafeIO x -> m x) -> Socket -> ((Socket, SockAddr) -> IO ()) -> ExceptionP p a' a b' b m ThreadId
- Control.Proxy.TCP.Safe: connect :: (Proxy p, Monad m) => (forall x. SafeIO x -> m x) -> HostName -> ServiceName -> ((Socket, SockAddr) -> ExceptionP p a' a b' b m r) -> ExceptionP p a' a b' b m r
- Control.Proxy.TCP.Safe: connectReadS :: Proxy p => Maybe Int -> Int -> HostName -> ServiceName -> () -> Producer (ExceptionP p) ByteString SafeIO ()
- Control.Proxy.TCP.Safe: connectWriteD :: Proxy p => Maybe Int -> HostName -> ServiceName -> x -> (ExceptionP p) x ByteString x ByteString SafeIO r
- Control.Proxy.TCP.Safe: data HostPreference :: *
- Control.Proxy.TCP.Safe: data Timeout
- Control.Proxy.TCP.Safe: listen :: (Proxy p, Monad m) => (forall x. SafeIO x -> m x) -> HostPreference -> ServiceName -> ((Socket, SockAddr) -> ExceptionP p a' a b' b m r) -> ExceptionP p a' a b' b m r
- Control.Proxy.TCP.Safe: nsocketReadS :: Proxy p => Maybe Int -> Socket -> Int -> Server (ExceptionP p) Int ByteString SafeIO ()
- Control.Proxy.TCP.Safe: serve :: (Proxy p, Monad m) => (forall x. SafeIO x -> m x) -> HostPreference -> ServiceName -> ((Socket, SockAddr) -> IO ()) -> ExceptionP p a' a b' b m r
- Control.Proxy.TCP.Safe: serveReadS :: Proxy p => Maybe Int -> Int -> HostPreference -> ServiceName -> () -> Producer (ExceptionP p) ByteString SafeIO ()
- Control.Proxy.TCP.Safe: serveWriteD :: Proxy p => Maybe Int -> HostPreference -> ServiceName -> x -> (ExceptionP p) x ByteString x ByteString SafeIO r
- Control.Proxy.TCP.Safe: socketReadS :: Proxy p => Maybe Int -> Int -> Socket -> () -> Producer (ExceptionP p) ByteString SafeIO ()
- Control.Proxy.TCP.Safe: socketWriteD :: Proxy p => Maybe Int -> Socket -> x -> (ExceptionP p) x ByteString x ByteString SafeIO r
- Control.Proxy.TCP.Safe: withSocketsDo :: IO a -> IO a
+ Pipes.Network.TCP: fromSocket :: MonadIO m => Socket -> Int -> Producer' ByteString m ()
+ Pipes.Network.TCP: fromSocketN :: MonadIO m => Socket -> Int -> Server' Int ByteString m ()
+ Pipes.Network.TCP: fromSocketTimeout :: MonadIO m => Int -> Socket -> Int -> Producer' ByteString m ()
+ Pipes.Network.TCP: fromSocketTimeoutN :: MonadIO m => Int -> Socket -> Int -> Server' Int ByteString m ()
+ Pipes.Network.TCP: toSocket :: MonadIO m => Socket -> Consumer' ByteString m r
+ Pipes.Network.TCP: toSocketTimeout :: MonadIO m => Int -> Socket -> Consumer' ByteString m r
+ Pipes.Network.TCP.Safe: accept :: (MonadSafe m, Base m ~ IO) => Socket -> ((Socket, SockAddr) -> m r) -> m r
+ Pipes.Network.TCP.Safe: connect :: (MonadSafe m, Base m ~ IO) => HostName -> ServiceName -> ((Socket, SockAddr) -> m r) -> m r
+ Pipes.Network.TCP.Safe: fromConnect :: (MonadSafe m, Base m ~ IO) => Int -> HostName -> ServiceName -> Producer' ByteString m ()
+ Pipes.Network.TCP.Safe: fromServe :: (MonadSafe m, Base m ~ IO) => Int -> HostPreference -> ServiceName -> Producer' ByteString m ()
+ Pipes.Network.TCP.Safe: listen :: (MonadSafe m, Base m ~ IO) => HostPreference -> ServiceName -> ((Socket, SockAddr) -> m r) -> m r
+ Pipes.Network.TCP.Safe: serve :: (MonadSafe m, Base m ~ IO) => HostPreference -> ServiceName -> ((Socket, SockAddr) -> IO ()) -> m r
+ Pipes.Network.TCP.Safe: toConnect :: (MonadSafe m, Base m ~ IO) => HostName -> ServiceName -> Consumer' ByteString m r
+ Pipes.Network.TCP.Safe: toServe :: (MonadSafe m, Base m ~ IO) => HostPreference -> ServiceName -> Consumer' ByteString m r
Files
- NEWS +9/−0
- PEOPLE +1/−0
- pipes-network.cabal +14/−28
- src/Control/Proxy/TCP.hs +0/−250
- src/Control/Proxy/TCP/Safe.hs +0/−468
- src/Pipes/Network/TCP.hs +194/−0
- src/Pipes/Network/TCP/Safe.hs +246/−0
- tests/Simple.hs +0/−146
NEWS view
@@ -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`.
PEOPLE view
@@ -6,3 +6,4 @@ Gabriel Gonzalez Paolo Capriotti Marius Ghita+Kyle Van Berendonck
pipes-network.cabal view
@@ -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
− src/Control/Proxy/TCP.hs
@@ -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
− src/Control/Proxy/TCP/Safe.hs
@@ -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 #-}--
+ src/Pipes/Network/TCP.hs view
@@ -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'.
+ src/Pipes/Network/TCP/Safe.hs view
@@ -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'.+
− tests/Simple.hs
@@ -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