diff --git a/changelog.md b/changelog.md --- a/changelog.md +++ b/changelog.md @@ -1,3 +1,15 @@ +# Version 1.2.0.0 + - Fixed bug #27 (https://github.com/snapframework/io-streams/issues/27): + makeOutputStream now properly shuts down the stream upon receiving EOF. The + new invariant might break user programs if they depended on the buggy + behaviour, which is the reason for the major version bump. + + - Fixed a few polymorphic bindings that started breaking in recent GHC. + + - Dependency bumps for: + - text 1.2 + - network 2.6 + # Version 1.1.4.6 Moved old changelog entries to `changelog.md`. diff --git a/io-streams.cabal b/io-streams.cabal --- a/io-streams.cabal +++ b/io-streams.cabal @@ -1,5 +1,5 @@ Name: io-streams -Version: 1.1.4.6 +Version: 1.2.0.0 License: BSD3 License-file: LICENSE Category: Data, Network, IO-Streams @@ -18,8 +18,8 @@ module "System.IO.Streams", which re-exports most of the library: . @ - import "System.IO.Streams" (InputStream, OutputStream) - import qualified "System.IO.Streams" as Streams + import System.IO.Streams (InputStream, OutputStream) + import qualified System.IO.Streams as Streams @ . For first-time users, @io-streams@ comes with an included tutorial, which can @@ -32,13 +32,13 @@ . @ \-\- read an item from an input stream - Streams.'System.IO.Streams.read' :: 'System.IO.Streams.InputStream' a -> IO (Maybe a) + Streams.read :: InputStream a -> IO (Maybe a) . \-\- push an item back to an input stream - Streams.'System.IO.Streams.unRead' :: a -> 'System.IO.Streams.InputStream' a -> IO () + Streams.unRead :: a -> InputStream a -> IO () . \-\- write to an output stream - Streams.'System.IO.Streams.write' :: Maybe a -> 'System.IO.Streams.OutputStream' a -> IO () + Streams.write :: Maybe a -> OutputStream a -> IO () @ . Streams can be transformed by composition and hooked together with provided combinators: @@ -90,7 +90,7 @@ hs-source-dirs: src Default-language: Haskell2010 - ghc-options: -O2 -Wall -fwarn-tabs -funbox-strict-fields + ghc-options: -Wall -fwarn-tabs -funbox-strict-fields -fno-warn-unused-do-bind ghc-prof-options: -prof -auto-all @@ -115,16 +115,17 @@ System.IO.Streams.Tutorial Other-modules: System.IO.Streams.Internal.Attoparsec, + System.IO.Streams.Internal.Network, System.IO.Streams.Internal.Search Build-depends: base >= 4 && <5, attoparsec >= 0.10 && <0.13, blaze-builder >= 0.3.1 && <0.4, bytestring >= 0.9 && <0.11, - network >= 2.3 && <2.6, + network >= 2.3 && <2.7, primitive >= 0.2 && <0.6, process >= 1 && <1.3, - text >= 0.10 && <1.2, + text >= 0.10 && <1.3, time >= 1.2 && <1.5, transformers >= 0.2 && <0.5, vector >= 0.7 && <0.11, @@ -186,10 +187,11 @@ System.IO.Streams.Zlib, System.IO.Streams.Internal, System.IO.Streams.Internal.Attoparsec, + System.IO.Streams.Internal.Network, System.IO.Streams.Internal.Search - ghc-options: -O2 -Wall -fhpc -fwarn-tabs -funbox-strict-fields -threaded + ghc-options: -Wall -fhpc -fwarn-tabs -funbox-strict-fields -threaded -fno-warn-unused-do-bind ghc-prof-options: -prof -auto-all @@ -205,10 +207,10 @@ directory >= 1.1 && <2, filepath >= 1.2 && <2, mtl >= 2 && <3, - network >= 2.3 && <2.6, + network >= 2.3 && <2.7, primitive >= 0.2 && <0.6, process >= 1 && <1.3, - text >= 0.10 && <1.2, + text >= 0.10 && <1.3, time >= 1.2 && <1.5, transformers >= 0.2 && <0.5, vector >= 0.7 && <0.11, diff --git a/src/System/IO/Streams/Builder.hs b/src/System/IO/Streams/Builder.hs --- a/src/System/IO/Streams/Builder.hs +++ b/src/System/IO/Streams/Builder.hs @@ -77,8 +77,8 @@ import Data.IORef (newIORef, readIORef, writeIORef) ------------------------------------------------------------------------------ import Blaze.ByteString.Builder.Internal (defaultBufferSize) -import Blaze.ByteString.Builder.Internal.Types (BufRange (..), BuildSignal (..), Builder (..), buildStep) import Blaze.ByteString.Builder.Internal.Buffer (Buffer, BufferAllocStrategy, allNewBuffersStrategy, execBuildStep, reuseBufferStrategy, unsafeFreezeBuffer, unsafeFreezeNonEmptyBuffer, updateEndOfSlice) +import Blaze.ByteString.Builder.Internal.Types (BufRange (..), BuildSignal (..), Builder (..), buildStep) import System.IO.Streams.Internal (OutputStream, makeOutputStream, write) diff --git a/src/System/IO/Streams/Combinators.hs b/src/System/IO/Streams/Combinators.hs --- a/src/System/IO/Streams/Combinators.hs +++ b/src/System/IO/Streams/Combinators.hs @@ -2,6 +2,7 @@ {-# LANGUAGE BangPatterns #-} {-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE RankNTypes #-} module System.IO.Streams.Combinators ( -- * Folds @@ -56,11 +57,11 @@ import Control.Monad (liftM, void, when) import Control.Monad.IO.Class (liftIO) import Data.Int (Int64) -import Data.IORef (atomicModifyIORef, modifyIORef, newIORef, readIORef, writeIORef) +import Data.IORef (IORef, atomicModifyIORef, modifyIORef, newIORef, readIORef, writeIORef) import Data.Maybe (isJust) import Prelude hiding (all, any, drop, filter, map, mapM, mapM_, maximum, minimum, read, take, unzip, zip, zipWith) ------------------------------------------------------------------------------ -import System.IO.Streams.Internal (InputStream (..), OutputStream, fromGenerator, makeInputStream, makeOutputStream, read, unRead, write, yield) +import System.IO.Streams.Internal (InputStream (..), OutputStream (..), fromGenerator, makeInputStream, makeOutputStream, read, unRead, write, yield) ------------------------------------------------------------------------------ @@ -606,35 +607,45 @@ -- stream. -- -- Access to the original stream is thread safe, i.e. guarded by a lock. -unzip :: InputStream (a, b) -> IO (InputStream a, InputStream b) +unzip :: forall a b . InputStream (a, b) -> IO (InputStream a, InputStream b) unzip os = do lock <- newMVar $! () buf1 <- newIORef id buf2 <- newIORef id - is1 <- makeInputStream $ src lock id buf1 buf2 - is2 <- makeInputStream $ src lock twist buf2 buf1 + is1 <- makeInputStream $ src1 lock buf1 buf2 + is2 <- makeInputStream $ src2 lock buf1 buf2 return (is1, is2) where - twist (a, b) = (b, a) + twist (a,b) = (b,a) - src lock proj myBuf theirBuf = withMVar lock $ const $ do - dl <- readIORef myBuf + src1 lock aBuf bBuf = withMVar lock $ const $ do + dl <- readIORef aBuf + case dl [] of + [] -> more os id bBuf + (x:xs) -> writeIORef aBuf (xs++) >> (return $! Just x) + src2 lock aBuf bBuf = withMVar lock $ const $ do + dl <- readIORef bBuf case dl [] of - [] -> more - (x:xs) -> writeIORef myBuf (xs++) >> (return $! Just x) - where - more = read os >>= - maybe (return Nothing) - (\x -> do - let (a, b) = proj x - modifyIORef theirBuf (. (b:)) - return $! Just a) + [] -> more os twist aBuf + (y:ys) -> writeIORef bBuf (ys++) >> (return $! Just y) + more :: forall a b x y . + InputStream (a,b) + -> ((a,b) -> (x,y)) + -> IORef ([y] -> [y]) + -> IO (Maybe x) + more origs proj buf = read origs >>= + maybe (return Nothing) + (\x -> do + let (a, b) = proj x + modifyIORef buf (. (b:)) + return $! Just a) + ------------------------------------------------------------------------------ -- | Wraps an 'InputStream', producing a new 'InputStream' that will produce at -- most @n@ items, subsequently yielding end-of-stream forever. @@ -755,7 +766,7 @@ -- /Since: 1.0.1.0/ -- ignoreEof :: OutputStream a -> IO (OutputStream a) -ignoreEof s = makeOutputStream f +ignoreEof s = return $ OutputStream f where f Nothing = return $! () f x = write x s diff --git a/src/System/IO/Streams/Internal.hs b/src/System/IO/Streams/Internal.hs --- a/src/System/IO/Streams/Internal.hs +++ b/src/System/IO/Streams/Internal.hs @@ -61,7 +61,7 @@ ) where ------------------------------------------------------------------------------ -import Control.Applicative (Applicative (..)) +import Control.Applicative (Applicative (..), (<$>)) import Control.Concurrent (newMVar, withMVar) import Control.Exception (throwIO) import Control.Monad (when) @@ -269,8 +269,21 @@ -- | Creates an 'OutputStream' from a value-consuming action. -- -- (@makeOutputStream f@) runs the computation @f@ on each value fed to it. +-- +-- Since version 1.2.0.0, 'makeOutputStream' also ensures that output streams +-- no longer receive data once EOF is received (i.e. you can now assume that +-- makeOutputStream will feed your function @Nothing@ at most once.) makeOutputStream :: (Maybe a -> IO ()) -> IO (OutputStream a) -makeOutputStream = return . OutputStream +makeOutputStream func = (OutputStream . go) <$> newIORef False + where + go closedRef !m = do + closed <- readIORef closedRef + if closed + then return $! () + else do + when (isNothing m) $ writeIORef closedRef True + func m +{-# INLINE makeOutputStream #-} ------------------------------------------------------------------------------ diff --git a/src/System/IO/Streams/Internal/Network.hs b/src/System/IO/Streams/Internal/Network.hs new file mode 100644 --- /dev/null +++ b/src/System/IO/Streams/Internal/Network.hs @@ -0,0 +1,97 @@ +{-# LANGUAGE CPP #-} + +module System.IO.Streams.Internal.Network + ( socketToStreams + , socketToStreamsWithBufferSize + , socketToStreamsWithBufferSizeImpl + ) where + + +------------------------------------------------------------------------------ +import Control.Exception (catch) +import qualified Data.ByteString.Char8 as S +import qualified Data.ByteString.Internal as S +import Data.Word (Word8) +import Foreign.ForeignPtr (newForeignPtr, withForeignPtr) +import Foreign.Marshal.Alloc (finalizerFree, mallocBytes) +import Foreign.Ptr (Ptr) +import Network.Socket (Socket) +import qualified Network.Socket as N +import qualified Network.Socket.ByteString as NB +import Prelude (IO, Int, Maybe (..), return, ($!), (<=), (>>=)) +import System.IO.Error (ioError, isEOFError) +------------------------------------------------------------------------------ +import System.IO.Streams.Internal (InputStream, OutputStream) +import qualified System.IO.Streams.Internal as Streams + + +------------------------------------------------------------------------------ +bUFSIZ :: Int +bUFSIZ = 4096 + + +------------------------------------------------------------------------------ +-- | Converts a 'Socket' to an 'InputStream' \/ 'OutputStream' pair. Note that, +-- as is usually the case in @io-streams@, writing a 'Nothing' to the generated +-- 'OutputStream' does not cause the underlying 'Socket' to be closed. +socketToStreams :: Socket + -> IO (InputStream S.ByteString, OutputStream S.ByteString) +socketToStreams = socketToStreamsWithBufferSize bUFSIZ + + +------------------------------------------------------------------------------ +-- | Converts a 'Socket' to an 'InputStream' \/ 'OutputStream' pair, with +-- control over the size of the receive buffers. Note that, as is usually the +-- case in @io-streams@, writing a 'Nothing' to the generated 'OutputStream' +-- does not cause the underlying 'Socket' to be closed. +socketToStreamsWithBufferSize + :: Int -- ^ how large the receive buffer should be + -> Socket -- ^ network socket + -> IO (InputStream S.ByteString, OutputStream S.ByteString) +#if MIN_VERSION_network(2,4,0) +socketToStreamsWithBufferSize = socketToStreamsWithBufferSizeImpl N.recvBuf +#else +socketToStreamsWithBufferSize bufsiz socket = do + is <- Streams.makeInputStream input + os <- Streams.makeOutputStream output + return $! (is, os) + + where + input = do + s <- NB.recv socket bufsiz + return $! if S.null s then Nothing else Just s + + output Nothing = return $! () + output (Just s) = if S.null s then return $! () else NB.sendAll socket s +#endif + + +------------------------------------------------------------------------------ +-- | Dependency-injected implementation of socketToStreamsWithBufferSize (for +-- testing) +socketToStreamsWithBufferSizeImpl + :: (N.Socket -> Ptr Word8 -> Int -> IO Int) -- ^ recvBuf + -> Int -- ^ how large the receive + -- buffer should be + -> Socket -- ^ network socket + -> IO (InputStream S.ByteString, OutputStream S.ByteString) +socketToStreamsWithBufferSizeImpl _recvBuf bufsiz socket = do + is <- Streams.makeInputStream input + os <- Streams.makeOutputStream output + return $! (is, os) + + where + recv buf = _recvBuf socket buf bufsiz `catch` \ioe -> + if isEOFError ioe then return 0 else ioError ioe + + mkFp = mallocBytes bufsiz >>= newForeignPtr finalizerFree + + input = do + fp <- mkFp + n <- withForeignPtr fp recv + return $! if n <= 0 + then Nothing + else Just $! S.fromForeignPtr fp 0 n + + output Nothing = return $! () + output (Just s) = if S.null s then return $! () else NB.sendAll socket s diff --git a/src/System/IO/Streams/Internal/Search.hs b/src/System/IO/Streams/Internal/Search.hs --- a/src/System/IO/Streams/Internal/Search.hs +++ b/src/System/IO/Streams/Internal/Search.hs @@ -1,5 +1,6 @@ {-# LANGUAGE BangPatterns #-} {-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RankNTypes #-} module System.IO.Streams.Internal.Search ( search @@ -9,6 +10,7 @@ ------------------------------------------------------------------------------ import Control.Monad (when) import Control.Monad.IO.Class (liftIO) +import Control.Monad.ST (ST) import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as S import qualified Data.ByteString.Unsafe as S @@ -175,6 +177,7 @@ go t where + go :: forall s . MV.MVector s Int -> ST s (MV.MVector s Int) go !t = go' 0 where go' !i | i >= lastIdx = return t diff --git a/src/System/IO/Streams/Network.hs b/src/System/IO/Streams/Network.hs --- a/src/System/IO/Streams/Network.hs +++ b/src/System/IO/Streams/Network.hs @@ -8,67 +8,5 @@ ) where ------------------------------------------------------------------------------ -import Control.Exception (catch) -import qualified Data.ByteString.Char8 as S -import qualified Data.ByteString.Internal as S -import Foreign.ForeignPtr (newForeignPtr, withForeignPtr) -import Foreign.Marshal.Alloc (finalizerFree, mallocBytes) -import Network.Socket (Socket) -import qualified Network.Socket as N -import qualified Network.Socket.ByteString as NB -import Prelude (IO, Int, Maybe (..), return, ($!), (<=), (>>=)) -import System.IO.Error (ioError, isEOFError) ------------------------------------------------------------------------------- -import System.IO.Streams.Internal (InputStream, OutputStream) -import qualified System.IO.Streams.Internal as Streams - - ------------------------------------------------------------------------------- -bUFSIZ :: Int -bUFSIZ = 4096 - - ------------------------------------------------------------------------------- --- | Converts a 'Socket' to an 'InputStream' \/ 'OutputStream' pair. Note that, --- as is usually the case in @io-streams@, writing a 'Nothing' to the generated --- 'OutputStream' does not cause the underlying 'Socket' to be closed. -socketToStreams :: Socket - -> IO (InputStream S.ByteString, OutputStream S.ByteString) -socketToStreams = socketToStreamsWithBufferSize bUFSIZ - - ------------------------------------------------------------------------------- --- | Converts a 'Socket' to an 'InputStream' \/ 'OutputStream' pair, with --- control over the size of the receive buffers. Note that, as is usually the --- case in @io-streams@, writing a 'Nothing' to the generated 'OutputStream' --- does not cause the underlying 'Socket' to be closed. -socketToStreamsWithBufferSize - :: Int -- ^ how large the receive buffer should be - -> Socket -- ^ network socket - -> IO (InputStream S.ByteString, OutputStream S.ByteString) -socketToStreamsWithBufferSize bufsiz socket = do - is <- Streams.makeInputStream input - os <- Streams.makeOutputStream output - return $! (is, os) - - where -#if MIN_VERSION_network(2,4,0) - recv buf = N.recvBuf socket buf bufsiz `catch` \ioe -> - if isEOFError ioe then return 0 else ioError ioe - - mkFp = mallocBytes bufsiz >>= newForeignPtr finalizerFree - - input = do - fp <- mkFp - n <- withForeignPtr fp recv - return $! if n <= 0 - then Nothing - else Just $! S.fromForeignPtr fp 0 n -#else - input = do - s <- NB.recv socket bufsiz - return $! if S.null s then Nothing else Just s -#endif +import System.IO.Streams.Internal.Network (socketToStreams, socketToStreamsWithBufferSize) - output Nothing = return $! () - output (Just s) = if S.null s then return $! () else NB.sendAll socket s diff --git a/test/System/IO/Streams/Tests/Builder.hs b/test/System/IO/Streams/Tests/Builder.hs --- a/test/System/IO/Streams/Tests/Builder.hs +++ b/test/System/IO/Streams/Tests/Builder.hs @@ -9,10 +9,8 @@ import qualified Data.ByteString.Char8 as S import Data.List import Data.Monoid -import System.IO.Streams hiding - (fromByteString, - intersperse, map, - take) +import System.IO.Streams hiding (fromByteString, intersperse, map, take) +import qualified System.IO.Streams as Streams import Test.Framework import Test.Framework.Providers.HUnit import Test.HUnit hiding (Test) @@ -20,6 +18,7 @@ tests :: [Test] tests = [ testBuilderStream + , testRepeatedConnects , testUnsafeBuilderStream , testSmallBuffer , testSmallBufferWithLargeOutput @@ -46,6 +45,22 @@ , "jumped over the" ] output + + +------------------------------------------------------------------------------ +testRepeatedConnects :: Test +testRepeatedConnects = testCase "builder/repeatedConnects" $ do + (os0, grab) <- Streams.listOutputStream + os <- Streams.builderStream os0 + is0 <- Streams.fromList ["Hello, world!\n"] + >>= Streams.map fromByteString + is1 <- Streams.fromList ["Bye, world!\n"] + >>= Streams.map fromByteString + Streams.connect is0 os + Streams.connect is1 os + Streams.write Nothing os + + grab >>= assertEqual "repeated connect" ["Hello, world!\n"] ------------------------------------------------------------------------------ diff --git a/test/System/IO/Streams/Tests/Handle.hs b/test/System/IO/Streams/Tests/Handle.hs --- a/test/System/IO/Streams/Tests/Handle.hs +++ b/test/System/IO/Streams/Tests/Handle.hs @@ -4,6 +4,7 @@ module System.IO.Streams.Tests.Handle (tests) where ------------------------------------------------------------------------------ +import Blaze.ByteString.Builder import Control.Exception import Control.Monad hiding (mapM) import qualified Data.ByteString.Char8 as S @@ -31,6 +32,7 @@ tests :: [Test] tests = [ testHandle , testStdHandles + , testRepeatedConnects , testInputStreamToHandle , testOutputStreamToHandle , testStreamPairToHandle @@ -59,6 +61,26 @@ Streams.toList) assertEqual "testFiles" "the quick brown fox" l + +------------------------------------------------------------------------------ +testRepeatedConnects :: Test +testRepeatedConnects = testCase "handle/repeatedConnects" $ do + createDirectoryIfMissing False dirname + tst `finally` eatException (removeFile fn >> removeDirectory dirname) + where + dirname = "tmp_r_c" + fn = dirname "data" + + tst = do + withBinaryFile fn WriteMode $ \h -> do + os0 <- Streams.handleToOutputStream h + os <- Streams.builderStream os0 + + let l1 = map fromByteString ["the ", "quick ", "brown "] + let l2 = map fromByteString ["fox ", "jumped"] + Streams.fromList l1 >>= Streams.connectTo os + Streams.fromList l2 >>= Streams.connectTo os + S.readFile fn >>= assertEqual "eof should close" "the quick brown " ------------------------------------------------------------------------------ testStdHandles :: Test diff --git a/test/System/IO/Streams/Tests/Network.hs b/test/System/IO/Streams/Tests/Network.hs --- a/test/System/IO/Streams/Tests/Network.hs +++ b/test/System/IO/Streams/Tests/Network.hs @@ -3,21 +3,27 @@ module System.IO.Streams.Tests.Network (tests) where ------------------------------------------------------------------------------ -import Control.Concurrent (forkIO, newEmptyMVar, - putMVar, takeMVar) -import qualified Network.Socket as N -import System.Timeout (timeout) +import Control.Concurrent (forkIO, newEmptyMVar, putMVar, takeMVar) +import Control.Monad (join) +import qualified Data.ByteString.Char8 as S +import Data.IORef (atomicModifyIORef, newIORef) +import qualified Network.Socket as N +import System.IO.Error (eofErrorType, mkIOError) +import System.Timeout (timeout) import Test.Framework import Test.Framework.Providers.HUnit -import Test.HUnit hiding (Test) +import Test.HUnit hiding (Test) ------------------------------------------------------------------------------ -import qualified System.IO.Streams.Internal as Streams -import qualified System.IO.Streams.List as Streams -import qualified System.IO.Streams.Network as Streams +import qualified System.IO.Streams.Internal as Streams +import qualified System.IO.Streams.Internal.Network as Streams +import qualified System.IO.Streams.List as Streams ------------------------------------------------------------------------------ +import System.IO.Streams.Tests.Common (expectExceptionH) tests :: [Test] -tests = [ testSocket ] +tests = [ testSocket + , testSocketWithError + ] testSocket :: Test testSocket = testCase "network/socket" $ @@ -59,3 +65,24 @@ Streams.toList is >>= flip Streams.writeList os N.sClose csock N.sClose sock + +testSocketWithError :: Test +testSocketWithError = testCase "network/socket-error" $ N.withSocketsDo $ do + codes1 <- newIORef [ return 1 + , ioError $ mkIOError eofErrorType "eof" Nothing Nothing ] + codes2 <- newIORef [ return 1 + , ioError $ userError "foo" ] + + (is1, _) <- Streams.socketToStreamsWithBufferSizeImpl (rbuf codes1) 64 (error "z") + (Just s1) <- Streams.read is1 + assertEqual "one byte" 1 $ S.length s1 + Nothing <- Streams.read is1 + + (is2, _) <- Streams.socketToStreamsWithBufferSizeImpl (rbuf codes2) 64 undefined + (Just s2) <- Streams.read is2 + assertEqual "one byte" 1 $ S.length s2 + expectExceptionH $ Streams.read is2 + + where + rbuf rcodes _ _ _ = join $ atomicModifyIORef rcodes $ \codes -> + (tail codes, head codes)