unagi-streams 0.1.0.0 → 0.1.1.0

raw patch · 4 files changed

+111/−57 lines, 4 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• LICENSE +2/−3
• src/System/IO/Streams/Concurrent/Unagi.hs +23/−49
• src/System/IO/Streams/Concurrent/Unagi/Bounded.hs +79/−0
• unagi-streams.cabal +7/−5

@@ -12,9 +12,8 @@ list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. -Neither the names of Google, Erudify, nor the names of other contributors may-be used to endorse or promote products derived from this software without-specific prior written permission.+The names of contributors may be used to endorse or promote products+derived from this software without specific prior written permission.  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

@@ -2,24 +2,20 @@  module System.IO.Streams.Concurrent.Unagi        ( -- * Channel conversions-           inputToChan+         inputToChan        , chanToInput        , chanToOutput-       , concurrentMerge        , makeChanPipe+       , chanToPipe+       , dupStream        ) where   ------------------------------------------------------------------------------ import           Control.Applicative           ((<$>), (<*>))-import           Control.Concurrent            (forkIO)-import           Control.Concurrent.Chan.Unagi (InChan, OutChan, newChan,-                                                readChan, writeChan)-import           Control.Concurrent.MVar       (modifyMVar, newEmptyMVar,-                                                newMVar, putMVar, takeMVar)-import           Control.Exception             (SomeException, mask, throwIO,-                                                try)-import           Control.Monad                 (forM_)+import           Control.Concurrent.Chan.Unagi (InChan, OutChan, dupChan,+                                                newChan, readChan, writeChan)+import           Control.Monad                 ((>=>)) import           Prelude                       hiding (read) import           System.IO.Streams.Internal    (InputStream, OutputStream,                                                 makeInputStream,@@ -53,45 +49,6 @@ chanToOutput = makeOutputStream . writeChan  ---------------------------------------------------------------------------------- | Concurrently merges a list of 'InputStream's, combining values in the--- order they become available.------ Note: does /not/ forward individual end-of-stream notifications, the--- produced stream does not yield end-of-stream until all of the input streams--- have finished.------ This traps exceptions in each concurrent thread and re-raises them in the--- current thread.-concurrentMerge :: [InputStream a] -> IO (InputStream a)-concurrentMerge iss = do-    mv    <- newEmptyMVar-    nleft <- newMVar $! length iss-    mask $ \restore -> forM_ iss $ \is -> forkIO $ do-        let producer = do-              emb <- try $ restore $ read is-              case emb of-                  Left exc      -> do putMVar mv (Left (exc :: SomeException))-                                      producer-                  Right Nothing -> putMVar mv $! Right Nothing-                  Right x       -> putMVar mv (Right x) >> producer-        producer-    makeInputStream $ chunk mv nleft--  where-    chunk mv nleft = do-        emb <- takeMVar mv-        case emb of-            Left exc      -> throwIO exc-            Right Nothing -> do x <- modifyMVar nleft $ \n ->-                                     let !n' = n - 1-                                     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'.@@ -102,3 +59,20 @@ makeChanPipe = do     (inChan, outChan) <- newChan     (,) <$> chanToInput outChan <*> chanToOutput inChan+++--------------------------------------------------------------------------------+-- | Create a new pair of streams form the given '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.+chanToPipe :: (InChan (Maybe a), OutChan (Maybe a)) -> IO (InputStream a, OutputStream a)+chanToPipe (inChan, outChan) = (,) <$> chanToInput outChan <*> chanToOutput inChan+++--------------------------------------------------------------------------------+-- | Create a new input stream duplicated from the 'InChan'+--+dupStream :: InChan (Maybe a) -> IO (InputStream a)+dupStream = dupChan >=> chanToInput

@@ -0,0 +1,79 @@+{-# LANGUAGE BangPatterns #-}++module System.IO.Streams.Concurrent.Unagi.Bounded+       ( -- * Channel conversions+         inputToChan+       , chanToInput+       , chanToOutput+       , makeChanPipe+       , chanToPipe+       , dupStream+       ) where+++------------------------------------------------------------------------------+import           Control.Applicative                   ((<$>), (<*>))+import           Control.Concurrent.Chan.Unagi.Bounded (InChan, OutChan,+                                                        dupChan, newChan,+                                                        readChan, writeChan)+import           Control.Monad                         ((>=>))+import           Prelude                               hiding (read)+import           System.IO.Streams.Internal            (InputStream,+                                                        OutputStream,+                                                        makeInputStream,+                                                        makeOutputStream, read)+++------------------------------------------------------------------------------+-- | Writes the contents of an input stream to a channel until the input stream+-- yields end-of-stream.+inputToChan :: InputStream a -> InChan (Maybe a) -> IO ()+inputToChan is ch = go+  where+    go = do+        mb <- read is+        writeChan ch mb+        maybe (return $! ()) (const go) mb+++------------------------------------------------------------------------------+-- | Turns an 'OutChan' into an input stream.+--+chanToInput :: OutChan (Maybe a) -> IO (InputStream a)+chanToInput ch = makeInputStream $! readChan ch+++------------------------------------------------------------------------------+-- | Turns an 'InChan' into an output stream.+--+chanToOutput :: InChan (Maybe a) -> IO (OutputStream a)+chanToOutput = makeOutputStream . writeChan+++--------------------------------------------------------------------------------+-- | 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 :: Int -> IO (InputStream a, OutputStream a)+makeChanPipe size = do+    (inChan, outChan) <- newChan size+    (,) <$> chanToInput outChan <*> chanToOutput inChan+++--------------------------------------------------------------------------------+-- | Create a new pair of streams form the given '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.+chanToPipe :: (InChan (Maybe a), OutChan (Maybe a)) -> IO (InputStream a, OutputStream a)+chanToPipe (inChan, outChan) = (,) <$> chanToInput outChan <*> chanToOutput inChan+++--------------------------------------------------------------------------------+-- | Create a new input stream duplicated from the 'InChan'+--+dupStream :: InChan (Maybe a) -> IO (InputStream a)+dupStream = dupChan >=> chanToInput

@@ -1,9 +1,9 @@ name:                unagi-streams-version:             0.1.0.0+version:             0.1.1.0 synopsis:            Unagi Chan IO-Streams-description:         IO-Streams implemented underneath with Unagi-                     channels. This library is a straight port of Greg Collins' IO-Streams-                     Chan implementation.+description:         Io-streams chans implemented underneath with+                     unagi-chans. This library is a straight port of+                     the default io-streams chan implementation. License:             BSD3 License-file:        LICENSE author:              Luke Hoersten@@ -15,10 +15,12 @@ library   exposed-modules:                 System.IO.Streams.Concurrent.Unagi+              , System.IO.Streams.Concurrent.Unagi.Bounded   build-depends:                 base       >= 4.7 && < 4.8               , unagi-chan >= 0.2 && < 0.3               , io-streams >= 1.2 && < 1.3 -  hs-source-dirs:      src/++  hs-source-dirs:      src   default-language:    Haskell2010