diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,3 +1,7 @@ +## 0.1.16 + +* Add `closeStreamingProcessHandle` + ## 0.1.15.5 * Make getSocket{Family}TCP try all addr candidates [#32](https://github.com/fpco/streaming-commons/pull/32) diff --git a/Data/Streaming/Network.hs b/Data/Streaming/Network.hs --- a/Data/Streaming/Network.hs +++ b/Data/Streaming/Network.hs @@ -182,10 +182,10 @@ theBody addr = bracketOnError (NS.socket (NS.addrFamily addr) (NS.addrSocketType addr) (NS.addrProtocol addr)) - NS.sClose + NS.close (\sock -> do mapM_ (\(opt,v) -> NS.setSocketOption sock opt v) sockOpts - NS.bindSocket sock (NS.addrAddress addr) + NS.bind sock (NS.addrAddress addr) return sock ) tryAddrs addrs' @@ -290,7 +290,7 @@ sock <- NS.socket NS.AF_UNIX NS.Stream 0 ee <- try' $ NS.connect sock (NS.SockAddrUnix path) case ee of - Left e -> NS.sClose sock >> throwIO e + Left e -> NS.close sock >> throwIO e Right () -> return sock where try' :: IO a -> IO (Either SomeException a) @@ -301,10 +301,10 @@ bindPath path = do sock <- bracketOnError (NS.socket NS.AF_UNIX NS.Stream 0) - NS.sClose + NS.close (\sock -> do removeFileSafe path -- Cannot bind if the socket file exists. - NS.bindSocket sock (NS.SockAddrUnix path) + NS.bind sock (NS.SockAddrUnix path) return sock) NS.listen sock (max 2048 NS.maxListenQueue) return sock @@ -426,7 +426,7 @@ NS.setSocketOption sock NS.NoDelay 1 return sock - connect addrInfo = E.bracketOnError (createSocket addrInfo) NS.sClose $ \sock -> do + connect addrInfo = E.bracketOnError (createSocket addrInfo) NS.close $ \sock -> do NS.connect sock (NS.addrAddress addrInfo) return (sock, NS.addrAddress addrInfo) @@ -584,20 +584,20 @@ runTCPServerWithHandle :: ServerSettings -> ConnectionHandle -> IO a runTCPServerWithHandle (ServerSettings port host msocket afterBind needLocalAddr _) handle = case msocket of - Nothing -> E.bracket (bindPortTCP port host) NS.sClose inner + Nothing -> E.bracket (bindPortTCP port host) NS.close inner Just lsocket -> inner lsocket where inner lsocket = afterBind lsocket >> forever (serve lsocket) serve lsocket = E.bracketOnError (acceptSafe lsocket) - (\(socket, _) -> NS.sClose socket) + (\(socket, _) -> NS.close socket) $ \(socket, addr) -> do mlocal <- if needLocalAddr then fmap Just $ NS.getSocketName socket else return Nothing _ <- E.mask $ \restore -> forkIO $ restore (handle socket addr mlocal) - `E.finally` NS.sClose socket + `E.finally` NS.close socket return () @@ -613,7 +613,7 @@ , appWrite' = sendAll socket , appSockAddr' = addr , appLocalAddr' = mlocal - , appCloseConnection' = NS.sClose socket + , appCloseConnection' = NS.close socket , appRawSocket' = Just socket } in @@ -623,13 +623,13 @@ runTCPClient :: ClientSettings -> (AppData -> IO a) -> IO a runTCPClient (ClientSettings port host addrFamily readBufferSize) app = E.bracket (getSocketFamilyTCP host port addrFamily) - (NS.sClose . fst) + (NS.close . fst) (\(s, address) -> app AppData { appRead' = safeRecv s readBufferSize , appWrite' = sendAll s , appSockAddr' = address , appLocalAddr' = Nothing - , appCloseConnection' = NS.sClose s + , appCloseConnection' = NS.close s , appRawSocket' = Just s }) @@ -677,14 +677,14 @@ runUnixServer :: ServerSettingsUnix -> (AppDataUnix -> IO ()) -> IO a runUnixServer (ServerSettingsUnix path afterBind readBufferSize) app = E.bracket (bindPath path) - NS.sClose + NS.close (\socket -> do afterBind socket forever $ serve socket) where serve lsocket = E.bracketOnError (acceptSafe lsocket) - (\(socket, _) -> NS.sClose socket) + (\(socket, _) -> NS.close socket) $ \(socket, _) -> do let ad = AppDataUnix { appReadUnix = safeRecv socket readBufferSize @@ -692,14 +692,14 @@ } _ <- E.mask $ \restore -> forkIO $ restore (app ad) - `E.finally` NS.sClose socket + `E.finally` NS.close socket return () -- | Run an @Application@ by connecting to the specified server. runUnixClient :: ClientSettingsUnix -> (AppDataUnix -> IO a) -> IO a runUnixClient (ClientSettingsUnix path readBufferSize) app = E.bracket (getSocketUnix path) - NS.sClose + NS.close (\sock -> app AppDataUnix { appReadUnix = safeRecv sock readBufferSize , appWriteUnix = sendAll sock diff --git a/Data/Streaming/Process.hs b/Data/Streaming/Process.hs --- a/Data/Streaming/Process.hs +++ b/Data/Streaming/Process.hs @@ -3,11 +3,12 @@ -- | A full tutorial for this module is available on FP School of Haskell: -- . -- --- Note that, while the tutorial covers @Data.Streaming.Process@, this module is --- the basis of the streaming version, and almost all concepts there apply here. +-- Note that, while the tutorial covers @Data.Conduit.Process@, that module closely +-- follows the present one, and almost all concepts in the tutorial apply here. module Data.Streaming.Process ( -- * Functions streamingProcess + , closeStreamingProcessHandle -- * Specialized streaming types , Inherited (..) , ClosedStream (..) @@ -36,7 +37,7 @@ newEmptyTMVar, putTMVar, readTMVar) import Control.Exception (Exception, throwIO, try, throw, - SomeException) + SomeException, finally) import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Maybe (fromMaybe) import Data.Streaming.Process.Internal @@ -125,14 +126,14 @@ -- -- Since 0.1.4 streamingProcessHandleRaw :: StreamingProcessHandle -> ProcessHandle -streamingProcessHandleRaw (StreamingProcessHandle ph _) = ph +streamingProcessHandleRaw (StreamingProcessHandle ph _ _) = ph -- | Get the @TMVar@ storing the process exit code. In general, one of the -- above functions should be used instead to avoid accidentally corrupting the variable\'s state.. -- -- Since 0.1.4 streamingProcessHandleTMVar :: StreamingProcessHandle -> TMVar ExitCode -streamingProcessHandleTMVar (StreamingProcessHandle _ var) = var +streamingProcessHandleTMVar (StreamingProcessHandle _ var _) = var -- | The primary function for running a process. Note that, with the -- exception of 'UseProvidedHandle', the values for @std_in@, @std_out@ @@ -169,12 +170,23 @@ (throw :: SomeException -> a) id + let close = + mclose stdinH `finally` mclose stdoutH `finally` mclose stderrH + where + mclose = maybe (return ()) hClose + (,,,) <$> getStdin stdinH <*> getStdout stdoutH <*> getStderr stderrH - <*> return (StreamingProcessHandle ph ec) + <*> return (StreamingProcessHandle ph ec close) +-- | Free any resources (e.g. @Handle@s) acquired by a call to 'streamingProcess'. +-- +-- @since 0.1.16 +closeStreamingProcessHandle :: MonadIO m => StreamingProcessHandle -> m () +closeStreamingProcessHandle (StreamingProcessHandle _ _ f) = liftIO f + -- | Indicates that a process exited with an non-success exit code. -- -- Since 0.1.7 @@ -202,9 +214,10 @@ -- code. If the exit code is not a success, throw a -- 'ProcessExitedUnsuccessfully'. -- --- NOTE: This function does not kill the child process in the event of an --- exception from the provided function. For that, please use --- @withCheckedProcessCleanup@ from the @conduit-extra@ package. +-- NOTE: This function does not kill the child process or ensure +-- resources are cleaned up in the event of an exception from the +-- provided function. For that, please use @withCheckedProcessCleanup@ +-- from the @conduit-extra@ package. -- -- Since 0.1.7 withCheckedProcess :: ( InputSource stdin @@ -218,7 +231,8 @@ withCheckedProcess cp f = do (x, y, z, sph) <- streamingProcess cp res <- f x y z - ec <- waitForStreamingProcess sph - if ec == ExitSuccess - then return res - else liftIO $ throwIO $ ProcessExitedUnsuccessfully cp ec + liftIO $ do + ec <- waitForStreamingProcess sph `finally` closeStreamingProcessHandle sph + if ec == ExitSuccess + then return res + else throwIO $ ProcessExitedUnsuccessfully cp ec diff --git a/Data/Streaming/Process/Internal.hs b/Data/Streaming/Process/Internal.hs --- a/Data/Streaming/Process/Internal.hs +++ b/Data/Streaming/Process/Internal.hs @@ -34,3 +34,4 @@ data StreamingProcessHandle = StreamingProcessHandle ProcessHandle (TMVar ExitCode) + (IO ()) -- cleanup resources diff --git a/streaming-commons.cabal b/streaming-commons.cabal --- a/streaming-commons.cabal +++ b/streaming-commons.cabal @@ -1,5 +1,5 @@ name: streaming-commons -version: 0.1.15.5 +version: 0.1.16 synopsis: Common lower-level functions needed by various streaming data libraries description: Provides low-dependency functionality commonly needed by various streaming data libraries, such as conduit and pipes. homepage: https://github.com/fpco/streaming-commons @@ -53,7 +53,7 @@ , blaze-builder >= 0.3 && < 0.5 , bytestring , directory - , network + , network >= 2.4.0.0 , random , process , stm @@ -102,7 +102,7 @@ , blaze-builder , bytestring , deepseq - , network + , network >= 2.4.0.0 , text , zlib diff --git a/test/Data/Streaming/NetworkSpec.hs b/test/Data/Streaming/NetworkSpec.hs --- a/test/Data/Streaming/NetworkSpec.hs +++ b/test/Data/Streaming/NetworkSpec.hs @@ -8,7 +8,7 @@ import qualified Data.ByteString.Char8 as S8 import Data.Char (toUpper) import Data.Streaming.Network -import Network.Socket (sClose) +import Network.Socket (close) import Test.Hspec import Test.Hspec.QuickCheck @@ -21,7 +21,7 @@ describe "bindRandomPortTCP" $ do modifyMaxSuccess (const 5) $ prop "sanity" $ \content -> bracket (bindRandomPortTCP "*4") - (sClose . snd) + (close . snd) $ \(port, socket) -> do let server ad = forever $ appRead ad >>= appWrite ad . S8.map toUpper client ad = do