pipes-network 0.2.0.0 → 0.3.0.0

raw patch · 7 files changed

+202/−109 lines, 7 filesdep ~pipes-safePVP ok

version bump matches the API change (PVP)

Dependency ranges changed: pipes-safe

API changes (from Hackage documentation)

Files

• NEWS +12/−0
• pipes-network.cabal +2/−2
• src/Control/Proxy/Network/Internal.hs +33/−1
• src/Control/Proxy/TCP.hs +41/−26
• src/Control/Proxy/TCP/Safe.hs +44/−24
• src/Control/Proxy/TCP/Safe/Sync.hs +36/−28
• src/Control/Proxy/TCP/Sync.hs +34/−28

@@ -1,3 +1,14 @@+# Version 0.3.0.0++* Quietly stop writing or reading bytes from a TCP socket if a+  “Broken Pipe” error happens, indicating that the remote end+  already closed the connection. Previously, a `ResourceVanished`+  exception was thrown.++* All the `*Write*D` proxies now return `()` if the remote end+  closed the connection.++ # Version 0.2.0.0  * Depend on network-simple 0.2@@ -12,6 +23,7 @@ * Split many of the non-pipes-related TCP utilities to the own   `network-simple` package. * Depend on `network-simple` and re-export its functions.+  # Version 0.1.0.1 

@@ -1,5 +1,5 @@ name:               pipes-network-version:            0.2.0.0+version:            0.3.0.0 license:            BSD3 license-file:       LICENSE copyright:          Copyright (c) Renzo Carbonara 2012-2013, Paolo Capriotti 2012-2012.@@ -42,7 +42,7 @@         network,         network-simple (>=0.2 && <0.3),         pipes          (>=3.1 && <3.3),-        pipes-safe     (>=1.0),+        pipes-safe     (>=1.1 && <1.2),         transformers   (>=0.2 && <0.4)     exposed-modules:         Control.Proxy.TCP

@@ -7,11 +7,17 @@ -- There's a @data-timeout@ package, maybe we should depend on that.  module Control.Proxy.Network.Internal (-  Timeout(..)+    Timeout(..)+  , recv+  , send   ) where +import qualified Data.ByteString               as B import qualified Control.Exception             as E import           Data.Typeable                 (Typeable)+import qualified GHC.IO.Exception              as Eg+import qualified Network.Socket                as NS+import qualified Network.Socket.ByteString   -- |Exception thrown when a timeout has elapsed.@@ -20,3 +26,29 @@   deriving (Eq, Show, Typeable)  instance E.Exception Timeout where+++--------------------------------------------------------------------------------++-- | Read up to a limited number of bytes from a socket.+--+-- Returns `Nothing` if the remote end closed the connection (“Broken Pipe”) or+-- EOF was reached.+recv :: NS.Socket -> Int -> IO (Maybe B.ByteString)+recv sock nbytes =+    E.handle (\Eg.IOError{Eg.ioe_type=Eg.ResourceVanished} -> return Nothing)+             (do bs <- Network.Socket.ByteString.recv sock nbytes+                 if B.null bs then return Nothing+                              else return (Just bs))+{-# INLINE recv #-}++-- | Writes the given bytes to the socket.+--+-- Returns `False` if the remote end closed the connection (“Broken Pipe”),+-- otherwise `True`.+send :: NS.Socket -> B.ByteString -> IO Bool+send sock bs =+    E.handle (\Eg.IOError{Eg.ioe_type=Eg.ResourceVanished} -> return False)+             (do Network.Socket.ByteString.sendAll sock bs+                 return True)+{-# INLINE send #-}

