io-streams 1.1.1.0 → 1.1.2.0
raw patch · 4 files changed
+119/−8 lines, 4 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
+ System.IO.Streams.Concurrent: makeChanPipe :: IO (InputStream a, OutputStream a)
Files
- io-streams.cabal +6/−1
- src/System/IO/Streams/Concurrent.hs +20/−7
- test/System/IO/Streams/Tests/Concurrent.hs +91/−0
- test/TestSuite.hs +2/−0
io-streams.cabal view
@@ -1,5 +1,5 @@ Name: io-streams-Version: 1.1.1.0+Version: 1.1.2.0 License: BSD3 License-file: LICENSE Category: Data, Network, IO-Streams@@ -80,6 +80,10 @@ . /ChangeLog/ .+ [@1.1.2.0@] Added @System.IO.Streams.Concurrent.makeChanPipe@, to create a+ simple concurrent pipe between an @InputStream@/@OutputStream@+ pair.+ . [@1.1.1.0@] Added @System.IO.Streams.Network.socketToStreamsWithBufferSize@, allowing control over the size of the receive buffers used when reading from sockets.@@ -188,6 +192,7 @@ System.IO.Streams.Tests.ByteString, System.IO.Streams.Tests.Combinators, System.IO.Streams.Tests.Common,+ System.IO.Streams.Tests.Concurrent, System.IO.Streams.Tests.Debug, System.IO.Streams.Tests.File, System.IO.Streams.Tests.Handle,
src/System/IO/Streams/Concurrent.hs view
@@ -8,11 +8,14 @@ , chanToInput , chanToOutput , concurrentMerge+ , makeChanPipe ) where ------------------------------------------------------------------------------+import Control.Applicative ((<$>), (<*>)) import Control.Concurrent (forkIO)-import Control.Concurrent.Chan (Chan, readChan, writeChan)+import Control.Concurrent.Chan (Chan, newChan, readChan,+ writeChan) import Control.Concurrent.MVar (modifyMVar, newEmptyMVar, newMVar, putMVar, takeMVar) import Control.Exception (SomeException, mask, throwIO,@@ -33,7 +36,7 @@ go = do mb <- read is writeChan ch mb- maybe (return ()) (const go) mb+ maybe (return $! ()) (const go) mb ------------------------------------------------------------------------------@@ -80,12 +83,22 @@ emb <- takeMVar mv case emb of Left exc -> throwIO exc- Right Nothing -> do b <- modifyMVar nleft $ \n ->+ Right Nothing -> do x <- modifyMVar nleft $ \n -> let !n' = n - 1- in return $! if n' == 0- then (n', False)- else (n', True)- if b+ in return $! (n', n')+ if x > 0 then chunk mv nleft else return Nothing Right x -> return x+++--------------------------------------------------------------------------------+-- | Create a new pair of streams using an underlying 'Chan'. Everything written+-- to the 'OutputStream' will appear as-is on the 'InputStream'.+--+-- Since reading from the 'InputStream' and writing to the 'OutputStream' are+-- blocking calls, be sure to do so in different threads.+makeChanPipe :: IO (InputStream a, OutputStream a)+makeChanPipe = do+ chan <- newChan+ (,) <$> chanToInput chan <*> chanToOutput chan
+ test/System/IO/Streams/Tests/Concurrent.hs view
@@ -0,0 +1,91 @@+module System.IO.Streams.Tests.Concurrent (tests) where++------------------------------------------------------------------------------+import Control.Concurrent+import Control.Monad+import Prelude hiding (lines, read,+ takeWhile, unlines,+ unwords, unwords,+ words)+import qualified Prelude+import qualified System.IO.Streams as Streams+import qualified System.IO.Streams.Concurrent as Streams+import System.IO.Streams.Tests.Common+import Test.Framework+import Test.Framework.Providers.HUnit+import Test.Framework.Providers.QuickCheck2+import Test.HUnit hiding (Test)+import Test.QuickCheck hiding (output)+import Test.QuickCheck.Monadic+------------------------------------------------------------------------------++tests :: [Test]+tests = [ testMakeChanPipe+ , testConcurrentMerge+ , testConcurrentMergeException+ , testInputOutput+ ]+++------------------------------------------------------------------------------+testMakeChanPipe :: Test+testMakeChanPipe = testProperty "concurrent/makeChanPipe" $+ monadicIO $+ forAllM arbitrary prop+ where+ prop :: [Int] -> PropertyM IO ()+ prop l = liftQ $ do+ (is, os) <- Streams.makeChanPipe+ _ <- forkIO $ Streams.writeList l os >> Streams.write Nothing os+ Streams.toList is >>= assertEqual "makeChanPipe" l+++------------------------------------------------------------------------------+testConcurrentMerge :: Test+testConcurrentMerge = testCase "concurrent/concurrentMerge" $ do+ mvars <- replicateM nthreads newEmptyMVar+ chans <- replicateM nthreads newChan+ let firstMVar = head mvars++ mapM_ (forkIO . ring) $ zip3 mvars (take nthreads $ drop 1 $ cycle mvars)+ chans+ inputs <- mapM Streams.chanToInput chans+ resultMVar <- newEmptyMVar+ forkIO (Streams.concurrentMerge inputs >>= Streams.toList+ >>= putMVar resultMVar)+ putMVar firstMVar 0+ result <- takeMVar resultMVar+ assertEqual "concurrent merge" [0..10] result++ where+ maxval = 10 :: Int+ nthreads = 4 :: Int++ ring (prev, next, chan) = loop+ where+ loop = do x <- takeMVar prev+ if x > maxval+ then do writeChan chan Nothing+ putMVar next x+ else do writeChan chan $ Just x+ threadDelay 100000+ putMVar next $! x + 1+ loop++------------------------------------------------------------------------------+testConcurrentMergeException :: Test+testConcurrentMergeException =+ testCase "concurrent/concurrentMerge/exception" $ do+ inp <- Streams.makeInputStream (error "bad") >>=+ Streams.concurrentMerge . (:[])+ expectExceptionH (Streams.toList inp)+++------------------------------------------------------------------------------+testInputOutput :: Test+testInputOutput = testCase "concurrent/input-output" $ do+ is <- Streams.fromList [1..10::Int]+ chan <- newChan+ is' <- Streams.chanToInput chan+ Streams.inputToChan is chan+ Streams.toList is' >>= assertEqual "input-output" [1..10]
test/TestSuite.hs view
@@ -4,6 +4,7 @@ import qualified System.IO.Streams.Tests.Builder as Builder import qualified System.IO.Streams.Tests.ByteString as ByteString import qualified System.IO.Streams.Tests.Combinators as Combinators+import qualified System.IO.Streams.Tests.Concurrent as Concurrent import qualified System.IO.Streams.Tests.Debug as Debug import qualified System.IO.Streams.Tests.File as File import qualified System.IO.Streams.Tests.Handle as Handle@@ -26,6 +27,7 @@ , testGroup "Tests.ByteString" ByteString.tests , testGroup "Tests.Debug" Debug.tests , testGroup "Tests.Combinators" Combinators.tests+ , testGroup "Tests.Concurrent" Concurrent.tests , testGroup "Tests.File" File.tests , testGroup "Tests.Handle" Handle.tests , testGroup "Tests.Internal" Internal.tests