diff --git a/AUTHORS b/AUTHORS new file mode 100644 --- /dev/null +++ b/AUTHORS @@ -0,0 +1,3 @@ +Renzo Carbonara +Gabriel Gonzalez +Paolo Capriotti diff --git a/Control/Pipe/Network.hs b/Control/Pipe/Network.hs deleted file mode 100644 --- a/Control/Pipe/Network.hs +++ /dev/null @@ -1,130 +0,0 @@ -{-# LANGUAGE ScopedTypeVariables #-} -module Control.Pipe.Network ( - Application, - socketReader, - socketWriter, - ServerSettings(..), - runTCPServer, - ClientSettings(..), - runTCPClient, - ) where - -import qualified Network.Socket as NS -import Network.Socket (Socket) -import Network.Socket.ByteString (sendAll, recv) -import Data.ByteString (ByteString) -import qualified Data.ByteString as B -import Control.Concurrent (forkIO) -import qualified Control.Exception as E -import Control.Monad (forever, unless) -import Control.Monad.IO.Class -import Control.Monad.Trans.Class -import Control.Pipe - --- adapted from conduit - --- | Stream data from the socket. -socketReader :: MonadIO m => Socket -> Pipe () ByteString m () -socketReader socket = go - where - go = do - bs <- lift . liftIO $ recv socket 4096 - unless (B.null bs) $ - yield bs >> go - --- | Stream data to the socket. -socketWriter :: MonadIO m => Socket -> Pipe ByteString Void m r -socketWriter socket = forever $ await >>= lift . liftIO . sendAll socket - --- | A simple TCP application. It takes two arguments: the 'Producer' to read --- input data from, and the 'Consumer' to send output data to. -type Application m r = Pipe () ByteString m () - -> Pipe ByteString Void m () - -> IO r - --- | Settings for a TCP server. It takes a port to listen on, and an optional --- hostname to bind to. -data ServerSettings = ServerSettings - { serverPort :: Int - , serverHost :: Maybe String -- ^ 'Nothing' indicates no preference - } - --- | Run an @Application@ with the given settings. This function will create a --- new listening socket, accept connections on it, and spawn a new thread for --- each connection. -runTCPServer :: MonadIO m => ServerSettings -> Application m r -> IO r -runTCPServer (ServerSettings port host) app = E.bracket - (bindPort host port) - NS.sClose - (forever . serve) - where - serve lsocket = do - (socket, _addr) <- NS.accept lsocket - forkIO $ do - E.finally - (app (socketReader socket) (socketWriter socket)) - (NS.sClose socket) - return () - --- | Settings for a TCP client, specifying how to connect to the server. -data ClientSettings = ClientSettings - { clientPort :: Int - , clientHost :: String - } - --- | Run an 'Application' by connecting to the specified server. -runTCPClient :: MonadIO m => ClientSettings -> Application m r -> IO r -runTCPClient (ClientSettings port host) app = E.bracket - (getSocket host port) - NS.sClose - (\s -> app (socketReader s) (socketWriter s)) - --- | Attempt to connect to the given host/port. -getSocket :: String -> Int -> IO NS.Socket -getSocket host' port' = do - let hints = NS.defaultHints { - NS.addrFlags = [NS.AI_ADDRCONFIG] - , NS.addrSocketType = NS.Stream - } - (addr:_) <- NS.getAddrInfo (Just hints) (Just host') (Just $ show port') - E.bracketOnError - (NS.socket (NS.addrFamily addr) - (NS.addrSocketType addr) - (NS.addrProtocol addr)) - NS.sClose - (\sock -> NS.connect sock (NS.addrAddress addr) >> return sock) - --- | Attempt to bind a listening @Socket@ on the given host/port. If no host is --- given, will use the first address available. -bindPort :: Maybe String -> Int -> IO Socket -bindPort host p = do - let hints = NS.defaultHints - { NS.addrFlags = - [ NS.AI_PASSIVE - , NS.AI_NUMERICSERV - , NS.AI_NUMERICHOST - ] - , NS.addrSocketType = NS.Stream - } - port = Just . show $ p - addrs <- NS.getAddrInfo (Just hints) host port - let - tryAddrs (addr1:rest@(_:_)) = E.catch - (theBody addr1) - (\(_ :: E.IOException) -> tryAddrs rest) - tryAddrs (addr1:[]) = theBody addr1 - tryAddrs _ = error "bindPort: addrs is empty" - theBody addr = - E.bracketOnError - (NS.socket - (NS.addrFamily addr) - (NS.addrSocketType addr) - (NS.addrProtocol addr)) - NS.sClose - (\sock -> do - NS.setSocketOption sock NS.ReuseAddr 1 - NS.bindSocket sock (NS.addrAddress addr) - NS.listen sock NS.maxListenQueue - return sock - ) - tryAddrs addrs diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,4 +1,5 @@ -Copyright (c)2012, Paolo Capriotti +Copyright (c) 2012-2013, Renzo Carbonara +Copyright (c) 2012-2012, Paolo Capriotti All rights reserved. @@ -13,7 +14,7 @@ disclaimer in the documentation and/or other materials provided with the distribution. - * Neither the name of Paolo Capriotti nor the names of other + * Neither the name of Renzo Carbonara nor the names of other contributors may be used to endorse or promote products derived from this software without specific prior written permission. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,13 @@ +# pipes-network + +Utilities to deal with sockets using the **pipes** and **pipes-safe** +libraries. + +Check the source or rendered Haddocks for extensive documentation. + +This code is licensed under the terms of the so called **3-clause BSD +license**. Read the file named ``LICENSE`` found in this same directory +for details. + +See the ``AUTHORS`` file to learn about the people involved in this +effort. diff --git a/pipes-network.cabal b/pipes-network.cabal --- a/pipes-network.cabal +++ b/pipes-network.cabal @@ -1,27 +1,65 @@ -Name: pipes-network -Version: 0.0.2 -License: BSD3 -License-file: LICENSE -Author: Paolo Capriotti -Maintainer: p.capriotti@gmail.com -Stability: Experimental -Homepage: https://github.com/pcapriotti/pipes-extra -Category: Control, Enumerator, Network -Build-type: Simple -Synopsis: Utilities to deal with sockets. -Description: Utilities to deal with sockets. -Cabal-version: >=1.8 +name: pipes-network +version: 0.1.0 +license: BSD3 +license-file: LICENSE +copyright: Copyright (c) Renzo Carbonara 2012-2013, Paolo Capriotti 2012-2012. +author: Renzo Carbonara +maintainer: renzocarbonaraλgmail.com +stability: Experimental +tested-with: GHC == 7.4.1 +homepage: https://github.com/k0001/pipes-network +bug-reports: https://github.com/k0001/pipes-network/issues +category: Control, Pipes, Network +build-type: Simple +synopsis: Use network sockets together with the pipes library. +cabal-version: >=1.8 +extra-source-files: README.md AUTHORS +description: + Use network sockets together with the @pipes@ library. + . + This package is organized using the following namespaces: + . + * "Control.Proxy.TCP" exports @pipes@ proxies and functions to deal with TCP + connections. Such proxies don't acquire nor release new resources within a + proxy pipeline. + . + * "Control.Proxy.TCP.Safe" exports @pipes-safe@ proxies and functions to deal + with TCP connections. Such proxies may safely acquire and release resources + within a pipeline, using the facilities provided by the @pipes-safe@ package. -Source-Repository head - Type: git - Location: https://github.com/pcapriotti/pipes-extra +source-repository head + type: git + location: git://github.com/k0001/pipes-network.git -Library - Build-Depends: +library + hs-source-dirs: src + build-depends: base (== 4.*), - transformers (>= 0.2 && < 0.4), - pipes-core (== 0.1.*), - bytestring (== 0.9.*), - network - Exposed-Modules: - Control.Pipe.Network + bytestring (>= 0.9.2.1), + network, + pipes (==3.1.*), + pipes-safe (>=1.0), + transformers (>= 0.2 && < 0.4) + exposed-modules: + Control.Proxy.Network.Internal + Control.Proxy.TCP + Control.Proxy.TCP.Sync + Control.Proxy.TCP.Safe + Control.Proxy.TCP.Safe.Sync + ghc-options: -Wall -fno-warn-unused-do-bind + +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/Network/Internal.hs b/src/Control/Proxy/Network/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Control/Proxy/Network/Internal.hs @@ -0,0 +1,59 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# OPTIONS_HADDOCK hide #-} + +-- | This module doesn't belong to this namespace, but really, I don't +-- know where it belongs. Suggestions welcome. + +-- Some code in this file was adapted from the @network-conduit@ library by +-- Michael Snoyman. Copyright (c) 2011. See its licensing terms (BSD3) at: +-- https://github.com/snoyberg/conduit/blob/master/network-conduit/LICENSE + +module Control.Proxy.Network.Internal ( + HostPreference(..), + hpHostName, + Timeout(..) + ) where + +import qualified Control.Exception as E +import Data.String (IsString (fromString)) +import Data.Typeable (Typeable) +import qualified Network.Socket as NS + +-- | Preferred host to bind. +data HostPreference + = HostAny -- ^Any avaiable host. + | HostIPv4 -- ^Any avaiable IPv4 host. + | HostIPv6 -- ^Any avaiable IPv6 host. + | Host NS.HostName -- ^An explicit host name. + deriving (Eq, Ord, Show, Read) + + +-- | The following special values are recognized: +-- +-- * @*@ means 'HostAny' +-- +-- * @*4@ means 'HostIPv4' +-- +-- * @*6@ means 'HostIPv6' +-- +-- * Any other string is 'Host' +instance IsString HostPreference where + fromString "*" = HostAny + fromString "*4" = HostIPv4 + fromString "*6" = HostIPv6 + fromString s = Host s + + +-- | Extract the 'NS.HostName' from a 'Host' preference, or 'Nothing' otherwise. +hpHostName:: HostPreference -> Maybe NS.HostName +hpHostName (Host s) = Just s +hpHostName _ = Nothing + +-------------------------------------------------------------------------------- + +-- |Exception thrown when a timeout has elapsed. +data Timeout + = Timeout String -- ^Timeouted with an additional explanatory message. + deriving (Eq, Show, Typeable) + +instance E.Exception Timeout where diff --git a/src/Control/Proxy/TCP.hs b/src/Control/Proxy/TCP.hs new file mode 100644 --- /dev/null +++ b/src/Control/Proxy/TCP.hs @@ -0,0 +1,385 @@ +-- | 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". + +-- Some code in this file was adapted from the @network-conduit@ library by +-- Michael Snoyman. Copyright (c) 2011. See its licensing terms (BSD3) at: +-- https://github.com/snoyberg/conduit/blob/master/network-conduit/LICENSE + +module Control.Proxy.TCP ( + -- * Server side + -- $server-side + serve, + serveFork, + -- ** Listening + listen, + -- ** Accepting + accept, + acceptFork, + + -- * Client side + -- $client-side + connect, + + -- * Socket streams + -- $socket-streaming + socketReadS, + nsocketReadS, + socketWriteD, + -- ** Timeouts + -- $socket-streaming-timeout + socketReadTimeoutS, + nsocketReadTimeoutS, + socketWriteTimeoutD, + -- * Low level support + bindSock, + connectSock, + -- * Exports + HostPreference(..), + Timeout(..) + ) where + +import Control.Concurrent (ThreadId, forkIO) +import qualified Control.Exception as E +import Control.Monad +import Control.Monad.Trans.Class +import qualified Control.Proxy as P +import qualified Control.Proxy.Trans.Either as PE +import Control.Proxy.Network.Internal +import qualified Data.ByteString as B +import Data.Monoid +import Data.List (partition) +import qualified Network.Socket as NS +import Network.Socket.ByteString (recv, sendAll) +import System.Timeout (timeout) + +-------------------------------------------------------------------------------- + +-- $client-side +-- +-- The following functions allow you to obtain and use 'NS.Socket's useful to +-- the client side of a TCP connection. +-- +-- 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 with any of the socketReadS, nsocketReadS or socketWriteD +-- > -- proxies explained below. + +-- | 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 and close the socket yourself, then use +-- 'connectSock' and the 'NS.sClose' function from "Network.Socket" instead. +connect + :: NS.HostName -- ^Server hostname. + -> NS.ServiceName -- ^Server service port. + -> ((NS.Socket, NS.SockAddr) -> IO r) + -- ^Computation taking the communication socket + -- and the server address. + -> IO r +connect host port = E.bracket (connectSock host port) (NS.sClose . fst) + +-------------------------------------------------------------------------------- + +-- $server-side +-- +-- The following functions allow you to obtain and use 'NS.Socket's useful to +-- the server side of a TCP connection. +-- +-- Here's how you could run a TCP server that handles in different threads each +-- incoming connection to port @8000@ at address @127.0.0.1@: +-- +-- > listen (Host "127.0.0.1") "8000" $ \(listeningSocket, listeningAddr) -> do +-- > putStrLn $ "Listening for incoming connections at " ++ show listeningAddr +-- > forever . acceptFork listeningSocket $ \(connectionSocket, remoteAddr) -> do +-- > putStrLn $ "Connection established from " ++ show remoteAddr +-- > -- now you may use connectionSocket as you please within this scope, +-- > -- possibly with any of the socketReadS, nsocketReadS or socketWriteD +-- > -- proxies explained below. +-- +-- If you keep reading you'll discover there are different ways to achieve +-- the same, some ways more general than others. The above one was just an +-- example using a pretty general approach, you are encouraged to use simpler +-- approaches such as 'serve' if those suit your needs. + +-- | 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 +-- '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 + :: HostPreference -- ^Preferred host to bind. + -> NS.ServiceName -- ^Service port to bind. + -> ((NS.Socket, NS.SockAddr) -> IO r) + -- ^Computation taking the listening socket and + -- the address it's bound to. + -> IO r +listen hp port = E.bracket listen' (NS.sClose . fst) + where + listen' = do x@(bsock,_) <- bindSock hp port + NS.listen bsock $ max 2048 NS.maxListenQueue + return x + +-- | Start a TCP server that sequentially accepts and uses each incoming +-- connection. +-- +-- Both the listening and connection sockets are closed when done or in case of +-- exceptions. +-- +-- Note: You don't need to use 'listen' nor 'accept' if you use this function. +serve + :: HostPreference -- ^Preferred host to bind. + -> NS.ServiceName -- ^Service port to bind. + -> ((NS.Socket, NS.SockAddr) -> IO r) + -- ^Computation to run once an incoming + -- connection is accepted. Takes the connection socket + -- and remote end address. + -> IO r +serve hp port k = do + listen hp port $ \(lsock,_) -> do + forever $ accept lsock k + +-- | Start a TCP server that accepts incoming connections and uses them +-- concurrently in different threads. +-- +-- The listening and connection sockets are closed when done or in case of +-- exceptions. +-- +-- Note: You don't need to use 'listen' nor 'acceptFork' if you use this +-- function. +serveFork + :: 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. + -> IO () +serveFork hp port k = do + listen hp port $ \(lsock,_) -> do + forever $ acceptFork lsock k + +-- | Accept a single incoming connection and use it. +-- +-- The connection socket is closed when done or in case of exceptions. +accept + :: NS.Socket -- ^Listening and bound socket. + -> ((NS.Socket, NS.SockAddr) -> IO b) + -- ^Computation to run once an incoming + -- connection is accepted. Takes the connection socket + -- and remote end address. + -> IO b +accept lsock k = do + conn@(csock,_) <- NS.accept lsock + E.finally (k conn) (NS.sClose csock) +{-# 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 + :: 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. + -> IO ThreadId +acceptFork lsock f = do + client@(csock,_) <- NS.accept lsock + forkIO $ E.finally (f client) (NS.sClose csock) +{-# INLINABLE 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. +-- +-- If the remote peer closes its side of the connection, this proxy returns. +socketReadS + :: P.Proxy p + => Int -- ^Maximum number of bytes to receive at once. + -> NS.Socket -- ^Connected socket. + -> () -> P.Producer p B.ByteString IO () +socketReadS nbytes sock () = P.runIdentityP loop where + loop = do + bs <- lift $ recv sock nbytes + unless (B.null bs) $ P.respond bs >> loop +{-# 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 + bs <- lift $ recv sock nbytes + unless (B.null bs) $ P.respond bs >>= loop +{-# 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 $ sendAll 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 at once. + -> NS.Socket -- ^Connected socket. + -> () -> P.Producer (PE.EitherP Timeout p) B.ByteString IO () +socketReadTimeoutS wait nbytes sock () = loop where + loop = do + mbs <- lift . timeout wait $ recv sock nbytes + case mbs of + Nothing -> PE.throw ex + Just bs -> unless (B.null bs) $ P.respond bs >> loop + ex = Timeout $ "recv: " <> 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 + mbs <- lift . timeout wait $ recv sock nbytes + case mbs of + Nothing -> PE.throw ex + Just bs -> unless (B.null bs) $ P.respond bs >>= loop + ex = Timeout $ "recv: " <> 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 + mbs <- lift . timeout wait $ sendAll sock a + case mbs of + Nothing -> PE.throw ex + Just () -> P.respond a >>= loop + ex = Timeout $ "recv: " <> show wait <> " microseconds." +{-# INLINABLE socketWriteTimeoutD #-} + +-------------------------------------------------------------------------------- + +-- | Obtain a 'NS.Socket' connected to the given host and TCP service port. +-- +-- The obtained 'NS.Socket' should be closed manually using 'NS.sClose' when +-- it's not needed anymore, otherwise you risk having the socket open for much +-- longer than needed. +-- +-- Prefer to use 'connect' if you will be using the socket within a limited +-- scope and would like it to be closed immediately after its usage or in case +-- of exceptions. +connectSock :: NS.HostName -> NS.ServiceName -> IO (NS.Socket, NS.SockAddr) +connectSock host port = do + (addr:_) <- NS.getAddrInfo (Just hints) (Just host) (Just port) + E.bracketOnError (newSocket addr) NS.sClose $ \sock -> do + let sockAddr = NS.addrAddress addr + NS.connect sock sockAddr + return (sock, sockAddr) + where + hints = NS.defaultHints { NS.addrFlags = [NS.AI_ADDRCONFIG] + , NS.addrSocketType = NS.Stream } + +-- | Obtain a 'NS.Socket' bound to the given host name and TCP service port. +-- +-- The obtained 'NS.Socket' should be closed manually using 'NS.sClose' when +-- it's not needed anymore. +-- +-- Prefer to use 'listen' if you will be listening on this socket and using it +-- within a limited scope, and would like it to be closed immediately after its +-- usage or in case of exceptions. +bindSock :: HostPreference -> NS.ServiceName -> IO (NS.Socket, NS.SockAddr) +bindSock hp port = do + addrs <- NS.getAddrInfo (Just hints) (hpHostName hp) (Just port) + let addrs' = case hp of + HostIPv4 -> prioritize isIPv4addr addrs + HostIPv6 -> prioritize isIPv6addr addrs + _ -> addrs + tryAddrs addrs' + where + hints = NS.defaultHints { NS.addrFlags = [NS.AI_PASSIVE] + , NS.addrSocketType = NS.Stream } + + tryAddrs [] = error "listen: no addresses available" + tryAddrs [x] = useAddr x + tryAddrs (x:xs) = E.catch (useAddr x) + (\e -> let _ = e :: E.IOException in tryAddrs xs) + + useAddr addr = E.bracketOnError (newSocket addr) NS.sClose $ \sock -> do + let sockAddr = NS.addrAddress addr + NS.setSocketOption sock NS.NoDelay 1 + NS.setSocketOption sock NS.ReuseAddr 1 + NS.bindSocket sock sockAddr + return (sock, sockAddr) + +-------------------------------------------------------------------------------- + +-- Misc + +newSocket :: NS.AddrInfo -> IO NS.Socket +newSocket addr = NS.socket (NS.addrFamily addr) + (NS.addrSocketType addr) + (NS.addrProtocol addr) + +isIPv4addr, isIPv6addr :: NS.AddrInfo -> Bool +isIPv4addr x = NS.addrFamily x == NS.AF_INET +isIPv6addr x = NS.addrFamily x == NS.AF_INET6 + +-- | Move the elements that match the predicate closer to the head of the list. +-- Preserve relative order. +prioritize :: (a -> Bool) -> [a] -> [a] +prioritize p = uncurry (++) . partition p diff --git a/src/Control/Proxy/TCP/Safe.hs b/src/Control/Proxy/TCP/Safe.hs new file mode 100644 --- /dev/null +++ b/src/Control/Proxy/TCP/Safe.hs @@ -0,0 +1,435 @@ +{-# 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 ( + -- * Server side + -- $server-side + serve, + serveFork, + -- ** Listening + listen, + -- ** Accepting + accept, + acceptFork, + -- ** Streaming + -- $server-streaming + serveReadS, + serveWriteD, + + -- * Client side + -- $client-side + connect, + -- ** Streaming + -- $client-streaming + connectReadS, + connectWriteD, + + -- * Socket streams + -- $socket-streaming + socketReadS, + nsocketReadS, + socketWriteD, + + -- * Exports + HostPreference(..), + Timeout(..) + ) where + +import Control.Concurrent (forkIO, ThreadId) +import qualified Control.Exception as E +import Control.Monad +import qualified Control.Proxy as P +import Control.Proxy.Network.Internal +import qualified Control.Proxy.TCP as T +import qualified Control.Proxy.Safe as P +import qualified Data.ByteString as B +import Data.Monoid +import qualified Network.Socket as NS +import Network.Socket.ByteString (sendAll, recv) +import System.Timeout (timeout) + +-------------------------------------------------------------------------------- + +-- $client-side +-- +-- The following functions allow you to obtain and use 'NS.Socket's useful to +-- the client side of a TCP connection. +-- +-- Here's how you could run a TCP client: +-- +-- > connect id "www.example.org" "80" $ \(connectionSocket, remoteAddr) -> do +-- > tryIO . putStrLn $ "Connection established to " ++ show remoteAddr +-- > -- now you may use connectionSocket as you please within this scope, +-- > -- possibly with any of the socketReadS, nsocketReadS or socketWriteD +-- > -- 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 +-- 'T.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 (T.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: +-- +-- >>> 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 at once. + -> 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 () +connectWriteD mwait hp port x = do + connect id hp port $ \(csock,_) -> + socketWriteD mwait csock x + +-------------------------------------------------------------------------------- + +-- $server-side +-- +-- The following functions allow you to obtain and use 'NS.Socket's useful to +-- the server side of a TCP connection. +-- +-- Here's how you could run a TCP server that handles in different threads each +-- incoming connection to port @8000@ at address @127.0.0.1@: +-- +-- > listen id (Host "127.0.0.1") "8000" $ \(listeningSocket, listeningAddr) -> do +-- > tryIO . putStrLn $ "Listening for incoming connections at " ++ show listeningAddr +-- > forever . acceptFork id listeningSocket $ \(connectionSocket, remoteAddr) -> do +-- > putStrLn $ "Connection established from " ++ show remoteAddr +-- > -- now you may use connectionSocket as you please within this scope, +-- > -- possibly with any of the socketReadS, nsocketReadS or socketWriteD +-- > -- proxies explained below. +-- +-- If you keep reading you'll discover there are different ways to achieve +-- the same, some ways more general than others. The above one was just an +-- example using a pretty general approach, you are encouraged to use simpler +-- approaches such as 'serve' or 'serveReadS' if those suit your needs. + +-- | 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 +-- 'T.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. + -> 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,_) <- T.bindSock hp port + NS.listen bsock $ max 2048 NS.maxListenQueue + return x + +-- | Start a TCP server that sequentially accepts and uses each incoming +-- connection. +-- +-- Both the listening and connection sockets are closed when done or in case of +-- exceptions. +-- +-- Note: You don't need to use 'listen' nor 'accept' if you use this function. +serve + :: (P.Proxy p, Monad m) + => (forall x. P.SafeIO x -> m x) -- ^Monad morphism. + -> 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 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 +serve morph hp port k = do + listen morph hp port $ \(lsock,_) -> do + forever $ accept morph lsock k + +-- | Start a TCP server that accepts incoming connections and uses them +-- concurrently in different threads. +-- +-- The listening and connection sockets are closed when done or in case of +-- exceptions. +-- +-- Note: You don't need to use 'listen' nor 'acceptFork' if you use this +-- function. +serveFork + :: (P.Proxy p, Monad m) + => (forall x. P.SafeIO x -> m x) -- ^Monad morphism. + -> 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 +serveFork morph hp port k = do + listen morph hp port $ \(lsock,_) -> do + forever $ acceptFork morph lsock k + +-- | 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 f = P.hoist morph . P.tryIO $ do + client@(csock,_) <- NS.accept lsock + forkIO $ E.finally (f client) (NS.sClose csock) +{-# 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. +-- +-- If the remote peer closes its side of the connection, this proxy returns. +-- +-- 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: +-- +-- >>> :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 at once. + -> 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). + -> HostPreference -- ^Preferred host to bind. + -> NS.ServiceName -- ^Service port to bind. + -> x -> (P.ExceptionP p) x B.ByteString x B.ByteString P.SafeIO () +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. +-- +-- If the remote peer closes its side of the connection, this proxy returns. +socketReadS + :: P.Proxy p + => Maybe Int -- ^Optional timeout in microseconds (1/10^6 seconds). + -> Int -- ^Maximum number of bytes to receive at once. + -> NS.Socket -- ^Connected socket. + -> () -> P.Producer (P.ExceptionP p) B.ByteString P.SafeIO () +socketReadS Nothing nbytes sock () = loop where + loop = do + bs <- P.tryIO $ recv sock nbytes + unless (B.null bs) $ P.respond bs >> loop +socketReadS (Just wait) nbytes sock () = loop where + loop = do + mbs <- P.tryIO . timeout wait $ recv sock nbytes + case mbs of + Nothing -> P.throw ex + Just bs -> unless (B.null bs) $ P.respond bs >> loop + ex = Timeout $ "recv: " <> 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 + bs <- P.tryIO $ recv sock nbytes + unless (B.null bs) $ P.respond bs >>= loop +nsocketReadS (Just wait) sock = loop where + loop nbytes = do + mbs <- P.tryIO . timeout wait $ recv sock nbytes + case mbs of + Nothing -> P.throw ex + Just bs -> unless (B.null bs) $ P.respond bs >>= loop + ex = Timeout $ "recv: " <> 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 $ sendAll sock a + P.respond a >>= loop +socketWriteD (Just wait) sock = loop where + loop x = do + a <- P.request x + m <- P.tryIO . timeout wait $ sendAll sock a + case m of + Nothing -> P.throw ex + Just () -> P.respond a >>= loop + ex = Timeout $ "sendAll: " <> show wait <> " microseconds." +{-# INLINABLE socketWriteD #-} + + + diff --git a/src/Control/Proxy/TCP/Safe/Sync.hs b/src/Control/Proxy/TCP/Safe/Sync.hs new file mode 100644 --- /dev/null +++ b/src/Control/Proxy/TCP/Safe/Sync.hs @@ -0,0 +1,121 @@ +-- | This module exports 'P.Proxy's that allow implementing synchronous RPC-like +-- communication with a remote end by using a simple protocol on their +-- downstream interface. +-- +-- As opposed to the similar proxies found in "Control.Proxy.TCP.Sync", +-- these use the exception handling facilities provided by 'P.ExceptionP'. +-- +-- You may prefer the more general proxies from +-- "Control.Proxy.TCP.Safe". + +module Control.Proxy.TCP.Safe.Sync ( + -- * Socket proxies + socketSyncServer, + socketSyncProxy, + -- * RPC support + syncDelimit, + -- * Protocol + Request(..), + Response(..), + ) where + +import Control.Monad +import qualified Control.Proxy as P +import Control.Proxy.TCP.Sync + (Request(..), Response(..), syncDelimit) +import Control.Proxy.Network.Internal +import qualified Control.Proxy.Safe as P +import qualified Data.ByteString as B +import Data.Monoid +import qualified Network.Socket as NS +import Network.Socket.ByteString (recv, sendAll) +import System.Timeout (timeout) + + +-- | 'P.Server' able to send and receive bytes through a 'NS.Socket'. +-- +-- If downstream requests @'Send' bytes@, then such @bytes@ are sent to the +-- remote end and then this proxy responds 'Sent' downstream. +-- +-- If downstream requests @'Receive' num@, then at most @num@ bytes are received +-- from the remote end. This proxy then responds downstream such received +-- bytes as @'Received' bytes@. Less than the specified maximum number of bytes +-- might be received at once. +-- +-- If an optional timeout is given and interactions with the remote end +-- take more time that such timeout, then throw a 'Timeout' exception in +-- the 'P.ExceptionP' proxy transformer. +-- +-- If the remote peer closes its side of the connection, this proxy returns. +socketSyncServer + :: P.Proxy p + => Maybe Int -- ^Optional timeout in microseconds (1/10^6 seconds). + -> NS.Socket -- ^Connected socket. + -> Request B.ByteString + -> P.Server (P.ExceptionP p) (Request B.ByteString) Response P.SafeIO() +socketSyncServer Nothing sock = loop where + loop (Send bs) = do + P.tryIO $ sendAll sock bs + P.respond Sent >>= loop + loop (Receive nbytes) = do + bs <- P.tryIO $ recv sock nbytes + unless (B.null bs) $ P.respond (Received bs) >>= loop +socketSyncServer (Just wait) sock = loop where + loop (Send bs) = do + m <- P.tryIO . timeout wait $ sendAll sock bs + case m of + Nothing -> P.throw $ ex "sendAll" + Just () -> P.respond Sent >>= loop + loop (Receive nbytes) = do + mbs <- P.tryIO . timeout wait $ recv sock nbytes + case mbs of + Nothing -> P.throw $ ex "recv" + Just bs -> unless (B.null bs) $ P.respond (Received bs) >>= loop + ex s = Timeout $ s <> ": " <> show wait <> " microseconds." +{-# INLINABLE socketSyncServer #-} + +-- | 'P.Proxy' able to send and receive bytes through a 'NS.Socket'. +-- +-- If downstream requests @'Send' a'@, then such @a'@ request is forwarded +-- upstream, which in return responds a 'B.ByteString' that this proxy sends to +-- the remote end. After sending to the remote end, this proxy responds 'Sent' +-- downstream. +-- +-- If downstream requests @'Receive' num@, then at most @num@ bytes are received +-- from the remote end. This proxy then responds downstream such received +-- bytes as @'Received' bytes@. Less than the specified maximum number of bytes +-- might be received at once. +-- +-- If an optional timeout is given and interactions with the remote end +-- take more time that such timeout, then throw a 'Timeout' exception in +-- the 'P.ExceptionP' proxy transformer. +-- +-- If the remote peer closes its side of the connection, this proxy returns. +socketSyncProxy + :: P.Proxy p + => Maybe Int -- ^Optional timeout in microseconds (1/10^6 seconds). + -> NS.Socket -- ^Connected socket. + -> Request a' + -> (P.ExceptionP p) a' B.ByteString (Request a') Response P.SafeIO () +socketSyncProxy Nothing sock = loop where + loop (Send a') = do + P.request a' >>= P.tryIO . sendAll sock + P.respond Sent >>= loop + loop (Receive nbytes) = do + bs <- P.tryIO $ recv sock nbytes + unless (B.null bs) $ P.respond (Received bs) >>= loop +socketSyncProxy (Just wait) sock = loop where + loop (Send a') = do + bs <- P.request a' + m <- P.tryIO . timeout wait $ sendAll sock bs + case m of + Nothing -> P.throw $ ex "sendAll" + Just () -> P.respond Sent >>= loop + loop (Receive nbytes) = do + mbs <- P.tryIO . timeout wait $ recv sock nbytes + case mbs of + Nothing -> P.throw $ ex "recv" + Just bs -> unless (B.null bs) $ P.respond (Received bs) >>= loop + ex s = Timeout $ s <> ": " <> show wait <> " microseconds." +{-# INLINABLE socketSyncProxy #-} + diff --git a/src/Control/Proxy/TCP/Sync.hs b/src/Control/Proxy/TCP/Sync.hs new file mode 100644 --- /dev/null +++ b/src/Control/Proxy/TCP/Sync.hs @@ -0,0 +1,203 @@ +{-# LANGUAGE OverloadedStrings #-} + +-- | This module exports 'P.Proxy's that allow implementing synchronous RPC-like +-- communication with a remote end by using a simple protocol on their +-- downstream interface. +-- +-- As opposed to the similar proxies found in +-- "Control.Proxy.TCP.Safe.Sync", these don't use the exception handling +-- facilities provided by 'P.ExceptionP'. +-- +-- You may prefer the more general and efficient proxies from +-- "Control.Proxy.TCP". + +module Control.Proxy.TCP.Sync ( + -- * Socket proxies + socketSyncServer, + socketSyncProxy, + -- ** Timeouts + -- $timeouts + socketSyncServerTimeout, + socketSyncProxyTimeout, + -- * RPC support + syncDelimit, + -- * Protocol types + Request(..), + Response(..), + ) where + +import Control.Monad +import Control.Monad.Trans.Class +import qualified Control.Proxy as P +import Control.Proxy.Network.Internal +import qualified Control.Proxy.Trans.Either as PE +import qualified Data.ByteString.Char8 as B +import Data.Monoid +import qualified Network.Socket as NS +import Network.Socket.ByteString (recv, sendAll) +import System.Timeout (timeout) + + +-- | A request made to one of the @socketSync*@ proxies. +data Request t = Send t | Receive Int + deriving (Eq, Read, Show) + +-- | A response received from one of the @socketSync*@ proxies. +data Response = Sent | Received B.ByteString + deriving (Eq, Read, Show) + +-------------------------------------------------------------------------------- + +-- | 'P.Server' able to send and receive bytes through a 'NS.Socket'. +-- +-- If downstream requests @'Send' bytes@, then such @bytes@ are sent to the +-- remote end and then this proxy responds 'Sent' downstream. +-- +-- If downstream requests @'Receive' num@, then at most @num@ bytes are received +-- from the remote end. This proxy then responds downstream such received +-- bytes as @'Received' bytes@. Less than the specified maximum number of bytes +-- might be received at once. +-- +-- If the remote peer closes its side of the connection, this proxy returns. +socketSyncServer + :: P.Proxy p + => NS.Socket -- ^Connected socket. + -> Request B.ByteString + -> P.Server p (Request B.ByteString) Response IO () +socketSyncServer sock = P.runIdentityK loop where + loop (Send bs) = do + lift $ sendAll sock bs + P.respond Sent >>= loop + loop (Receive nbytes) = do + bs <- lift $ recv sock nbytes + unless (B.null bs) $ P.respond (Received bs) >>= loop +{-# INLINABLE socketSyncServer #-} + +-- | 'P.Proxy' able to send and receive bytes through a 'NS.Socket'. +-- +-- If downstream requests @'Send' a'@, then such @a'@ request is forwarded +-- upstream, which in return responds a 'B.ByteString' that this proxy sends to +-- the remote end. After sending to the remote end, this proxy responds 'Sent' +-- downstream. +-- +-- If downstream requests @'Receive' num@, then at most @num@ bytes are received +-- from the remote end. This proxy then responds downstream such received +-- bytes as @'Received' bytes@. Less than the specified maximum number of bytes +-- might be received at once. +-- +-- If the remote peer closes its side of the connection, this proxy returns. +socketSyncProxy + :: P.Proxy p + => NS.Socket -- ^Connected socket. + -> Request a' + -> p a' B.ByteString (Request a') Response IO () +socketSyncProxy sock = P.runIdentityK loop where + loop (Send a') = do + P.request a' >>= lift . sendAll sock + P.respond Sent >>= loop + loop (Receive nbytes) = do + bs <- lift $ recv sock nbytes + unless (B.null bs) $ P.respond (Received bs) >>= loop +{-# INLINABLE socketSyncProxy #-} + +-------------------------------------------------------------------------------- + +-- $timeouts +-- +-- These proxies behave like the similarly named ones above, except support for +-- timing out the interaction with the remote end is added. + +-- | Like 'socketSyncServer', except it throws a 'Timeout' exception in the +-- 'PE.EitherP' proxy transformer if interacting with the remote end takes +-- more time than specified. +socketSyncServerTimeout + :: P.Proxy p + => Int -- ^Timeout in microseconds (1/10^6 seconds). + -> NS.Socket -- ^Connected socket. + -> Request B.ByteString + -> P.Server (PE.EitherP Timeout p) (Request B.ByteString) Response IO () +socketSyncServerTimeout wait sock = loop where + loop (Send bs) = do + m <- lift . timeout wait $ sendAll sock bs + case m of + Nothing -> PE.throw $ ex "sendAll" + Just () -> P.respond Sent >>= loop + loop (Receive nbytes) = do + mbs <- lift . timeout wait $ recv sock nbytes + case mbs of + Nothing -> PE.throw $ ex "recv" + Just bs -> unless (B.null bs) $ P.respond (Received bs) >>= loop + ex s = Timeout $ s <> ": " <> show wait <> " microseconds." +{-# INLINABLE socketSyncServerTimeout #-} + +-- | Like 'socketSyncProxy', except it throws a 'Timeout' exception in the +-- 'PE.EitherP' proxy transformer if interacting with the remote end takes +-- more time than specified. +socketSyncProxyTimeout + :: P.Proxy p + => Int -- ^Timeout in microseconds (1/10^6 seconds). + -> NS.Socket -- ^Connected socket. + -> Request a' + -> (PE.EitherP Timeout p) a' B.ByteString (Request a') Response IO () +socketSyncProxyTimeout wait sock = loop where + loop (Send a') = do + bs <- P.request a' + m <- lift . timeout wait $ sendAll sock bs + case m of + Nothing -> PE.throw $ ex "sendAll" + Just () -> P.respond Sent >>= loop + loop (Receive nbytes) = do + mbs <- lift . timeout wait $ recv sock nbytes + case mbs of + Nothing -> PE.throw $ ex "recv" + Just bs -> unless (B.null bs) $ P.respond (Received bs) >>= loop + ex s = Timeout $ s <> ": " <> show wait <> " microseconds." +{-# INLINABLE socketSyncProxyTimeout #-} + +-------------------------------------------------------------------------------- + +-- | When used together with one of the @socketSync*@ proxies upstream, this +-- proxy sends a single 'B.ByteString' to the remote end and then repeatedly +-- receives bytes from the remote end until the given delimiter is found. +-- Finally, a single 'B.ByteString' up to the given delimiter (inclusive) is +-- sent downstream and then the whole process is repeated. +-- +-- This proxy works cooperatively with any @socketSync*@ proxy immediately +-- upstream, so read their documentation to understand the purpose of the +-- @b'@ value received from downstream. +-- +-- For example, if you'd like to convert a 'NS.Socket' into an synchronous +-- line-oriented RPC client implemented as a 'P.Server' in which RPC calls are +-- received via the downstream interface and RPC responses are sent downstream, +-- then you could use this proxy as: +-- +-- > socketSyncServer ... >-> syncDelimit 4096 "\r\n" +-- +-- Otherwise, if you'd like to convert a 'NS.Socket' into an synchronous +-- line-oriented RPC client implemented as a 'P.Proxy' in which RPC calls are +-- received via the upstream interface and RPC responses are sent downstream, +-- then you could use this proxy as: +-- +-- > socketSyncProxy ... >-> syncDelimit 4096 "\r\n" +syncDelimit + :: (Monad m, P.Proxy p) + => Int -- ^Maximum number of bytes to receive at once. + -> B.ByteString -- ^Delimiting bytes. + -> b'-> p (Request b') Response b' B.ByteString m r +syncDelimit nbytes delim b' = + -- XXX this implementation might be inefficient. + P.runIdentityP $ use =<< more mempty (Send b') + where + more buf req = do + a <- P.request req + case a of + Received bs -> return (buf <> bs) + Sent -> more buf (Receive nbytes) + use buf = do + let (pre,suf) = B.breakSubstring delim buf + case B.length suf of + 0 -> use =<< more buf (Receive nbytes) + _ -> do b'2 <- P.respond (pre <> delim) + use =<< more (B.drop (B.length delim) suf) (Send b'2) +{-# INLINABLE syncDelimit #-} + diff --git a/tests/Simple.hs b/tests/Simple.hs new file mode 100644 --- /dev/null +++ b/tests/Simple.hs @@ -0,0 +1,146 @@ +{-# 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