@@ -48,7 +48,6 @@ import qualified Data.ByteString                as B import           Data.Monoid import qualified Network.Socket                 as NS-import           Network.Socket.ByteString      (recv, sendAll) import qualified Network.Simple.TCP             as S import           System.Timeout                 (timeout) @@ -94,13 +93,18 @@ -- 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.+  => 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-      bs <- lift $ recv sock nbytes-      unless (B.null bs) $ P.respond bs >> loop+      mbs <- lift (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@@ -111,23 +115,27 @@   -> 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+      mbs <- lift (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.+--+-- If the remote peer closes its side of the connection, this proxy returns. socketWriteD   :: P.Proxy p   => NS.Socket          -- ^Connected socket.-  -> x -> p x B.ByteString x B.ByteString IO r+  -> x -> p x B.ByteString x B.ByteString IO () socketWriteD sock = P.runIdentityK loop where     loop x = do       a <- P.request x-      lift $ sendAll sock a-      P.respond a >>= loop+      ok <- lift (send sock a)+      when ok (P.respond a >>= loop) {-# INLINABLE socketWriteD #-}  --------------------------------------------------------------------------------@@ -143,16 +151,20 @@ socketReadTimeoutS   :: P.Proxy p   => Int                -- ^Timeout in microseconds (1/10^6 seconds).-  -> Int                -- ^Maximum number of bytes to receive at once.+  -> 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-      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."+      mmbs <- lift (timeout wait (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@@ -165,11 +177,12 @@   -> 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."+      mmbs <- lift (timeout wait (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@@ -179,14 +192,16 @@   :: 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+  -> x -> (PE.EitherP Timeout p) x B.ByteString x B.ByteString IO () 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."+      mt <- lift (timeout wait (send sock a))+      case mt of+        Just True  -> P.respond a >>= loop+        Just False -> return ()+        Nothing    -> PE.throw ex+    ex = Timeout $ "socketWriteTimeoutD: " <> show wait <> " microseconds." {-# INLINABLE socketWriteTimeoutD #-}+ 

@@ -51,7 +51,6 @@ import qualified Data.ByteString                as B import           Data.Monoid import qualified Network.Socket                 as NS-import           Network.Socket.ByteString      (sendAll, recv) import qualified Network.Simple.TCP             as S import           System.Timeout                 (timeout) @@ -112,13 +111,16 @@ -- -- 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:+-- 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 at once.+  -> 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 ()@@ -137,6 +139,8 @@ -- -- The connection socket is closed when done or in case of exceptions. --+-- If the remote peer closes its side of the connection, this proxy returns.+-- -- Using this proxy you can write straightforward code like the following, which -- greets a TCP client listening locally at port 9000: --@@ -285,14 +289,18 @@ -- 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:+-- 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 at once.+  -> 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 ()@@ -351,19 +359,25 @@ socketReadS   :: P.Proxy p   => Maybe Int          -- ^Optional timeout in microseconds (1/10^6 seconds).-  -> Int                -- ^Maximum number of bytes to receive at once.+  -> 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-      bs <- P.tryIO $ recv sock nbytes-      unless (B.null bs) $ P.respond bs >> loop+      mbs <- P.tryIO (recv sock nbytes)+      case mbs of+        Just bs -> P.respond bs >> loop+        Nothing -> return () 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+      mmbs <- P.tryIO (timeout wait (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 #-} @@ -376,14 +390,17 @@   -> 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+      mbs <- P.tryIO (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 $ recv sock nbytes+      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+        Just (Just bs) -> P.respond bs >>= loop+        Just Nothing   -> return ()+        Nothing        -> P.throw ex     ex = Timeout $ "nsocketReadS: " <> show wait <> " microseconds." {-# INLINABLE nsocketReadS #-} @@ -395,23 +412,26 @@ -- 'P.ExceptionP' proxy transformer. -- -- Requests from downstream are forwarded upstream.+--+-- If the remote peer closes its side of the connection, this proxy returns. 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+  -> x -> (P.ExceptionP p) x B.ByteString x B.ByteString P.SafeIO () socketWriteD Nothing sock = loop where     loop x = do       a <- P.request x-      P.tryIO $ sendAll sock a-      P.respond a >>= loop+      ok <- P.tryIO (send sock a)+      when ok (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+      mok <- P.tryIO (timeout wait (send sock a))+      case mok of+        Just True  -> P.respond a >>= loop+        Just False -> return ()+        Nothing    -> P.throw ex     ex = Timeout $ "socketWriteD: " <> show wait <> " microseconds." {-# INLINABLE socketWriteD #-} 

@@ -28,7 +28,6 @@ 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)  @@ -52,26 +51,30 @@   => 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()+  -> 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+        ok <- P.tryIO (send sock bs)+        when ok (P.respond Sent >>= loop)     loop (Receive nbytes) = do-        bs <- P.tryIO $ recv sock nbytes-        unless (B.null bs) $ P.respond (Received bs) >>= loop+        mbs <- P.tryIO (recv sock nbytes)+        case mbs of+          Just bs -> P.respond (Received bs) >>= loop+          Nothing -> return () 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+        mok <- P.tryIO (timeout wait (send sock bs))+        case mok of+          Just True  -> P.respond Sent >>= loop+          Just False -> return ()+          Nothing    -> P.throw ex     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."+        mmbs <- P.tryIO (timeout wait (recv sock nbytes))+        case mmbs of+          Just (Just bs) -> P.respond (Received bs) >>= loop+          Just Nothing   -> return ()+          Nothing        -> P.throw ex+    ex = Timeout $ "socketSyncServer: " <> show wait <> " microseconds." {-# INLINABLE socketSyncServer #-}  -- | 'P.Proxy' able to send and receive bytes through a 'NS.Socket'.@@ -99,23 +102,28 @@   -> (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+        ok <- P.tryIO . send sock =<< P.request a'+        when ok (P.respond Sent >>= loop)     loop (Receive nbytes) = do-        bs <- P.tryIO $ recv sock nbytes-        unless (B.null bs) $ P.respond (Received bs) >>= loop+        mbs <- P.tryIO (recv sock nbytes)+        case mbs of+          Just bs -> P.respond (Received bs) >>= loop+          Nothing -> return () 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+        mok <- P.tryIO (timeout wait (send sock bs))+        case mok of+          Just True  -> P.respond Sent >>= loop+          Just False -> return ()+          Nothing    -> P.throw ex     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."+        mmbs <- P.tryIO (timeout wait (recv sock nbytes))+        case mmbs of+          Just (Just bs) -> P.respond (Received bs) >>= loop+          Just Nothing   -> return ()+          Nothing        -> P.throw ex+    ex = Timeout $ "socketSyncProxy: " <> show wait <> " microseconds." {-# INLINABLE socketSyncProxy #-}+ 

@@ -34,7 +34,6 @@ 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)  @@ -66,11 +65,13 @@   -> 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+        ok <- lift (send sock bs)+        when ok (P.respond Sent >>= loop)     loop (Receive nbytes) = do-        bs <- lift $ recv sock nbytes-        unless (B.null bs) $ P.respond (Received bs) >>= loop+        mbs <- lift (recv sock nbytes)+        case mbs of+          Just bs -> P.respond (Received bs) >>= loop+          Nothing -> return () {-# INLINABLE socketSyncServer #-}  -- | 'P.Proxy' able to send and receive bytes through a 'NS.Socket'.@@ -93,11 +94,13 @@   -> 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+        ok <- lift . send sock =<< P.request a'+        when ok (P.respond Sent >>= loop)     loop (Receive nbytes) = do-        bs <- lift $ recv sock nbytes-        unless (B.null bs) $ P.respond (Received bs) >>= loop+        mbs <- lift (recv sock nbytes)+        case mbs of+          Just bs -> P.respond (Received bs) >>= loop+          Nothing -> return () {-# INLINABLE socketSyncProxy #-}  --------------------------------------------------------------------------------@@ -118,16 +121,18 @@   -> 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+        mok <- lift (timeout wait (send sock bs))+        case mok of+          Just True  -> P.respond Sent >>= loop+          Just False -> return ()+          Nothing    -> PE.throw ex     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."+        mmbs <- lift (timeout wait (recv sock nbytes))+        case mmbs of+          Just (Just bs) -> P.respond (Received bs) >>= loop+          Just Nothing   -> return ()+          Nothing        -> PE.throw ex+    ex = Timeout $ "socketSyncServerTimeout: " <> show wait <> " microseconds." {-# INLINABLE socketSyncServerTimeout #-}  -- | Like 'socketSyncProxy', except it throws a 'Timeout' exception in the@@ -141,17 +146,18 @@   -> (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+        mok <- lift . timeout wait . send sock =<< P.request a'+        case mok of+          Just True  -> P.respond Sent >>= loop+          Just False -> return ()+          Nothing    -> PE.throw ex     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."+        mmbs <- lift (timeout wait (recv sock nbytes))+        case mmbs of+          Just (Just bs) -> P.respond (Received bs) >>= loop+          Just Nothing   -> return ()+          Nothing        -> PE.throw ex+    ex = Timeout $ "socketSyncProxyTimeout: " <> show wait <> " microseconds." {-# INLINABLE socketSyncProxyTimeout #-}  --------------------------------------------------------------------------------