io-streams 1.1.4.6 → 1.2.0.0
raw patch · 11 files changed
+249/−109 lines, 11 files
Files
- changelog.md +12/−0
- io-streams.cabal +14/−12
- src/System/IO/Streams/Builder.hs +1/−1
- src/System/IO/Streams/Combinators.hs +29/−18
- src/System/IO/Streams/Internal.hs +15/−2
- src/System/IO/Streams/Internal/Network.hs +97/−0
- src/System/IO/Streams/Internal/Search.hs +3/−0
- src/System/IO/Streams/Network.hs +1/−63
- test/System/IO/Streams/Tests/Builder.hs +19/−4
- test/System/IO/Streams/Tests/Handle.hs +22/−0
- test/System/IO/Streams/Tests/Network.hs +36/−9
changelog.md view
@@ -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`.
io-streams.cabal view
@@ -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,
src/System/IO/Streams/Builder.hs view
@@ -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)
src/System/IO/Streams/Combinators.hs view
@@ -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
src/System/IO/Streams/Internal.hs view
@@ -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 #-} ------------------------------------------------------------------------------
+ src/System/IO/Streams/Internal/Network.hs view
@@ -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
src/System/IO/Streams/Internal/Search.hs view
@@ -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
src/System/IO/Streams/Network.hs view
@@ -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
test/System/IO/Streams/Tests/Builder.hs view
@@ -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"] ------------------------------------------------------------------------------
test/System/IO/Streams/Tests/Handle.hs view
@@ -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
test/System/IO/Streams/Tests/Network.hs view
@@ -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)